【MySQL】百万级数据快速对比与批量插入实践

MySQL 数据库性能优化批量处理

数据库 MySQL

发布日期: 2025-03-03

更新日期: 2025-03-20

文章字数: 1.1k

阅读时长: 5 分

阅读次数:

问题背景

在实际业务中，我们经常会遇到需要对比和同步大量数据的场景，比如：

数据迁移和同步
系统对账
数据一致性校验
历史数据清洗

当数据量达到百万级别时，如何高效地完成数据对比和插入就成为一个重要的技术挑战。

解决方案

1. 数据预处理

在进行大规模数据对比前，首先要做好数据预处理：

// 1. 数据分组
public class DataPartitioner {
    public List<List<Data>> partition(List<Data> dataList, int batchSize) {
        return Lists.partition(dataList, batchSize);
    }
}

// 2. 数据排序
public class DataSorter {
    public void sort(List<Data> dataList) {
        dataList.sort(Comparator.comparing(Data::getId));
    }
}

2. 高效的数据对比策略

2.1 Hash对比

使用Hash对比可以快速发现数据差异：

public class HashComparator {
    public Map<String, List<Data>> compareByHash(List<Data> sourceList, List<Data> targetList) {
        // 构建源数据的Hash映射
        Map<String, Data> sourceMap = sourceList.stream()
            .collect(Collectors.toMap(this::calculateHash, data -> data));
        
        // 构建目标数据的Hash映射
        Map<String, Data> targetMap = targetList.stream()
            .collect(Collectors.toMap(this::calculateHash, data -> data));
        
        // 找出差异数据
        List<Data> insertList = new ArrayList<>();
        List<Data> updateList = new ArrayList<>();
        
        targetMap.forEach((hash, targetData) -> {
            Data sourceData = sourceMap.get(hash);
            if (sourceData == null) {
                insertList.add(targetData);
            } else if (!sourceData.equals(targetData)) {
                updateList.add(targetData);
            }
        });
        
        return Map.of("insert", insertList, "update", updateList);
    }
    
    private String calculateHash(Data data) {
        // 根据业务字段计算Hash
        return DigestUtils.md5Hex(data.toString());
    }
}

2.2 分片对比

对于超大数据量，可以采用分片对比策略：

public class ShardingComparator {
    private static final int SHARD_SIZE = 100000; // 每片10万条
    
    public void compareBySharding(List<Data> sourceList, List<Data> targetList) {
        // 1. 数据分片
        List<List<Data>> sourceShards = Lists.partition(sourceList, SHARD_SIZE);
        List<List<Data>> targetShards = Lists.partition(targetList, SHARD_SIZE);
        
        // 2. 并行对比各分片数据
        ExecutorService executor = Executors.newFixedThreadPool(4);
        CountDownLatch latch = new CountDownLatch(sourceShards.size());
        
        for (int i = 0; i < sourceShards.size(); i++) {
            final int shardIndex = i;
            executor.submit(() -> {
                try {
                    compareShardData(sourceShards.get(shardIndex), 
                                   targetShards.get(shardIndex));
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await();
        executor.shutdown();
    }
}

3. 高效的数据插入方法

3.1 批量插入

使用批量插入可以显著提升插入性能：

public class BatchInserter {
    private static final int BATCH_SIZE = 5000;
    
    @Transactional
    public void batchInsert(List<Data> dataList) {
        // 1. 数据分批
        List<List<Data>> batches = Lists.partition(dataList, BATCH_SIZE);
        
        // 2. 构建批量插入SQL
        String sql = "INSERT INTO table_name (id, name, value) VALUES (?, ?, ?)";
        
        // 3. 执行批量插入
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                Data data = dataList.get(i);
                ps.setLong(1, data.getId());
                ps.setString(2, data.getName());
                ps.setString(3, data.getValue());
            }
            
            @Override
            public int getBatchSize() {
                return dataList.size();
            }
        });
    }
}

3.2 多线程并行插入

结合线程池实现并行插入：

public class ParallelInserter {
    private static final int THREAD_COUNT = 4;
    private static final int BATCH_SIZE = 5000;
    
    public void parallelInsert(List<Data> dataList) {
        // 1. 数据分片
        List<List<Data>> shards = Lists.partition(dataList, dataList.size() / THREAD_COUNT);
        
        // 2. 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
        CountDownLatch latch = new CountDownLatch(shards.size());
        
        // 3. 并行插入
        for (List<Data> shard : shards) {
            executor.submit(() -> {
                try {
                    batchInsert(shard);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await();
        executor.shutdown();
    }
}

性能优化建议

1. 数据库优化

索引优化
- 创建合适的索引
- 避免过多索引
- 定期维护索引

配置优化

# 批量插入相关配置
innodb_buffer_pool_size=4G
innodb_flush_log_at_trx_commit=2
innodb_flush_method=O_DIRECT
innodb_log_file_size=1G

# 并发相关配置
max_connections=1000
innodb_thread_concurrency=0