分库分表—3.详细介绍二

技术分享 3周前 (12-08) 0 999+

关注

大纲

18.基于Canal和RocketMQ的增量同步

19.增量同步任务的背景介绍

20.增量同步任务查询与线程池提交

21.RocketMQ里的binlog消息的消费逻辑分析

22.新增binlog的数据同步逻辑分析

23.binlog基于内存队列的异步转发逻辑

24.基于CAS加锁的读写队列互换机制

25.binlog基于内存的merge合并逻辑

26.对merge数据从目标库里分批查询

27.对merge数据基于目标库数据做过滤

28.将过滤后的merge数据写入目标库

29.offset提交线程的启动和逻辑分析

30.增量同步过程中binlog写入失败的恢复

31.增量同步过程中的各种失败场景的恢复机制

32.定时移除已提交的增量同步消息

33.增量与全量并行运行的场景分析

34.增量与全量并发同步一批数据的冲突

35.同步完成后的数据校验逻辑分析

36.数据迁移完成后的无损发布方案

37.分库分表线上运维与扩容方案

 

整个分库分表的流程：

全量同步 + 增量同步 -> 让多库多表数据和单库单表数据持平 -> 数据校验 -> 无损发布 -> 老系统下线 -> 线上多库多表DDL运维 -> 多库多表再次扩容

 

18.基于Canal和RocketMQ的增量同步

(1)在全量数据同步的过程中存在的问题

(2)增量同步方案概述

 

(1)在全量数据同步的过程中存在的问题

刚从源数据库查出一批数据写入到目标库，已经开始去同步下一批数据了，但上一批数据可能又在源数据库里进行了修改或删除操作。为了处理这种问题，就需要引入增量同步方案。

 

(2)增量同步方案概述

在发起全量数据迁移任务之前，需要基于Canal去监听源数据库的增删改binlog。Canal监听到binlog后需要将binlog消息发送到RocketMQ消息中间件集群中，接着数据迁移系统上会有一个RocketMQ消费者去消费这些binlog消息，这样数据迁移系统就会把源数据库的增删改操作往目标库中进行同步。

 

19.增量同步方案的整体介绍

(1)全量同步是一个滚动查询 + 数据插入的过程

(2)增量同步是为了解决全量同步中已同步数据出现的数据变动问题

(3)增量同步的数据高效写入和数据防止丢失方案

 

(1)全量同步是一个滚动查询 + 数据插入的过程

其中会涉及：一批批查 + 设置滚动ID + 数据过滤 + 去重校验 + 记录迁移明细 + 中断恢复 + 进度统计。

 

(2)增量同步是为了解决全量同步中已同步数据出现的数据变动问题

开启全量同步之前，都会先开启增量同步。

 

增量同步需要分析以下三种情况：

情况一：还没同步的一批数据发生增删改

情况二：正在同步的一批数据发生增删改

情况三：已经同步的一批数据发生增删改

 

(3)增量同步的数据高效写入和数据防止丢失方案

在增量同步中，首先会通过Canal监听源数据库中的binlog⽇志，然后Canal再将监听到的binlog⽇志发送放到RocketMQ中，接着数据迁移系统会消费RocketMQ中的binlog消息，把增删改操作同步到目标数据库。

 

问题一：数据迁移系统消费MQ消息时，如何保证从MQ获取到的binlog消息不会丢失

如果源数据库增删改操作了，但由于消费异常导致binlog消息丢失了，那么目标数据库中就没有对应的增量数据操作，这样源数据库和目标数据库的数据就会不⼀致。为了避免消费异常导致binlog消息丢失，需要设置禁止自动提交消息。

 

消费MQ的binlog消息时，为了提升消费速度，可以采用多线程进行消费。比如每消费一条MQ消息，就向线程池提交一个任务，任务执行完才提交消息。当这些任务的执行速度慢于消费MQ消息的速度时，线程池的阻塞队列中就会积压一些任务。如果此时机器发布重启，那么就可能会导致线程池中阻塞队列里积压的任务丢失。但是由于禁止消息自动提交，所以这些丢失任务对应的MQ消息后续还可以重新被消费，然后再次被提交到线程池中进行处理。

 

为了方便对binlog消息进行管理和确保binlog消息不丢失且有记录可查，这里引⼊消息拉取落库和异步消息提交机制，由两个定时任务来完成。如下所示：

⾸先源数据库中会有⼀张消费记录表，定时任务1每次从MQ拉取并消费⼀条消息时，都会先在消费记录表中新增⼀条消费记录，每条消费记录的初始状态都为未消费。然后定时任务1再将获取到的binlog消息，在目标数据库中重做对应的binlog⽇志。也就是将旧库中的增删改操作，在目标数据库中重做⼀遍。重做完成后，再来更新刚刚添加的消费记录的状态，从未消费更新为已消费状态。

 

此时需要注意：定时任务1消费MQ的binlog消息后，并不是自动向MQ提交消息，⽽是需要进行⼿动提交。否则如果消息都没有消费成功，就自动向MQ提交消息，则可能会出现消息丢失的情况。所以为了保证binlog消息不丢失，不会⾃动提交消息，⽽是将提交消息的任务交给定时任务2来处理。

 

定时任务2会专⻔从消费记录表中，查询已消费的那些记录，然后向MQ提交消息，这样下次就不会从MQ中消费到了。向MQ提交完消息后，同时会将消费记录表中的记录状态，从已消费更新为已提交。⾄此，⼀个消息的消费流程才算结束。

 

问题二：如何提高增量同步时的数据写入效率

为了提高数据写入目标数据库的效率，这里引入了数据合并、过滤、读写队列的机制，读写队列和数据合并流程图如下：

定时任务1添加完消费记录后，并不会⻢上把数据写入目标库，⽽是把binlog日志先放到⼀个写队列中，与写队列相对的还有⼀个读队列。读队列是专⻔用于提供给定时任务3进行处理消息写⼊操作的。

数据合并提升写入效率：如果源数据库中的数据在短时间内进⾏了多次操作，其实只需要保留最新的binlog⽇志即可。所以才使用了一个内存队列来存放binlog消息，而且会每隔15秒批量处理一次内存队列的所有binlog消息，以此减少同一条数据对应多条binlog的写入处理。

binlog日志的处理细节：从合并后的binlog⽇志中获取主键ID，根据主键ID到目标库中查询对应的数据。

如果目标库中能查到这条数据，那么需要和源数据库的binlog数据进⾏对⽐。只有当源数据库的更新时间⼤于目标库的更新时间，才允许更新数据到目标库中。如果当前的binlog⽇志的操作类型为删除操作，则可不⽤对⽐更新时间，直接在目标库中重做这条binlog⽇志，毕竟源数据库在删除⼀条数据时不会更新修改时间。

如果源数据库的⼀条binlog⽇志对应的数据在目标库中没有查到，那么继续判断。如果binlog⽇志是删除操作，那就没必要在目标库中重做这条⽇志了，直接过滤掉。目标库都没有数据了，就没必要执⾏删除操作。如果binlog⽇志的类型为修改操作，那也没必要执⾏修改操作。因为目标库没数据，直接update也不⾏，可以将binlog的操作类型修改为新增操作。毕竟在binlog⽇志中，包含了⼀条订单数据的所有字段的值，⾜以满⾜新增数据需要的所有字段。

经过以上的数据过滤操作，⼀⽅⾯避免源数据库中的旧数据覆盖了目标库的新数据，另⼀⽅⾯避免了没必要执⾏的删除和更新操作也在目标库中继续执⾏。

20.增量同步任务查询与线程池提交

CanalConsumeTask继承自ApplicationRunner，也就是系统启动的时候，它就会跑起来。

CanalConsumeTask首先会查出当前配置好的需要滚动查询全量数据的迁移任务。每个滚动查询全量数据的迁移任务就对应一个增量同步任务。然后创建一个线程池，线程数量 = 已配置好的滚动查询全量数据的迁移任务的数量。接着遍历每一个滚动查询数据的迁移业务，提交两个任务到线程池中。其中的CanalPullRunner拉取任务，会设置禁止Consumer自动提交offset，只拉取不提交。另外的CanalPullCommitRunner提交任务，此时才会提交offset。

需要注意的是：从RocketMQ里消费binlog消息时，需要避免Consumer自动去提交offset。需要精准控制offset提交，当每一条binlog都已经被应用到目标数据里后，才能对这条offset进行提交。因为RocketMQ的Consumer提交offset默认是自动提交：即会先提交到本地缓存，再提交到RocketMQ。而这就可能会导致offset提交时数据还没被应用到目标数据库。

@Component public class CanalConsumeTask implements ApplicationRunner {     //RocketMQ的nameServer地址     @Value("${rocketmq.name-server:127.0.0.1:9876}")     private String nameServerUrl;      //可以从migrateConfigService拿到增量同步配置     //要从RocketMQ里监听到binlog变更，需要知道要关注的是哪些库哪些表的binlog变更     @Autowired     private MigrateConfigService migrateConfigService;      //ApplicationRunner在系统启动时就会运行run()方法     @Override     public void run(ApplicationArguments args) throws Exception {         //首先查出当前配置好的需要滚动查询全量数据的迁移任务，每个滚动查询全量数据的迁移任务就对应一个增量同步任务         List<ScrollDomain> scrollDomainList = migrateConfigService.queryScrollDomainList();          //这里会创建一个线程池，线程数量 = 已配置好的滚动查询全量数据的迁移任务的数量         ExecutorService executors = Executors.newFixedThreadPool(scrollDomainList.size());          for (ScrollDomain scrollDomain : scrollDomainList) {             //接下来会提交两个任务             if (scrollDomain.getDataSourceType().equals(1)) {                 //从RocketMQ里消费binlog消息时，需要避免Consumer自动去提交offset                 //需要精准控制offset提交，当每一条binlog都已经被应用到目标数据里后，才能对这条offset进行提交                 //因为RocketMQ的Consumer提交offset默认都是自动提交：也就是会先提交到本地缓存，再提交到RocketMQ                 //而这就可能会导致offset提交时数据还没被应用到目标数据库                  //执行拉取任务，此时设置Consumer不自动提交offset，只拉取不提交                 executors.execute(new CanalPullRunner(scrollDomain.getDomainTopic(), nameServerUrl));                  //执行提交任务，此时才会提交offset                 executors.execute(new CanalPullCommitRunner(scrollDomain.getDomainTopic(), nameServerUrl));             }         }     } }  //需要滚动查询全量数据的迁移任务配置表 @Data public class ScrollDomain implements Serializable {     //主键ID     private Long id;     //所属系统(会员、订单、交易)     //需要进行增量同步的表，是来源于哪个业务的，这个业务可以是会员、订单、交易等     private String domain;     //当数据源为来源的时候，配置对应的消息topic     //需要进行同步的表，该表的binlog会被写入到RocketMQ的哪个topic里去     private String domainTopic;     //数据源类型，1需要读取数据，2需要写入数据     private Integer dataSourceType;     //是否显示 ShardingSphere SQL执行日志     private Integer sqlshow;     //每个逻辑库中表的数量     private int tableNum; }

21.RocketMQ里的binlog消息的消费逻辑分析

RocketMQ里的binlog消息会由线程池里的CanalPullRunner任务来处理。往线程池提交CanalPullRunner任务时，会传入RocketMQ的地址和主题。然后CanalPullRunner任务首先会设置RocketMQ Consumer禁止自动提交offset，让当前Consumer不要自动去提交offset，防止还没处理完binlog消息就提交offset。如果offset已经被自动提交，但是binlog消息却处理失败，那么RocketMQ就不会让消费者再次消费了。之后CanalPullRunner会设置Consumer订阅指定的topic和nameServer地址，然后启动这个Consumer。

接着会进入while死循环中，并不断通过Consumer从RocketMQ中一批批的Pull消息出来消费处理。对于Consumer从RocketMQ中Pull到的每一条消息，会获取对应的topic、queue、offset、body。然后判断该消息是否已被处理过，即是否已经存在于消费记录表中，如果存在则跳过执行。如果还没处理过，那么调用processNewMsg()方法进行处理，并往消费记录表中插入一条记录。

//binlog消息拉取任务(只拉不提交) public class CanalPullRunner implements Runnable {     //消息主题     private final String topic;     //RocketMQ的NameServer地址     private final String nameServerUrl;     //binlog消息同步消费记录表Mapper     private final EtlBinlogConsumeRecordMapper consumeRecordMapper;      //消息拉取任务构造方法     //@param topic         消息主题     //@param nameServerUrl RocketMQ的NameServer地址     public CanalPullRunner(String topic, String nameServerUrl) {         this.topic = topic;         this.nameServerUrl = nameServerUrl;         this.consumeRecordMapper = ApplicationContextUtil.getBean(EtlBinlogConsumeRecordMapper.class);     }      @Override     public void run() {         pullRun();     }      //执行消息拉取     private void pullRun() {         try {             DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("binlogPullConsumer");             //设置RocketMQ Consumer禁止自动提交offset，让它不要自动去提交offset，防止还没完处理完binlog消息就提交offset             //如果offset已经被自动提交，但是binlog消息却处理失败，那么RocketMQ就不会让消费者再次消费了             litePullConsumer.setAutoCommit(false);             litePullConsumer.setNamesrvAddr(nameServerUrl);             litePullConsumer.subscribe(topic, "*");             litePullConsumer.start();              try {                 //进入while死循环中，通过Consumer从RocketMQ中一批批的Pull消息出来消费处理                 while (true) {                     //拉取未消费消息                     List<MessageExt> messageExts = litePullConsumer.poll();                     if (CollUtil.isNotEmpty(messageExts)) {                         for (MessageExt messageExt : messageExts) {                             byte[] body = messageExt.getBody();                             String msg = new String(body);                             //记录queueId和offset                             int queueId = messageExt.getQueueId();                             long offset = messageExt.getQueueOffset();                             String topic = messageExt.getTopic();                              //topic、queue、offset、msg，四位一体，把所有的信息都拿到                              //判断该消息是否已被处理过，即是否已经存在于消费记录表中，如果存在则跳过执行                             EtlBinlogConsumeRecord existsRecord = consumeRecordMapper.getExistsRecord(queueId, offset, topic);                             if (null == existsRecord) {                                 //如果还没处理过，那么进行处理，并往消费记录表中插入一条记录                                 processNewMsg(messageExt, msg);                             } else {                                 //处理已经存在的消费记录                                 proccessExistsRecord(litePullConsumer, msg, existsRecord);                             }                         }                     } else {                         Thread.sleep(5000);                     }                 }             } finally {                 litePullConsumer.shutdown();             }         } catch (InterruptedException | MQClientException e) {             try {                 //假设要拉取消息的主题还不存在，则会抛出异常，这种情况下休眠五秒再重试                 Thread.sleep(5000);                 pullRun();             } catch (InterruptedException ignored) {                          }         }     }     ... }

22.新增binlog的数据同步逻辑分析

对于新增的binlog数据，会调用CanalPullRunner的processNewMsg()方法进行处理。

由于首先拿到的msg是一个字符串格式的binlog，所以需要对该字符串格式的binlog做一个解析，然后把这个解析后的binlog的信息封装到自定义的BinlogData对象里去。

由于binlog的字符串格式是json格式，所以可以把binlog的字符串转成json对象。然后从json对象里提取一个一个字段出来，构建BinlogData对象。

拿到一个BinlogData对象之后，接着会封装binlog消费处理记录对象EtlBinlogConsumeRecord。此时不会把binlog的数据放到EtlBinlogConsumeRecord里，但会设置topic、queue、offset等信息。

之所以关闭自动提交offset的功能，是因为这里的每一条binlog消息都将被异步处理。也就是每一条binlog消息，都会提交到本地队列来实现异步化处理。所以如果异步化都没处理完毕，就自动提交了offset告诉MQ已处理成功，那是有问题的。

因此这里Consumer消费消息时，会把消息先放入本地队列。然后把binlog在RocketMQ里的topic、queue、offset封装成Record对象，插入到DB。

当Record对象插入DB后，binlog自己的数据才会提交到本地队列进行异步化处理。此时offset还不会提交给Broker，必须等到异步化处理完该binlog消息后，才能提交offset。

public class CanalPullRunner implements Runnable {     ...     //处理新的消息     //@param messageExt mq消息对象     //@param msg        消息内容     private void processNewMsg(MessageExt messageExt, String msg) {         try {             //由于首先拿到的msg是一个字符串格式的binlog，所以需要对该字符串格式的binlog做一个解析             //然后把这个解析后的binlog的信息封装到自定义的BinlogData对象里去             BinlogData binlogData = BinlogUtils.getBinlogDataMap(msg);             Boolean targetOperateType = BinlogType.INSERT.getValue().equals(binlogData.getOperateType()) || BinlogType.DELETE.getValue().equals(binlogData.getOperateType()) || BinlogType.UPDATE.getValue().equals(binlogData.getOperateType());             if (!targetOperateType || null == binlogData || null == binlogData.getDataMap()) {                 return;             }              //拿到一个BinlogData对象之后，接着会封装一个binlog消费处理记录EtlBinlogConsumeRecord             //此时不会把binlog自己的数据放到EtlBinlogConsumeRecord里，但会设置topic、queue、offset等信息              //之所以关闭RocketMQ Consumer自动提交offset的功能，是因为这里的每一条binlog消息都将被异步处理             //也就是每一条binlog消息，都会提交到本地队列里，依托本地队列来实现异步化处理             //所以对于Consumer来说，如果异步化都没处理完毕，就自动提交了offset，告诉MQ已经处理成功了，那是有问题的              //因此这里Consumer消费消息时，是先放入队列处理             //然后把binlog在RocketMQ里的topic、queue、offset封装成EtlBinlogConsumeRecord对象，插入到DB里             EtlBinlogConsumeRecord consumeRecord = new EtlBinlogConsumeRecord();             consumeRecord.setQueueId(messageExt.getQueueId());             consumeRecord.setOffset(messageExt.getQueueOffset());             consumeRecord.setTopic(messageExt.getTopic());             consumeRecord.setBrokerName(messageExt.getBrokerName());             consumeRecord.setConsumeStatus(ConsumerStatus.NOT_CONSUME.getValue());             consumeRecord.setCreateTime(new Date());             consumeRecordMapper.insert(consumeRecord);              //EtlBinlogConsumeRecord对象插入DB后，binlog自己的数据会提交到本地队列进行异步化处理             //此时offset还不会提交给Broker的，必须等到异步化处理完该binlog消息后，才能提交offset给Broker             LocalQueue.getInstance().submit(binlogData, consumeRecord);         } catch (Exception e) {             log.error("新增消费记录失败", e);         }     }     ... }

解析binlog的json字符串为对象的逻辑如下：

//MySQL binlog解析工具类 public abstract class BinlogUtils {     ...     //解析binlog json字符串     //@param binlogStr binlog json字符串     //@param dataType  解析后的data的类型(实体类还是map)     //@return BinlogData     //@throws ClassNotFoundException 找不到实体类异常     //@throws InstantiationException 实例化实体类异常     //@throws IllegalAccessException 非法访问异常     private static BinlogData getBinlogData(String binlogStr, String dataType) throws ClassNotFoundException, InstantiationException, IllegalAccessException {         //isJson方法里面会判断字符串是不是为空，所以这里不需要重复判断         //由于binlog的字符串格式是json格式，所以可以把binlog的字符串转成json对象，然后从json对象里提取一个一个字段出来         if (JSONUtil.isJson(binlogStr)) {             JSONObject binlogJson = JSONUtil.parseObj(binlogStr);             BinlogData binlogData = new BinlogData();             //表名             String tableName = binlogJson.getStr("table");             binlogData.setTableName(tableName);             //操作类型             String operateType = binlogJson.getStr("type");             binlogData.setOperateType(operateType);             //操作时间             Long operateTime = binlogJson.getLong("ts");             binlogData.setOperateTime(operateTime);             //获取数据json数组             JSONArray dataArray = binlogJson.getJSONArray("data");             if (null != dataArray) {                 Iterable<JSONObject> dataArrayIterator = dataArray.jsonIter();                 //遍历data节点并反射生成对象                 if (null != dataArrayIterator) {                     //binlog的data数组里数据的类型为实体类                     if (DATATYPE_DOMAIN.equals(dataType)) {                         //获取实体类名称                         String domainName = DOMAIN_PATH + '.' + StrUtil.upperFirst(StrUtil.toCamelCase(tableName));                         //获取表对应的实体类（这里出现异常就抛出去了，实际使用时应该捕获并记录日志，因为根据表名找不到对象，那么这个表的所有数据都无法同步，这种情况肯定要记录日志并告警的）                         Class<?> domainClass = Class.forName(domainName);                         List<Object> datas = new ArrayList<>();                         while (dataArrayIterator.iterator().hasNext()) {                             JSONObject jsonObject = dataArrayIterator.iterator().next();                             Field[] fields = domainClass.getDeclaredFields();                             //通过反射创建实体类实例，这里的异常也直接外抛，实际处理时需要记录这个异常并告警                             Object domain = domainClass.newInstance();                             for (Field field : fields) {                                 //根据属性名称反向取得对应的表中的字段名称，然后根据属性的类型取得字段值并通过set方法设置进去                                 String fieldName = field.getName();                                 String columnName = StrUtil.toSymbolCase(fieldName, '_');                                 //因为我们的属性是私有的，所以这里需要设置为可访问方便直接设值                                 field.setAccessible(true);                                 Object fieldValue = getFieldValue(field.getType(), columnName, jsonObject);                                 if (null != fieldValue) {                                     field.set(domain, fieldValue);                                 }                             }                             datas.add(domain);                         }                         binlogData.setDatas(datas);                     } else if (DATATYPE_MAP.equals(dataType)) {                         //binlog的data数组里数据的类型为Map                         List<Map<String, Object>> dataMap = new ArrayList<>();                         while (dataArrayIterator.iterator().hasNext()) {                             JSONObject jsonObject = dataArrayIterator.iterator().next();                             Map<String, Object> data = new HashMap<>();                             jsonObject.keySet().forEach(key -> {                                 data.put(key, jsonObject.get(key));                             });                             dataMap.add(data);                         }                         binlogData.setDataMap(dataMap);                     }                 }             }             return binlogData;         }         return null;     }     ... }

23.binlog基于内存队列的异步转发逻辑

CanalPullRunner在processNewMsg()方法处理一条新增的binlog时：首先会封装一条binlog消费处理记录写入到数据库中，然后再将该binlog消息提交到自定义的内存队列LocalQueue。

LocalQueue.getInstance().submit(binlogData, consumeRecord)

在往内存队列写入binlog消息时，会首先加一个轻量级的Atomic——PutBinlogLock。因为LocalQueue内存写队列writeQueue是LinkedList，LinkedList并不是线程安全的。成功加锁的线程才能把binlog消息写入到内存队列LocalQueue的内存写队列writeQueue里，然后释放锁。

这个轻量级Atomic锁会进行如下设计：假设多个线程同时来进行加锁，多个线程都会去对Atomic变量进行CAS操作。但只有一个线程可以把Atomic变量从true变为false，Atomic变量默认就是支持线程安全的。也就是只有一个线程可以完成加锁的逻辑(即在执行compareAndSet()前flag为true，执行后flag为false。而其他线程加锁都会失败并在进入自旋(因为执行compareAndSet()前flag=flag，执行后flag还是false。

//binlog消息拉取任务(只拉不提交) public class CanalPullRunner implements Runnable {     ...     private void processNewMsg(MessageExt messageExt, String msg) {         ...         consumeRecordMapper.insert(consumeRecord);         LocalQueue.getInstance().submit(binlogData, consumeRecord);         ...     }     ... }  //数据缓存阻塞队列类 public class LocalQueue {     private static volatile LocalQueue localQueue;     //提供锁的实例对象     private final PutBinlogLock lock = new PutBinlogLock();     //数据同步的写队列     private volatile LinkedList<BinlogData> writeQueue = new LinkedList<>();     //数据同步的读队列     private volatile LinkedList<BinlogData> readQueue = new LinkedList<>();     //由于可能会多线程进行并发读和写，所以一般定义为volatile类型，来保证线程之间的可见性     //isRead始终只有一些相隔15秒的线程在写     private volatile boolean isRead = false;          private LocalQueue() {          }          //构建一个单例模式对象     public static LocalQueue getInstance() {         if (null == localQueue) {             synchronized (LocalQueue.class) {                 if (null == localQueue) {                     localQueue = new LocalQueue();                 }             }         }         return localQueue;     }      //数据写入队列     //@param binlogData    MySQL的binlog对象     //@param consumeRecord 消费记录     public void submit(BinlogData binlogData, EtlBinlogConsumeRecord consumeRecord) {         //writeQueue是LinkedList，LinkedList并不是线程安全的         lock.lock();         try {             binlogData.setConsumeRecord(consumeRecord);             writeQueue.add(binlogData);         } finally {             lock.unlock();         }     }     ... }  //锁竞争类对象 //基于Atomic的轻量级锁 public class PutBinlogLock {     private final AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);      //加锁     public void lock() {         //假设多个线程同时来进行加锁         boolean flag;         do {             //多个线程都会去对Atomic变量进行CAS操作，只有一个线程可以把Atomic变量从true变为false，Atomic变量默认就是支持线程安全的             //也就是只有一个线程可以完成加锁的逻辑(此时执行下面compareAndSet前flag=true，执行后flag=false)             //其他线程CAS加锁都会失败并在这里进入自旋(此时执行下面compareAndSet前flag=flag，执行后flag还是false，从而自旋)             flag = this.putMessageSpinLock.compareAndSet(true, false);         } while (!flag);     }      //释放锁     public void unlock() {         //只有一个线程可以成功的执行cas，把false变为true         this.putMessageSpinLock.compareAndSet(false, true);     } }

24.基于CAS加锁的读写队列互换机制

对binlog的处理是基于定时批处理的，不是来一条binlog就处理一条binlog。默认会每隔15秒处理内存队列里writeQueue的binlog数据，从而实现统一的批处理。

所以会有一个定时调度任务IncrementTask，负责定时对增量binlog数据进行处理。该定时调度任务IncrementTask会每隔15秒跑一次处理内存队列writeQueue的binlog数据。也就是定时调度任务IncrementTask会每隔15秒执行一次LocalQueue的doCommit()方法。

当LocalQueue的doCommit()方法对内存队列writeQueue的binlog数据进行处理时，为了提升性能，避免处理内存队列writeQueue的binlog数据时，对writeQueue持有锁耗时过长。所以设计了一个交换队列readQueue，会快速将writeQueue的数据复制到readQueue中。只有两个队列在交换数据时才会加锁，避免对writeQueue的长时间操作。

首先会设置一个标记isRead表明是否正在处理binlog，isRead始终只有一些相隔15秒的线程在写。只有isRead为false时，才能对内存队列writeQueue的binlog数据进行批处理。开始进行批处理时，会设置isRead为true。队列交换前，readQueue是空的，writeQueue是有数据的。交换进行中，writeQueue会被加锁，等待交换完成才释放锁，才能往writeQueue里写。当然往writeQueue里写binlog数据时，也会加锁的，写完之后才释放锁。交换完成后，writeQueue是空的，readQueue是有数据的。当后续慢慢处理完readQueue里的binlog数据了，设置isRead为false。

需要注意：加锁都是基于CAS轻量级的加锁，为什么不去用JDK提供的线程并发安全的队列呢？因为JDK提供的线程并发安全的队列，仅仅是队列自己内部是线程并发安全而已。而这里我们需要确保多个队列queue，在同时操作时，也都是线程并发安全的。所以这里才需要我们自己去对队列writeQueue的操作进行加锁。