批量插入

批量数据库入库，指的是一次性将多条数据通过一条 SQL 或一次数据库交互插入到数据库中，而不是逐条执行多次 INSERT 操作。

它的最大优势：

:::color1

1. 减少网络开销：单条插入需要客户端和数据库反复通信；批量插入则一次发送多条数据，减少网络往返（RTT）。
2. 提升数据库写入性能：数据库在执行一条 SQL 时会有解析、编译、日志写入等开销，批量写入能把这些成本摊薄。

:::

**eg：**假设你要插入 10 万条用户记录：

单条插入：

▼
sql
复制代码
INSERT INTO user(name, age) VALUES ('yes', 18);
INSERT INTO user(name, age) VALUES ('yupi', 19);
...

批量插入：

▼
sql
复制代码
INSERT INTO user(name, age) VALUES 
('yes', 18),
('yupi', 19),
...

Java 中的批量插入实现

1. JDBC 原生批量插入

基本实现

▼
java
复制代码
import java.sql.*;

public class JdbcBatchInsert {
    public static void main(String[] args) {
        String url = "jdbc:mysql://localhost:3306/test";
        String user = "root";
        String password = "password";
        
        Connection conn = null;
        PreparedStatement pstmt = null;
        
        try {
            // 1. 获取连接
            conn = DriverManager.getConnection(url, user, password);
            
            // 2. 关闭自动提交，开启事务
            conn.setAutoCommit(false);
            
            // 3. 创建 PreparedStatement
            String sql = "INSERT INTO users(name, email, age) VALUES (?, ?, ?)";
            pstmt = conn.prepareStatement(sql);
            
            // 4. 批量添加数据
            for (int i = 1; i <= 10000; i++) {
                pstmt.setString(1, "用户" + i);
                pstmt.setString(2, "user" + i + "@email.com");
                pstmt.setInt(3, 20 + i % 30);
                
                // 添加到批处理
                pstmt.addBatch();
                
                // 每1000条执行一次
                if (i % 1000 == 0) {
                    int[] result = pstmt.executeBatch();
                    pstmt.clearBatch();
                    System.out.println("已插入: " + i + " 条记录");
                }
            }
            
            // 5. 执行剩余的数据
            int[] remaining = pstmt.executeBatch();
            pstmt.clearBatch();
            
            // 6. 提交事务
            conn.commit();
            
            System.out.println("批量插入完成，总共插入: " + 10000 + " 条记录");
            
        } catch (SQLException e) {
            // 回滚事务
            if (conn != null) {
                try {
                    conn.rollback();
                } catch (SQLException ex) {
                    ex.printStackTrace();
                }
            }
            e.printStackTrace();
        } finally {
            // 关闭资源
            try {
                if (pstmt != null) pstmt.close();
                if (conn != null) conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}

优化版本（支持事务控制）

▼
java
复制代码
import java.sql.*;
import java.util.List;

public class AdvancedBatchInsert {
    
    /**
     * 批量插入方法
     * @param users 用户列表
     * @param batchSize 批处理大小
     * @param connection 数据库连接
     */
    public static int batchInsertUsers(List<User> users, int batchSize, 
                                      Connection connection) throws SQLException {
        
        String sql = "INSERT INTO users(name, email, age, created_at) VALUES (?, ?, ?, ?)";
        
        try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
            connection.setAutoCommit(false);
            
            int count = 0;
            int totalInserted = 0;
            
            for (User user : users) {
                pstmt.setString(1, user.getName());
                pstmt.setString(2, user.getEmail());
                pstmt.setInt(3, user.getAge());
                pstmt.setTimestamp(4, new Timestamp(user.getCreatedAt().getTime()));
                
                pstmt.addBatch();
                count++;
                
                // 达到批处理大小时执行
                if (count % batchSize == 0) {
                    int[] batchResult = pstmt.executeBatch();
                    totalInserted += batchResult.length;
                    pstmt.clearBatch();
                    
                    // 每批提交一次，避免事务过大
                    connection.commit();
                }
            }
            
            // 执行剩余数据
            if (count % batchSize != 0) {
                int[] batchResult = pstmt.executeBatch();
                totalInserted += batchResult.length;
                connection.commit();
            }
            
            connection.setAutoCommit(true);
            return totalInserted;
            
        } catch (SQLException e) {
            connection.rollback();
            throw e;
        }
    }
    
    static class User {
        private String name;
        private String email;
        private int age;
        private Date createdAt;
        
        // 构造方法、getter、setter
    }
}

2. 使用 Spring JdbcTemplate 批量插入

Maven 依赖

▼
xml
复制代码
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.33</version>
</dependency>

实现代码

▼
java
复制代码
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

@Repository
public class UserRepository {
    
    private final JdbcTemplate jdbcTemplate;
    
    public UserRepository(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }
    
    /**
     * 方法1：使用 BatchPreparedStatementSetter
     */
    @Transactional
    public int[] batchInsert(List<User> users) {
        String sql = "INSERT INTO users(name, email, age) VALUES (?, ?, ?)";
        
        return jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                User user = users.get(i);
                ps.setString(1, user.getName());
                ps.setString(2, user.getEmail());
                ps.setInt(3, user.getAge());
            }
            
            @Override
            public int getBatchSize() {
                return users.size();
            }
        });
    }
    
    /**
     * 方法2：分批插入（避免内存溢出）
     */
    @Transactional
    public void batchInsertInChunks(List<User> users, int chunkSize) {
        String sql = "INSERT INTO users(name, email, age) VALUES (?, ?, ?)";
        
        for (int i = 0; i < users.size(); i += chunkSize) {
            int end = Math.min(users.size(), i + chunkSize);
            List<User> subList = users.subList(i, end);
            
            jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
                @Override
                public void setValues(PreparedStatement ps, int index) throws SQLException {
                    User user = subList.get(index);
                    ps.setString(1, user.getName());
                    ps.setString(2, user.getEmail());
                    ps.setInt(3, user.getAge());
                }
                
                @Override
                public int getBatchSize() {
                    return subList.size();
                }
            });
            
            System.out.println("已插入 " + end + " 条记录");
        }
    }
}

3. 使用 Spring Data JPA 批量插入

实体类

▼
java
复制代码
@Entity
@Table(name = "users")
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    private String name;
    private String email;
    private Integer age;
    
    // 构造方法、getter、setter
}

Repository 实现

▼
java
复制代码
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface UserJpaRepository extends JpaRepository<User, Long> {
}

批量插入服务

▼
java
复制代码
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import java.util.List;

@Service
public class UserBatchService {
    
    @PersistenceContext
    private EntityManager entityManager;
    
    private final UserJpaRepository userRepository;
    
    public UserBatchService(UserJpaRepository userRepository) {
        this.userRepository = userRepository;
    }
    
    /**
     * 方法1：使用 saveAll（性能一般）
     */
    @Transactional
    public List<User> batchSave(List<User> users) {
        return userRepository.saveAll(users);
    }
    
    /**
     * 方法2：使用 EntityManager 手动批量插入（高性能）
     */
    @Transactional
    public void batchInsertWithEntityManager(List<User> users, int batchSize) {
        for (int i = 0; i < users.size(); i++) {
            entityManager.persist(users.get(i));
            
            // 每 batchSize 条刷新并清除缓存
            if (i > 0 && i % batchSize == 0) {
                entityManager.flush();
                entityManager.clear();
            }
        }
        
        // 处理剩余数据
        entityManager.flush();
        entityManager.clear();
    }
    
    /**
     * 方法3：使用 JDBC 风格批量插入（最高性能）
     */
    @Transactional
    public void batchInsertWithJdbcStyle(List<User> users, int batchSize) {
        Session session = entityManager.unwrap(Session.class);
        
        for (int i = 0; i < users.size(); i++) {
            session.save(users.get(i));
            
            if (i > 0 && i % batchSize == 0) {
                session.flush();
                session.clear();
            }
        }
        
        session.flush();
        session.clear();
    }
}

4. 使用 MyBatis 批量插入

Mapper 接口

▼
java
复制代码
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Param;
import java.util.List;

public interface UserMapper {
    
    // 方法1：使用 foreach 标签
    @Insert({
        "<script>",
        "INSERT INTO users(name, email, age) VALUES ",
        "<foreach collection='users' item='user' separator=','>",
        "(#{user.name}, #{user.email}, #{user.age})",
        "</foreach>",
        "</script>"
    })
    int batchInsert(@Param("users") List<User> users);
    
    // 方法2：使用 ExecutorType.BATCH
    void insertUser(User user);
}

批量插入服务

▼
java
复制代码
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class MyBatisBatchService {
    
    private final SqlSessionFactory sqlSessionFactory;
    
    public MyBatisBatchService(SqlSessionFactory sqlSessionFactory) {
        this.sqlSessionFactory = sqlSessionFactory;
    }
    
    /**
     * 使用 Batch Executor（推荐）
     */
    public void batchInsert(List<User> users) {
        try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            UserMapper mapper = sqlSession.getMapper(UserMapper.class);
            
            for (int i = 0; i < users.size(); i++) {
                mapper.insertUser(users.get(i));
                
                // 每1000条提交一次
                if (i > 0 && i % 1000 == 0) {
                    sqlSession.commit();
                }
            }
            
            sqlSession.commit();  // 提交剩余数据
        }
    }
    
    /**
     * 配置示例（mybatis-config.xml）
     */
    /*
    <settings>
        <setting name="defaultExecutorType" value="BATCH"/>
        <setting name="jdbcBatchSize" value="1000"/>
    </settings>
    */
}

5. 性能优化建议

1. 合适的批处理大小

▼
java
复制代码
// 不同数据库的建议批处理大小
public class BatchSizeConfig {
    // MySQL: 100-1000条
    public static final int MYSQL_BATCH_SIZE = 500;
    
    // PostgreSQL: 100-1000条
    public static final int POSTGRESQL_BATCH_SIZE = 500;
    
    // SQL Server: 100-1000条
    public static final int SQLSERVER_BATCH_SIZE = 1000;
    
    // Oracle: 50-100条（Oracle对批处理有限制）
    public static final int ORACLE_BATCH_SIZE = 50;
}

2. 连接池配置优化

▼
yaml
复制代码
# application.yml
spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 10
      connection-timeout: 30000
      max-lifetime: 1800000
      idle-timeout: 600000
      
    # 批处理相关优化
    hikari:
      data-source-properties:
        rewriteBatchedStatements: true  # MySQL 优化
        useServerPrepStmts: true
        cachePrepStmts: true
        prepStmtCacheSize: 250
        prepStmtCacheSqlLimit: 2048

3. 完整的批量插入工具类

▼
java
复制代码
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.function.BiConsumer;

public class BatchInsertUtils {
    
    /**
     * 通用批量插入工具方法
     * @param connection 数据库连接
     * @param sql SQL语句
     * @param dataList 数据列表
     * @param batchSize 批处理大小
     * @param parameterSetter 参数设置器
     * @return 插入的总行数
     */
    public static <T> int batchInsert(Connection connection,
                                     String sql,
                                     List<T> dataList,
                                     int batchSize,
                                     BiConsumer<PreparedStatement, T> parameterSetter) 
            throws SQLException {
        
        if (dataList == null || dataList.isEmpty()) {
            return 0;
        }
        
        boolean originalAutoCommit = connection.getAutoCommit();
        int totalInserted = 0;
        
        try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
            connection.setAutoCommit(false);
            
            int count = 0;
            for (T data : dataList) {
                parameterSetter.accept(pstmt, data);
                pstmt.addBatch();
                count++;
                
                if (count % batchSize == 0) {
                    int[] results = pstmt.executeBatch();
                    totalInserted += results.length;
                    pstmt.clearBatch();
                    
                    // 分批提交，避免事务过大
                    connection.commit();
                }
            }
            
            // 执行剩余数据
            if (count % batchSize != 0) {
                int[] results = pstmt.executeBatch();
                totalInserted += results.length;
                connection.commit();
            }
            
            return totalInserted;
            
        } catch (SQLException e) {
            connection.rollback();
            throw e;
        } finally {
            connection.setAutoCommit(originalAutoCommit);
        }
    }
    
    /**
     * 使用示例
     */
    public static void example() throws SQLException {
        List<User> users = getUsers(); // 获取用户列表
        
        int inserted = batchInsert(
            dataSource.getConnection(),
            "INSERT INTO users(name, email, age) VALUES (?, ?, ?)",
            users,
            500,  // 批处理大小
            (pstmt, user) -> {
                pstmt.setString(1, user.getName());
                pstmt.setString(2, user.getEmail());
                pstmt.setInt(3, user.getAge());
            }
        );
        
        System.out.println("成功插入 " + inserted + " 条记录");
    }
}

6. 注意事项和最佳实践

注意事项：

1. 内存管理：大数据量时要分批次处理
2. 事务控制：适当的事务大小，避免锁表时间过长
3. 错误处理：部分失败时要有回滚机制
4. 连接释放：确保连接正确关闭

最佳实践：

1. 测试最佳批处理大小：不同数据库、不同数据量最佳大小不同
2. 监控性能：使用连接池监控工具
3. 异步处理：大数据量考虑异步批量插入
4. 使用连接池：使用 HikariCP、Druid 等高性能连接池

常见问题解决：

▼
java
复制代码
// 1. MySQL 批处理优化（需要在连接字符串中添加参数）
String url = "jdbc:mysql://localhost:3306/test?" +
             "rewriteBatchedStatements=true&" +  // 关键参数
             "useServerPrepStmts=true&" +
             "cachePrepStmts=true";

// 2. 处理批处理中的部分失败
public static int safeBatchInsert(...) {
    try {
        return batchInsert(...);
    } catch (BatchUpdateException e) {
        // 获取成功执行的条数
        int[] updateCounts = e.getUpdateCounts();
        int successCount = 0;
        for (int count : updateCounts) {
            if (count >= 0) successCount++;
        }
        return successCount;
    }
}

总结

:::color2 在 Java 中实现批量插入的主要方法：

:::

:::color2 推荐方案：

• 小数据量（<1000）：Spring Data JPA 的 saveAll()
• 中等数据量（1000-100000）：Spring JdbcTemplate
• 大数据量（>100000）：JDBC 原生批处理或 MyBatis Batch Executor

关键优化点：

1. 设置合适的批处理大小（通常 500-1000）
2. 关闭自动提交，手动控制事务
3. 使用连接池并配置优化参数
4. 分批次处理避免内存溢出
5. 异常处理和回滚机制

:::