首先根据 binlog 事件类型来决定是否调用 flush 方法，这个就是实现将一个事务的事务一起提交到消费端，回到环形缓存区的具体实现，我们重点关注 put 方法 与 flush 方法的实现。

EventTransactionBuffer#put


其实现的核心步骤：

1. 检测当前环形缓存区是否已满，如果未满，则向缓存区中添加一条数据。添加数据的具体逻辑：

• 获取下一个写入的序号 next，等于当前已写入的序号 + 1，即 putSequence + 1。

• 通过 next & indexMask 取得放入 CannalEntry.Entry[] entries 中的下标，与 next % bufferSize 效果等同。

2. 如果已满，则首先将缓存区中的数据刷新，即将未处理的数据全部抽取，提交到数据消费方，然后释放缓存区，继续添加数据。

关键在于如何判断环形缓存区已满，具体算法如下：

EventTransactionBuffer#checkFreeSlotAt

为了加强对这段代码的理解，我举一个示例，在一个 8 个元素的环形缓存区中，假设一个事务包含 5 条日志，首先依次写入 5 条日志，其环形缓存区如下：

此时 putSequence 为 4，flushSequence 为 -1 ，我们应该能发现，在第一轮时，由于 sequeue 小于 bufferSize ，如果不执行 flush 操作，连续写入 8 条数据，sequence = 7 时，sequence - bufferSize > flushSequence 这个表达式都不会满足，即代表缓存区未满，但在写入第 9 条消息时，sequence = 8 ，此时 sequence - bufferSize > flushSequence 已满足，即缓存区已满，需要先刷新数据，然后才能再填充。

再回到本示例中，一个事务只包含 5 条日志，在写满 5 条日志后会即调用 flush 方法，将环形缓存区中下标为 0~4 的消息传入数据消费方，在 Canal 中会将这批消息一次传入 EventSink 组件。执行完 flush 方法后，flushSequence 等于 4，其环形缓存区如下图所示：


此时 putSequence = flushSequence = 4，那这个时候环形缓存区的容量是多少呢？其实就是又恢复到 bufferSize 了，那我们怎么计算环形缓存区当前已写入的消息呢？其实很简单，putSequence - flushSequence 表示已写入的元素数量。那当前剩余容量就等于 bufferSize - (putSequence - flushSequence )，即只需要 bufferSize - (putSequence - flushSequence ) > 0 就表示有剩余空闲。有了这一层思路，就能明白 checkFreeSlotAt 的算法，这也是环形缓存区的核心所在。

思考，Canal 基于环形缓存区的实现，一定能保证一个事务的所有变更日志都一次提交到 EventSink 组件吗，大家可以简单思考一下，在文末的总结部分有笔者的思考。

答案是否定的，如果一个事务包含的日志条目超过了环形缓存区的长度，为了保证数据不丢失，会首先将环形缓存区的数据全部提交，然后接收新的数据，这样一个事务中的消息会被分成多次提交到 EventSink。

好了，本文就介绍到这里了，您的点赞与转发是对我持续输出高质量文章最大的鼓励。

欢迎加笔者微信号(dingwpmz)，加群探讨，笔者优质专栏目录：

1、[源码分析 RocketMQ 专栏(40 篇+)](()

2、[源码分析 Sentinel 专栏(12 篇+)](()

最后

无论是哪家公司，都很重视基础，大厂更加重视技术的深度和广度，面试是一个双向选择的过程，不要抱着畏惧的心态去面试，不利于自己的发挥。同时看中的应该不止薪资，还要看你是不是真的喜欢这家公司，是不是能真的得到锻炼。

针对以上面试技术点，我在这里也做一些分享，希望能更好的帮助到大家。