攒批分组（batchGroup） ：攒批分组的设计借鉴了Kafka Topic分组的思想，例如在消费上游消息进行内存攒批的场景中，我们接收到的消息由于Tag不同需要加入到不同的分组进行攒批时，就可以使用到“攒批分组”。攒批分组的生命周期比“攒批队列”要长很多，一个分组中一个时刻总会有1个活跃的攒批队列在进行攒批，如下图所示groupA中存在一个攒批ProducerBatch-groupA，当这个攒批攒满以后，会重新创建一个新的攒批队列再次进行攒批，但是这个新的攒批队列所属的分组名称是不变的，仍然是groupA。

攒批容器（ProducerBatchContainer） ：攒批容器则是用来管理多个攒批分组的容器，通过这个攒批容器可以知道当前有多少攒批队列在进行攒批。

攒批回调（CallBack） ：攒批回调是个可选功能，如果在业务场景中我们需要知悉每次追加的消息内容是否被成功消费，则可以在将数据加入攒批队列时指定回调，这样当这批数据被消费的时候就会触发成功回调函数。

三、整体架构

四、快速入门

（1）创建一个生产者配置类，最简单的生产者配置示例如下：

此外还可以再去设置攒批最大占据内存大小，攒批容器最大占据内存大小，等参数。详情可以参考 ProducerConfig 类进行设定。

（2）创建 MessageProducer 对象，同时指定攒批条件达成时，需要执行的业务逻辑，示例如下：

在消费代码中，如果主动返回 MessageProcessResultEnum.SUCCESS; WBB会认为当前攒批已经被下游成功消费，返回 SUCCESS 则认为消费成功，返回RETRY 则会进入重试队列，等待重试。需要关注的是如果你的业务代码处理时出现了异常也会加入重试队列。

（3）接收上游的消息开启内存攒批，这里以RocketMQ消息消费为例：

调用messageProducer的send方法以后，就会将收到的消息加入到攒批过程中，在上面的例子中当攒批队列长度达到20个元素以上或者攒批时间已经超过100s时（或者攒批队列的占据最大内存超过设定值，默认8M）就会触发攒批消费，也就是会执行创建MessageProducer 时指定的业务逻辑。

此外WBB还支持设置攒批回调，也就是为每一次往攒批队列中追加操作设置回调函数，如下面的示例代码所示，当消息被追加到攒批队列中并被成功消费后，会执行回调函数。

五、关键设计与源码分析

5.1 WBB核心对象建模

面向对象设计思路最重要的就是对事物的抽象，在WBB中主要的建模有两个，一个是“攒批队列”一个是“攒批容器”。

攒批队列的基本属性和行为如下，一个攒批队列实际最重要的就是一个List结构，此外还需要记录重试信息、当前队列的空间占用等信息，并暴露tryAppend() 方法由攒批容器来调用，从而进行数据追加。

攒批容器则是主要是对多个攒批队列的管理，承担管理者的角色，对于需要攒批的数据需要找到匹配的攒批队列将数据追加进去，此外还需要对外暴露部分接口，让外部领域了解攒批的具体运行情况。

5.2 攒批功能实现

在WBB中一个攒批操作，实际上就是一个将消息追加到消息队列中的动作。核心入口位于com.waiting.bus.core.containers.ProducerBatchContainer#doAppend方法中。在这个方法中通过 Semaphore 来保证所有攒批队列的总容量不会超过设定值（避免内存爆炸问题），每次往队列中追加元素，都需要使用semaphore来确保能够申请到一批存储空间。此外在攒批时会考虑是否满足“转化条件”也就是说是不是要停止攒批开始消费过程，关键代码如下：

细心的读者可能发现了追加消息的时候检查只能检查攒批队列的长度或者占据的内存大小是否触发攒批发送条件，如果设定的条件最多内存驻留5s，时间到时无论攒批了多少元素，都需要进行下游消费，这种攒批条件应该如何实现。事实上，WBB的攒批时间检测则是交给ExpireBatchHandler 来实现的，核心代码如下图所示：

ExpireBatchHandler本质是一个线程对象，这个线程在启动之后会不断检测攒批容器和重试队列是否存在过期的攒批任务，如果存在就会将这些到达存活时间的攒批队列转为“攒批消费任务”并执行指定的业务逻辑。

5.3 攒批重试

WBB支持重试逻辑，当业务对攒批的处理出现异常或者明确返回失败的时候，如果还有重试机会，WBB会将执行失败的攒批队列加入重试队列，整个重试的逻辑使用的是JAVA并发包的神器 —— DelayQueue

业务执行失败的攒批如果还有重试机会就会加入到DelayQueue 进行排队。这里额外解释下DelayQueue的功能，DelayQueue 是一个无界的 BlockingQueue，用于放置实现了 Delayed 接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是线程安全的，适合用来实现延时功能。

在WBB中当一个 ProducerBatch 攒批失败以后，会给这个攒批计算一个延时时间值，保证这个攒批只有在延时时间到了以后才能被再次取到。如下面的代码所示在加入重试队列之前会对这个队列计算一个下次能够被取出的时间点，这样在指定时间点未到的时候无法从队列中取出这个元素。

有兴趣的同学可以去了解下 DelayQueue的poll方法实现，代码很短，其实就是一个经典的“生产者消费者”模型。这里也考察下读者，如果现在让你写出生产者消费者模型的代码你能有多少种实现方案？这也是个面试常考的问题……
在WBB设计中使用到了 BlockingQueue、EvictingQueue（GUAVA）、DelayQueue三种队列，后期可能会总结一篇文章专门介绍下这些队列的使用的原理。

此外有过业务开发经验的同学一般都会清楚，重试一般情况下会有个延时梯度，也就是说第一次失败以后会间隔较短的一段时间重试，后面每次重试都会加大重试时间，在WBB的重试的时间模型设计的比较简单，就是在用户设定的基础重试时间的基础上每次乘以2，比如用户规定基础重试时间是1s那么后面依次的重试时间为2s、4s、8s.....直到达到用户规定的最大重试时间, 代码核心实现如下。

5.3 优雅关闭实现

优雅关闭是一个很重要的需求，考虑下面的几种情况。

情况1： 需要关停组件，但是此时还有数据在内存攒批没有达到攒批完成条件触发消费，这时候如果不做任何处理可能这批数据就没能正确的执行业务逻辑。

情况2： 需要关停组件，但是此时还有消息正在加入攒批队列，还没有成功追加到攒批队列中，此时如果直接关停，可能导致这些消息没有被正确处理。

情况3： 需要关停组件，但是用户的消费业务逻辑还没有执行完毕，期望的是用户指定的消费业务逻辑执行完毕后再退出。

WBB在设计时充分考虑了上面的这三种问题。

对于情况1，ExpireBatchHandler中设计了一个close标记位，当close标记位为false的时候所有的while循环都会退出执行，例如ExpireBatchHandler的部分实现如下：

在上面的代码中当close方法调用之后，while循环会及时退出，然后会检测攒批容器中是否存在未完成的攒批，如果有未完成的攒批会直接提交攒批队列。

对于情况2 ，则在攒批容器中设计了一个AtomicInteger类型的appendsInProgress作为追加计数器，如下面的代码所示，这样在关闭的时候只需要不停检查计数器是否归零就可以判断出是否存在元素还正处于追加过程中。

对于情况3则比较简单，因为业务的消费逻辑都在单独的线程池中执行的，只需要执行线程池的优雅关闭策略即可。

当然优雅关闭的这些策略能够实现的前提是组件不能被强制Kill，因为WBB本质不是分布式组件，仅仅是基于内存设计的，所以WBB不能完全避免消息丢失问题，这个也是这个组件目前比较不完善的地方。

六、总结

本文重点介绍了一种攒批队列的设计与实现，在业务开发过程中如果遇到需要进行“延迟处理”的场景时可以考虑本文的部分设计思路和源码进行实现。

在实现WBB的过程中一个很深的感触就是一个很简单的需求往往需要注意的细节越多，攒批的功能很简单但是需要考虑内存占用、并发、线程安全、组件启停等方方面面，想要写好一个工具其实需要花费很多心思。

最后将最近一段时间比较喜欢的Charles Eames的一句名言送给各位读者“The details are not the details. They make the design”我翻译为 “细节成就设计”，在软件开发中要永远尽可能多的去考虑细节。

Last update: 2024-10-23