最近在搞物联网平台,海量级别的消息Push导致MQ处理速度下降,对MQ进行单队列性能压测,结果很不如意啊!下游设备是通过NB模块和ESP进行双链路数据采集,由于场景就是抄表,但是下游设备太多,老板也没给多少银子买云服务,所以只能自己研究一波儿了~

抄表也就意味着单Topic,进行测试的时候单个Topic消费端TPS到1.7w/s,大量的消息处于unconfirmed未确认状态,达到了TPS上限,然后通过新增消费端仍然是无法解决,那么就将性能瓶颈的视角转向了MQ服务。

对于瞬间大量并发的数据平台来说,1.7w感觉有点少,之前的处理对策是分批汇报,但是这次数据量太大,导致分批汇报也会出现这种问题

由于之前的项目对AMQP协议有进行压测,很明显MQTT比AMQP低很多。研究了一下rabbitmq_mqtt插件,发现是将MQTT转化为AMQP放入消息队列。那么是转换过程中出现了性能瓶颈吗?

研究发现:和QOS有关,delivery_mode在源码中是2,这表示每一条消息都要走磁盘I/O。那么为啥这个插件要这么设计呢?QOS=1表示消息最少到达一次,也就是失败后可以重发一次,消息持久化机制在Server挂掉的情况下也会保证消息不丢失,确保了QOS1的特性。但是抄表数据时累加的,而且考虑到某些数据的汇报实时性,因此放弃QOS=1方案:

// 修改src/rabbit_mqtt_processor.erl
delivery_mode(?QOS_1) -> 1;

然后进行测试,结果可达4w/s的TPS,同样硬件客户端代码也进行了修改,使得QOS等于0,那么表示这个消息处理平台只支持QOS=0了,这样虽然有可能损失部分数据,但是解决了消息堆积问题。