TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("sentences", new RandomSentenceSpout(), 5);
builder.setBolt("split", new SplitSentence(), 8)
    .shuffleGrouping("sentences");
builder.setBolt("count", new WordCount(), 12)
    .fieldsGrouping("split", new Fields("word"));

DataStream stream = env.socketTextStream("localhost", 9999);
stream.flatMap(new Splitter())
    .keyBy(0)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .sum(1);

// DatabaseConfig.java
public class DatabaseConfig {
    public static final String URL = "jdbc:mysql://localhost:3306/demo_db?useSSL=false&serverTimezone=UTC";
    public static final String USERNAME = "root";
    public static final String PASSWORD = "password";
    public static final String DRIVER = "com.mysql.cj.jdbc.Driver";
}

// JDBCBasicExample.java
import java.sql.*;

public class JDBCBasicExample {

    public static void main(String[] args) {
        // 加载驱动
        try {
            Class.forName(DatabaseConfig.DRIVER);
        } catch (ClassNotFoundException e) {
            System.out.println("驱动加载失败: " + e.getMessage());
            return;
        }

        // 建立连接并执行查询
        try (Connection conn = DriverManager.getConnection(
                DatabaseConfig.URL,
                DatabaseConfig.USERNAME,
                DatabaseConfig.PASSWORD)) {

            System.out.println("数据库连接成功！");

            // 执行查询
            String sql = "SELECT * FROM users";
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);

            // 处理结果集
            System.out.println("用户列表：");
            while (rs.next()) {
                int id = rs.getInt("id");
                String username = rs.getString("username");
                String email = rs.getString("email");
                int age = rs.getInt("age");
                Timestamp createdAt = rs.getTimestamp("created_at");

                System.out.printf("ID: %d, 用户名: %s, 邮箱: %s, 年龄: %d, 创建时间: %s%n",
                    id, username, email, age, createdAt);
            }

        } catch (SQLException e) {
            System.out.println("数据库操作异常: " + e.getMessage());
        }
    }
}

// User.java
import java.sql.Timestamp;

public class User {
    private int id;
    private String username;
    private String email;
    private int age;
    private Timestamp createdAt;

    // 构造函数
    public User() {}

    public User(String username, String email, int age) {
        this.username = username;
        this.email = email;
        this.age = age;
    }

    // Getter和Setter方法
    public int getId() { return id; }
    public void setId(int id) { this.id = id; }

    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }

    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }

    public int getAge() { return age; }
    public void setAge(int age) { this.age = age; }

    public Timestamp getCreatedAt() { return createdAt; }
    public void setCreatedAt(Timestamp createdAt) { this.createdAt = createdAt; }

    @Override
    public String toString() {
        return String.format("User{id=%d, username='%s', email='%s', age=%d, createdAt=%s}",
            id, username, email, age, createdAt);
    }
}

// UserDAO.java - 数据访问对象
import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class UserDAO {

    /**
     * 创建用户
     */
    public boolean createUser(User user) {
        String sql = "INSERT INTO users (username, email, age) VALUES (?, ?, ?)";

        try (Connection conn = getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS)) {

            pstmt.setString(1, user.getUsername());
            pstmt.setString(2, user.getEmail());
            pstmt.setInt(3, user.getAge());

            int affectedRows = pstmt.executeUpdate();

            if (affectedRows > 0) {
                // 获取生成的主键
                ResultSet generatedKeys = pstmt.getGeneratedKeys();
                if (generatedKeys.next()) {
                    user.setId(generatedKeys.getInt(1));
                }
                return true;
            }

        } catch (SQLException e) {
            System.out.println("创建用户失败: " + e.getMessage());
        }
        return false;
    }

    /**
     * 根据ID查询用户
     */
    public User getUserById(int id) {
        String sql = "SELECT * FROM users WHERE id = ?";

        try (Connection conn = getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setInt(1, id);
            ResultSet rs = pstmt.executeQuery();

            if (rs.next()) {
                return mapResultSetToUser(rs);
            }

        } catch (SQLException e) {
            System.out.println("查询用户失败: " + e.getMessage());
        }
        return null;
    }

    /**
     * 查询所有用户
     */
    public List<User> getAllUsers() {
        List<User> users = new ArrayList<>();
        String sql = "SELECT * FROM users ORDER BY created_at DESC";

        try (Connection conn = getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(sql)) {

            while (rs.next()) {
                users.add(mapResultSetToUser(rs));
            }

        } catch (SQLException e) {
            System.out.println("查询用户列表失败: " + e.getMessage());
        }
        return users;
    }

    /**
     * 更新用户信息
     */
    public boolean updateUser(User user) {
        String sql = "UPDATE users SET username = ?, email = ?, age = ? WHERE id = ?";

        try (Connection conn = getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setString(1, user.getUsername());
            pstmt.setString(2, user.getEmail());
            pstmt.setInt(3, user.getAge());
            pstmt.setInt(4, user.getId());

            return pstmt.executeUpdate() > 0;

        } catch (SQLException e) {
            System.out.println("更新用户失败: " + e.getMessage());
        }
        return false;
    }

    /**
     * 删除用户
     */
    public boolean deleteUser(int id) {
        String sql = "DELETE FROM users WHERE id = ?";

        try (Connection conn = getConnection();
             PreparedStatement pstmt = conn.prepareStatement(sql)) {

            pstmt.setInt(1, id);
            return pstmt.executeUpdate() > 0;

        } catch (SQLException e) {
            System.out.println("删除用户失败: " + e.getMessage());
        }
        return false;
    }

    /**
     * 获取数据库连接
     */
    private Connection getConnection() throws SQLException {
        return DriverManager.getConnection(
            DatabaseConfig.URL,
            DatabaseConfig.USERNAME,
            DatabaseConfig.PASSWORD
        );
    }

    /**
     * 将ResultSet映射为User对象
     */
    private User mapResultSetToUser(ResultSet rs) throws SQLException {
        User user = new User();
        user.setId(rs.getInt("id"));
        user.setUsername(rs.getString("username"));
        user.setEmail(rs.getString("email"));
        user.setAge(rs.getInt("age"));
        user.setCreatedAt(rs.getTimestamp("created_at"));
        return user;
    }
}

// JDBCTest.java
public class JDBCTest {
    public static void main(String[] args) {
        UserDAO userDAO = new UserDAO();

        // 1. 创建用户
        System.out.println("=== 创建用户 ===");
        User newUser = new User("赵六", "zhaoliu@example.com", 26);
        if (userDAO.createUser(newUser)) {
            System.out.println("用户创建成功，ID: " + newUser.getId());
        }

        // 2. 查询单个用户
        System.out.println("\n=== 查询用户 ===");
        User user = userDAO.getUserById(newUser.getId());
        if (user != null) {
            System.out.println("查询到用户: " + user);
        }

        // 3. 查询所有用户
        System.out.println("\n=== 所有用户列表 ===");
        List<User> users = userDAO.getAllUsers();
        users.forEach(System.out::println);

        // 4. 更新用户
        System.out.println("\n=== 更新用户 ===");
        user.setAge(27);
        user.setEmail("zhaoliu_new@example.com");
        if (userDAO.updateUser(user)) {
            System.out.println("用户信息更新成功");
            System.out.println("更新后: " + userDAO.getUserById(user.getId()));
        }

        // 5. 删除用户
        System.out.println("\n=== 删除用户 ===");
        if (userDAO.deleteUser(user.getId())) {
            System.out.println("用户删除成功");
        }

        // 再次查询验证删除
        System.out.println("删除后查询结果: " + userDAO.getUserById(user.getId()));
    }
}

// ConnectionPool.java
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import javax.sql.DataSource;

public class ConnectionPool {
    private static HikariDataSource dataSource;

    static {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl(DatabaseConfig.URL);
        config.setUsername(DatabaseConfig.USERNAME);
        config.setPassword(DatabaseConfig.PASSWORD);
        config.setDriverClassName(DatabaseConfig.DRIVER);

        // 连接池配置
        config.setMaximumPoolSize(20);
        config.setMinimumIdle(5);
        config.setConnectionTimeout(30000);
        config.setIdleTimeout(600000);
        config.setMaxLifetime(1800000);

        dataSource = new HikariDataSource(config);
    }

    public static DataSource getDataSource() {
        return dataSource;
    }

    public static void close() {
        if (dataSource != null) {
            dataSource.close();
        }
    }
}

// TransactionExample.java
import java.sql.Connection;
import java.sql.SQLException;

public class TransactionExample {

    /**
     * 转账操作 - 事务示例
     */
    public boolean transferMoney(int fromUserId, int toUserId, double amount) {
        Connection conn = null;
        try {
            conn = ConnectionPool.getDataSource().getConnection();
            conn.setAutoCommit(false); // 开启事务

            // 1. 检查余额
            if (!checkBalance(conn, fromUserId, amount)) {
                throw new SQLException("余额不足");
            }

            // 2. 扣除转出账户金额
            updateBalance(conn, fromUserId, -amount);

            // 3. 增加转入账户金额
            updateBalance(conn, toUserId, amount);

            conn.commit(); // 提交事务
            System.out.println("转账成功");
            return true;

        } catch (SQLException e) {
            try {
                if (conn != null) {
                    conn.rollback(); // 回滚事务
                    System.out.println("转账失败，已回滚: " + e.getMessage());
                }
            } catch (SQLException ex) {
                System.out.println("回滚失败: " + ex.getMessage());
            }
            return false;
        } finally {
            try {
                if (conn != null) {
                    conn.setAutoCommit(true); // 恢复自动提交
                    conn.close();
                }
            } catch (SQLException e) {
                System.out.println("关闭连接失败: " + e.getMessage());
            }
        }
    }

    private boolean checkBalance(Connection conn, int userId, double amount) throws SQLException {
        // 检查余额的实现
        return true; // 简化示例
    }

    private void updateBalance(Connection conn, int userId, double amount) throws SQLException {
        // 更新余额的实现
    }
}

// User.java (增强版)
import java.sql.Timestamp;
import java.util.List;

public class User {
    private Integer id;
    private String username;
    private String email;
    private Integer age;
    private Timestamp createdAt;
    private List<Order> orders; // 一对多关系

    // 构造函数
    public User() {}

    public User(String username, String email, Integer age) {
        this.username = username;
        this.email = email;
        this.age = age;
    }

    // 所有Getter和Setter方法
    public Integer getId() { return id; }
    public void setId(Integer id) { this.id = id; }

    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }

    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }

    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }

    public Timestamp getCreatedAt() { return createdAt; }
    public void setCreatedAt(Timestamp createdAt) { this.createdAt = createdAt; }

    public List<Order> getOrders() { return orders; }
    public void setOrders(List<Order> orders) { this.orders = orders; }

    @Override
    public String toString() {
        return String.format("User{id=%d, username='%s', email='%s', age=%d, createdAt=%s, orders=%d}",
            id, username, email, age, createdAt, orders != null ? orders.size() : 0);
    }
}

// Order.java - 订单实体类
import java.math.BigDecimal;
import java.sql.Timestamp;

public class Order {
    private Integer id;
    private Integer userId;
    private String orderNo;
    private BigDecimal amount;
    private String status;
    private Timestamp createdAt;
    private User user; // 多对一关系

    // 构造函数
    public Order() {}

    public Order(Integer userId, String orderNo, BigDecimal amount, String status) {
        this.userId = userId;
        this.orderNo = orderNo;
        this.amount = amount;
        this.status = status;
    }

    // 所有Getter和Setter方法
    public Integer getId() { return id; }
    public void setId(Integer id) { this.id = id; }

    public Integer getUserId() { return userId; }
    public void setUserId(Integer userId) { this.userId = userId; }

    public String getOrderNo() { return orderNo; }
    public void setOrderNo(String orderNo) { this.orderNo = orderNo; }

    public BigDecimal getAmount() { return amount; }
    public void setAmount(BigDecimal amount) { this.amount = amount; }

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public Timestamp getCreatedAt() { return createdAt; }
    public void setCreatedAt(Timestamp createdAt) { this.createdAt = createdAt; }

    public User getUser() { return user; }
    public void setUser(User user) { this.user = user; }

    @Override
    public String toString() {
        return String.format("Order{id=%d, userId=%d, orderNo='%s', amount=%s, status='%s', createdAt=%s}",
            id, userId, orderNo, amount, status, createdAt);
    }
}

// UserMapper.java
import java.util.List;
import java.util.Map;
import org.apache.ibatis.annotations.*;

public interface UserMapper {

    // 基础CRUD操作
    int insertUser(User user);
    User selectUserById(Integer id);
    List<User> selectAllUsers();
    int updateUser(User user);
    int deleteUser(Integer id);

    // 条件查询
    List<User> selectUsersByAge(@Param("minAge") Integer minAge, @Param("maxAge") Integer maxAge);
    List<User> selectUsersByCondition(Map<String, Object> params);

    // 关联查询
    User selectUserWithOrders(Integer id);
    List<User> selectUsersWithOrders();

    // 分页查询
    List<User> selectUsersByPage(@Param("offset") Integer offset, @Param("limit") Integer limit);

    // 统计查询
    int countUsers();
    int countUsersByStatus(String status);

    // 批量操作
    int batchInsertUsers(@Param("users") List<User> users);
    int batchUpdateUsers(@Param("users") List<User> users);
}

// MyBatisUtil.java
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;

import java.io.IOException;
import java.io.InputStream;

public class MyBatisUtil {
    private static SqlSessionFactory sqlSessionFactory;

    static {
        try {
            String resource = "mybatis-config.xml";
            InputStream inputStream = Resources.getResourceAsStream(resource);
            sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
        } catch (IOException e) {
            throw new RuntimeException("初始化MyBatis失败", e);
        }
    }

    /**
     * 获取SqlSession
     */
    public static SqlSession getSqlSession() {
        return sqlSessionFactory.openSession();
    }

    /**
     * 获取SqlSession（自动提交）
     */
    public static SqlSession getSqlSession(boolean autoCommit) {
        return sqlSessionFactory.openSession(autoCommit);
    }

    /**
     * 获取Mapper
     */
    public static <T> T getMapper(Class<T> mapperClass) {
        SqlSession session = getSqlSession();
        return session.getMapper(mapperClass);
    }

    /**
     * 关闭SqlSession
     */
    public static void closeSqlSession(SqlSession session) {
        if (session != null) {
            session.close();
        }
    }
}

// UserService.java
import org.apache.ibatis.session.SqlSession;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class UserService {

    /**
     * 创建用户
     */
    public boolean createUser(User user) {
        SqlSession session = MyBatisUtil.getSqlSession();
        try {
            UserMapper mapper = session.getMapper(UserMapper.class);
            int result = mapper.insertUser(user);
            session.commit();
            return result > 0;
        } catch (Exception e) {
            session.rollback();
            System.out.println("创建用户失败: " + e.getMessage());
            return false;
        } finally {
            MyBatisUtil.closeSqlSession(session);
        }
    }

    /**
     * 根据ID查询用户
     */
    public User getUserById(Integer id) {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);
            return mapper.selectUserById(id);
        } catch (Exception e) {
            System.out.println("查询用户失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 查询所有用户
     */
    public List<User> getAllUsers() {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);
            return mapper.selectAllUsers();
        } catch (Exception e) {
            System.out.println("查询用户列表失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 更新用户
     */
    public boolean updateUser(User user) {
        SqlSession session = MyBatisUtil.getSqlSession();
        try {
            UserMapper mapper = session.getMapper(UserMapper.class);
            int result = mapper.updateUser(user);
            session.commit();
            return result > 0;
        } catch (Exception e) {
            session.rollback();
            System.out.println("更新用户失败: " + e.getMessage());
            return false;
        } finally {
            MyBatisUtil.closeSqlSession(session);
        }
    }

    /**
     * 删除用户
     */
    public boolean deleteUser(Integer id) {
        SqlSession session = MyBatisUtil.getSqlSession();
        try {
            UserMapper mapper = session.getMapper(UserMapper.class);
            int result = mapper.deleteUser(id);
            session.commit();
            return result > 0;
        } catch (Exception e) {
            session.rollback();
            System.out.println("删除用户失败: " + e.getMessage());
            return false;
        } finally {
            MyBatisUtil.closeSqlSession(session);
        }
    }

    /**
     * 按年龄范围查询用户
     */
    public List<User> getUsersByAgeRange(Integer minAge, Integer maxAge) {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);
            return mapper.selectUsersByAge(minAge, maxAge);
        } catch (Exception e) {
            System.out.println("按年龄查询用户失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 动态条件查询
     */
    public List<User> getUsersByCondition(String username, String email, Integer minAge, Integer maxAge) {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);

            Map<String, Object> params = new HashMap<>();
            params.put("username", username);
            params.put("email", email);
            params.put("minAge", minAge);
            params.put("maxAge", maxAge);

            return mapper.selectUsersByCondition(params);
        } catch (Exception e) {
            System.out.println("条件查询用户失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 查询用户及其订单
     */
    public User getUserWithOrders(Integer id) {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);
            return mapper.selectUserWithOrders(id);
        } catch (Exception e) {
            System.out.println("查询用户订单失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 分页查询用户
     */
    public List<User> getUsersByPage(int page, int size) {
        try (SqlSession session = MyBatisUtil.getSqlSession()) {
            UserMapper mapper = session.getMapper(UserMapper.class);
            int offset = (page - 1) * size;
            return mapper.selectUsersByPage(offset, size);
        } catch (Exception e) {
            System.out.println("分页查询用户失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 批量创建用户
     */
    public boolean batchCreateUsers(List<User> users) {
        SqlSession session = MyBatisUtil.getSqlSession();
        try {
            UserMapper mapper = session.getMapper(UserMapper.class);
            int result = mapper.batchInsertUsers(users);
            session.commit();
            return result > 0;
        } catch (Exception e) {
            session.rollback();
            System.out.println("批量创建用户失败: " + e.getMessage());
            return false;
        } finally {
            MyBatisUtil.closeSqlSession(session);
        }
    }
}

// MyBatisTest.java
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;

public class MyBatisTest {
    public static void main(String[] args) {
        UserService userService = new UserService();

        // 1. 创建用户测试
        System.out.println("=== 创建用户测试 ===");
        User newUser = new User("陈七", "chenqi@example.com", 24);
        if (userService.createUser(newUser)) {
            System.out.println("用户创建成功，ID: " + newUser.getId());
        }

        // 2. 查询用户测试
        System.out.println("\n=== 查询用户测试 ===");
        User user = userService.getUserById(newUser.getId());
        if (user != null) {
            System.out.println("查询到用户: " + user);
        }

        // 3. 查询所有用户
        System.out.println("\n=== 所有用户列表 ===");
        List<User> users = userService.getAllUsers();
        if (users != null) {
            users.forEach(System.out::println);
        }

        // 4. 按年龄范围查询
        System.out.println("\n=== 年龄范围查询 (25-30岁) ===");
        List<User> usersByAge = userService.getUsersByAgeRange(25, 30);
        if (usersByAge != null) {
            usersByAge.forEach(System.out::println);
        }

        // 5. 动态条件查询
        System.out.println("\n=== 动态条件查询 ===");
        List<User> usersByCondition = userService.getUsersByCondition("张", null, 20, 35);
        if (usersByCondition != null) {
            usersByCondition.forEach(System.out::println);
        }

        // 6. 查询用户及订单
        System.out.println("\n=== 查询用户及订单 ===");
        User userWithOrders = userService.getUserWithOrders(1);
        if (userWithOrders != null) {
            System.out.println("用户信息: " + userWithOrders);
            if (userWithOrders.getOrders() != null) {
                System.out.println("订单列表:");
                userWithOrders.getOrders().forEach(order ->
                    System.out.println("  " + order));
            }
        }

        // 7. 分页查询
        System.out.println("\n=== 分页查询 (第1页，每页2条) ===");
        List<User> pageUsers = userService.getUsersByPage(1, 2);
        if (pageUsers != null) {
            pageUsers.forEach(System.out::println);
        }

        // 8. 批量创建用户
        System.out.println("\n=== 批量创建用户 ===");
        List<User> batchUsers = Arrays.asList(
            new User("批量用户1", "batch1@example.com", 22),
            new User("批量用户2", "batch2@example.com", 23),
            new User("批量用户3", "batch3@example.com", 24)
        );
        if (userService.batchCreateUsers(batchUsers)) {
            System.out.println("批量创建用户成功");
            batchUsers.forEach(u -> System.out.println("新用户ID: " + u.getId()));
        }

        // 9. 更新用户
        System.out.println("\n=== 更新用户 ===");
        user.setAge(25);
        user.setEmail("chenqi_updated@example.com");
        if (userService.updateUser(user)) {
            System.out.println("用户更新成功");
            System.out.println("更新后: " + userService.getUserById(user.getId()));
        }

        // 10. 删除用户
        System.out.println("\n=== 删除用户 ===");
        if (userService.deleteUser(user.getId())) {
            System.out.println("用户删除成功");
        }
    }
}

// UserAnnotationMapper.java - 注解方式
import org.apache.ibatis.annotations.*;
import java.util.List;

public interface UserAnnotationMapper {

    @Select("SELECT * FROM users WHERE id = #{id}")
    @Results({
        @Result(property = "id", column = "id"),
        @Result(property = "username", column = "username"),
        @Result(property = "email", column = "email"),
        @Result(property = "age", column = "age"),
        @Result(property = "createdAt", column = "created_at")
    })
    User selectUserById(Integer id);

    @Insert("INSERT INTO users(username, email, age) VALUES(#{username}, #{email}, #{age})")
    @Options(useGeneratedKeys = true, keyProperty = "id")
    int insertUser(User user);

    @Update("UPDATE users SET username=#{username}, email=#{email}, age=#{age} WHERE id=#{id}")
    int updateUser(User user);

    @Delete("DELETE FROM users WHERE id = #{id}")
    int deleteUser(Integer id);

    // 动态SQL注解
    @SelectProvider(type = UserSqlProvider.class, method = "selectUsersByCondition")
    List<User> selectUsersByCondition(Map<String, Object> params);
}