MQ·将多消息合并为一条消息的发送、消费的设计与实现

以下文章来源于Java艺术，作者wujiuye

优质文章，及时送达

这是笔者最近处理一个叫异步大点击的业务问题所思考出来的方案。由于mq使用的是亚马逊的sqs服务，而sqs是按请求数消费的原因，所以才有的将多消息合并为一条消息发送的想法。

这个想法从sqs的消息批量发送以及阿里限流中间件的qps统计、netty的EventLoopGroup设计中得到启发。本篇将介绍如何将多个消息合并成一个消息发送而不影响服务的并发性能，以及由于合并后产生的大消息消费出现的消息堆积现象，开的消费者越多反而消息堆积越多的bug。

为什么要将多消息合并为一个消息发送？

前面也说了，为了节约成本。以每分钟50w的广告点击数来算，一个月将产生50＊60＊24＊31w的点击消息，再乘以3就是每个月的sqs请求数，3代表的是发送消息、拉取消息、删除消息，按每100w请求0.4美刀的价格计算大概一个月要26784美刀。

由于sqs限制单条消息的大小最大为256k，根据业务场景估算每点击消息也不可能达到1k,，所以我将256个请求合并为一个消息发送，或者1s内未达到256个消息也合并为一个消息发送，这样每月的费用可以直接除以256，这不是一个小数目。

什么样的业务场景下才适合这么干？

将大量消息合并为一个消息后会导致消息消费失去原子性。你无法保证原本是256个消息的合并为一个消息后，这256个消息能全部消费成功或者全部消费失败，因此要求业务必须允许消息消费失败直接丢弃的情况。无论多少个成功多少个失败，都需要将整条消息从mq中删除。笔者考虑过这个问题才决定是否要这样做的，也考虑过失败重试的问题，但我觉得没必要为这种概率买单，因为一个点击在非异步的情况下，失败就是失败了。

如何将大量消息合并为一条消息发送而不影响服务的高并发性能呢？

其实不影响是不存在的，只是让影响变得微弱。经过长时间的观察，我了解该高并发服务对内存的消耗并不高，最大qps下也就消耗1.5g左右的堆内存，而netty使用的直接内存大概在2g这样，对于2核8g的机器，有足够多的内存来实现队列缓存消息。

我借签Dubbo的客户端与服务端配置多个连接时使用轮询方式使用连接，同时也借签了netty的EventLoop的设计，实现消息合并发送。我定义一个MesaageLoopGroup，一个MesaageLoopGroup可以配置有多少个MesaageLooper，而每个MesaageLooper就是一个线程，且维护一个阻塞队列，默认队列大小是102400，这个数字是我配置单个进程所能打开的最大文件句柄数。

当往MesaageLoopGroup push一个点击消息时，先用原子类自增1与MesaageLooper数组的长度取余，选出一个MesaageLooper。然后再将消息push到这个MesaageLooper的阻塞队列。

每个MesaageLooper的run方法实现的就是一个死循环，从阻塞队列中拿消息，当消息等于256时，或者阻塞超过1s就将拿到的消息合并成一个消息发送到mq。如果阻塞队列满，那么push会直接将消息发送到mq。因此，服务重启时如果使用kill 9强行结束进程，至多只会有1s的数据丢失。设置1s还有一个原因就是控制消息的实时性。

灰度上线测试一天后也证明此方案对服务的影响并不大，无论是gc还是内存占用，都看不出加了这么一层逻辑。1s的平均请求按50w计算，四台机器分担，每个服务的每秒请求数平均是2000。

为何用golang实现消费者？

然而消息的消费并不顺利。一个是因为消息消费我用了golang实现，我也是刚入门，写起代码来还感觉别扭，二是一个消息是由原本256个消息组合而成的问题。

使用golang其实是有原因的。原本计划是让消费者占用较小的内存，以实现将消费者寄生在其它服务所在的机器上，充分利用其它耗内存而cpu利用率低的服务所在的机器。同时利用docker实现快速部署，让docker 的镜像更小，不需要安装jdk什么的。还有就是利用go的协程并发处理能力吧，让消费者消费消息的速度能赶上消息的产生速度。

为入门golang买单

为了便于理解，我还是以java的线程池来说明。假设我配置的线程池线程数量是512。寄生在其它服务的机器上需要给主人点面子，不能把人家的cpu全部吃完，导致主服务不可用，所以线程的数量结合消息的消费情况综合考虑，不能超过一半的cpu使用率，而选择512这个数量。

Sqs支持一次拉取多条消息，并且有一个可见性超时的特性，当消息被消费者拉取到之后，在多长时间内未删除，下次可能还会被拉取到，或者其它消费者还能拉取到。最初我设置的可见性超时是60s。

一开始我开启5个线程拉取消息，每次最多拉取10条消息。那么很可能同一时间内会拉取到50条消息。由于一条消息是由原本256条消息合并而成的，所以512个线程同一时间段至多只能消费2条消息，而一条消息(合并后的)的消费平均耗时是10s，也就是说一分钟内最多消费12条消息，其它38条消息在一分钟后会被其它消费者拉取到，所以就会出现大量消息重复消费的情况，久而久之，消息越积累越多。

我用golang的channel实现生产者与消费者，channel的大小可设置，当channel满时，拉取到的消息是放不进channel的，因此会将拉取线程阻塞住，只有消费者从 channel取数据才能继续放入。但阻塞的那段时间要小于消息的可见性超时，因为消息只有在开始消费时我才会将其从mq中删除。

后面的改进就是根据消费能力去调整消息的拉取线程数，以及每次拉取的消息数。还有一点要注意，为保证时刻有消息准备就绪开始消费，最好不要让消息消费完再从mq中拉取。但这也会导致另一个问题，一些消息拉取到本地后，由于channel已满，放不进，而其它空闲消费节点又拉不到，导致消息被消费到的时间延长。这就需要作出取舍。

祝大家在2020年工作顺路，家庭幸福，合家团圆