Debezium Kafka 全量同步，处理大量写入问题

        Debezium 是一个通用的数据实时同步库，使用它我们可以解决大数据量的数据全量同步和增量同步，它可以完成全量同步，增量同步，先全量后增量3种数据同步。代码不复杂，参考网上例子几十行代码即可实现，但问题是同步速度太慢，我们就是要解决这个问题，我们的程序是从MySQL 同步数据到 MySQL 和达梦数据库。当然写入的目标库最好要关闭索引，禁用外键检查，调大 innodb 的 buffer poll ，log buffer poll 等参数，单次批量插入的数据条数不要特别大。

        Debezium 无法设置一次拉取的数据条数，这个不受用户端代码控制，它一次能拉取多少条数据和用户的网络带宽，内存，磁盘等硬件相关，我本机 16GB内存，1T磁盘，锐龙5800H 的机器每秒能拉取并处理 7千到1万 条数据，如果为了架构简单，不引入Kafka等消息队列，每次处理一条数据就写入到MySQL ，那MySQL 每秒可能只写入 10条左右的数据，慢到无法忍受；如果使用 CopyOnWriteArrayList 等线程安全的列表先缓存，再批量写入到MySQL，但这个写入到MySQL 的线程要和拉取并处理原始sql的线程抢锁，有时候甚至连续几分钟甚至几小时都抢不到锁，最终导致 CopyOnWriteArrayList 里积压了几万甚至几十万条数据，有内存泄漏导致数据丢失的风险。

        因为生产数据太快而消费太慢，而且消费数据的线程还是单线程，所以必须引入一个消息中间件去存储已处理好的消息，这样做可以保证数据不会丢失，多开启几个消费线程一起工作加快消费消息的速度。最终我们决定使用 Kafka 。

        Kafka 一个主题可以有多个分区，每个分区都可以被一个线程所消费，因为要写入消息到MySQL ，批量插入比单次插入效率高很多，如果还是单次插入到MySQL，那么引入这个Kafka就没什么用，还是那么慢，所以我们决定多开几个线程批量拉取Kafka里的数据，并批量插入到MySQL，因为当前我们程序每秒能单次插入1700多条数据到Kafka，所以我们的Kafka 一个主题开4个分区，每个分区都有一个线程批量拉取消息进行消费，到此消息消费速度刚好赶上了消息生成速度了。

@Component
@Configuration
public class KafkaConfig {@BeanKafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();// 设置批量拉取factory.setBatchListener(true);factory.setConsumerFactory(consumerFactory());factory.getContainerProperties().setPollTimeout(2000);return factory;}@Beanpublic ConsumerFactory<String, String> consumerFactory() {HashMap<String, Object> propsMap = new HashMap<>();propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");// 单次拉取最大数目，即MySQL 批量插入数据条数propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");return new DefaultKafkaConsumerFactory<>(propsMap);}
}

@Component
public class KafkaListen {    // 批量拉取，批量消费，concurrency  是并发线程，和Kafka的分区数一致@KafkaListener(topics = "data-sync", concurrency = "4", containerFactory = "batchFactory", groupId = "consumer-test")public void listen3(List<ConsumerRecord<String, String>> records) {ArrayList<Sync> syncs = new ArrayList<>();for (ConsumerRecord<String, String> record : records) {SqlData sqlData = JSONUtil.toBean(record.value(), SqlData.class);if ("null0".equalsIgnoreCase(sqlData.getGtid())) {sqlData.setGtid("");}Sync sync = Sync.builder().id(IdWorker.getId()).gtid(sqlData.getGtid()).tableName(sqlData.getTable()).createTime(new Date()).dataDefinition(sqlData.getSql()).database(sqlData.getDatabase()).build();syncs.add(sync);}System.out.println(DateUtil.now() + " " + Thread.currentThread().getName() + "  插入 " + syncs.size() + " 条数据到 MySQL ...");MysqlWriterUtil.addAll(syncs);}
}

       

刚才说每秒程序能生成 1w 多条数据但只有1700 多条数据被插入到 Kafka，因为我们每生成一条数据就插入到Kafka，不是每秒钟定时插入，所以慢了一点。这个怎么解决呢？使用Guava或是Caffeine 等缓存是不行的，虽然这些本地缓存框架可以缓存数据并按上次访问时间进行清除数据，清除数据时监听器还是按每一条数据进行清除的，这不满足我们的要求，我们的要求是按上次访问时间过去了1秒就把所有积累的数据批量插入到Kafka，我们可以这样做：新建一个CopyOnWriteArrayList 保存数据，新建一个时间变量保存前一秒钟，只要当前时间已经相对这个时间遍历过去了1秒，就把CopyOnWriteArrayList 里的所有数据批量写入到 Kafka，并刷新这个时间变量为当前时间，因为我们的程序是先全量后增量，所以不存在说最后一秒种生成的数据都还留在了内存没有被写入到Kafka 。

        所以数据同步一是要保证生成数据的准确性，二是保证能快速大量地从源端拉取数据，处理数据，写入到Kafka，然后写入到最终的目标库。