传统的消息队列(Kafka、RocketMQ等)经过多年打磨,在高性能、海量堆积、消息可靠性等诸多方面都已经做得非常极致,但在IoT物联网场景中,往往需要面临着海量的消息传递,传统的消息队列表现的“力不从心”。
IoT场景中,从应用服务器到嵌入式芯片,都需要传递事件消息,比如共享充电宝的开柜子、开灯指令从服务器发到设备、工业网关高频消息流等,在这些信息传递的过程中,队列最大意义在于让整个消息在不可控的环境中平稳运行,因为IoT设备时不时会由于故障或网络抖动会导致大量消息洪峰。
一、IoT队列和普通队列的差异点
1,上下行隔离拆分
在IoT场景中,我们把需要队列分为两个场景,一个是上行队列,一个是下行队列。 拆分之后,可以隔离上下行链路,控制一个设备,比如支付成功要下发打开柜子等,上行出任何问题,千万不能影响到下行业务。另外,上下行两条链路的特点差异非常大。设备上行消息,并发量非常高,但很多场景下对于可靠性和时延要求低,而设备下行消息,并发量则比较低,但下行消息(一般是控制设备指令)要求到达成功率很高。
2,支持设备级的海量topic
传统队列的核心诉求是,不论堆积多少不影响它的性能。kafka的topic一多,原本消息顺序写文件优势就会导致一个broker要退化到随机写,失去优势,另外要zookeeper来协调这么多topic也是有局限,所以这些队列本身有提供一个外挂代理桥接器对外入口是多个设备topic,再桥接映射到少量的实际kafka topic,这方案有一定可行性,但做不到隔离效果,治标不治本。
通过,图1和图2对比较明显,一个队列拥塞尽量减少对其它设备影响。我们需要的是“海量topic尽量相互隔离,并且不影响整体性能”,尽量做到设备A的消息堆积topic,不影响设备B。
3,实时生成消息优先发送
先举一个例子,一个快递柜业务的队列堆积,然后“此时此刻”在柜子旁边的用户死命的在旁边用手机点开柜子怎么也打不开(此时后端系统都恢复了),问题就是队列里面还有几十万条的消息,新来的消息需要排队, 等着之前的那些消息消费完,甭管这些消息还有没有用。 因此,实时生成消息优先发送,堆积的消息进入降级模式。
二、IoT消息队列诞生
1, IoT队列的设计思路
设计目标是为了打造一个支持上下行隔离、实时优先、及海量topic的队列网关,设计原则如下:
图片只是个片段,从这个模式可以看出来机制差别,大家都没有错,只是出发点不同。
3, 连接、计算、存储分离
broker不做连接,连接网关代理,broker只做流转分发,无状态+水平扩展;存储交给nosql DB,高吞吐写。
4, 消息策略-推拉结合
这个应该是队列的核心难点之一,和传统队列区分在于,我们考虑为平台化模式,独享资源过于昂贵。但带来问题是消费端不可控,所以使用结合模式,只有在消费者在线时会拉取堆积消息,而拉取是由AMQP队列网关来做,给到用户接口始终是推送过去的onMessage回调。
broker不是直接让consumer来连接,而是把队列网关剥离出来, 这样会更灵活,甚至对于部分用户我们的queue可以切换到ons、kafka等实现。kafka、rocketmq做法是在连接时会分配给客户端一个broker接入地址。
broker实时消息优先推送给consumer,失败才会落到queue ;这是一个完整事件,如果没有完成则不给producer commit。
异步ACK
5, 线性扩展-离线消息部分
实时部分消息采用推方式,基本上不会成为瓶颈,消费不过来消息进入堆积模式。由于底层依赖存储已经帮我们解决核心存储的扩展,剩下主要问题点在于如何消除写入热点和消费热点,这样broker可以完全做到无状态。
三,一个思考——如何解决海量topic问题?
首先面对“大量”的问题一般都是考虑分区,单元化,分组等隔离和拆分,这里海量topic我们讨论针对一个单实例模式下如何尽可能做到更多topic,完全任意数量都能100%没问题肯定是不现实的。
由于broker和存储已经隔离,broker和topic已经没有什么关系,或者说任何topic数据生成,broker做的事情就是写入和分发。
- 海量topic,每个topic有限数量订阅: topic和订阅者关系使用redis缓存或本地缓存,针对mqtt topic匹配有个topic tree的树算法,hivemq有实现版本。
- 单个topic 海量订阅: 这个场景其实是组播和广播,我们不会考虑在队列本身上面去做这个事情,而是在上层封装广播组件来协调任务和批量发送。
四, 阿里云AIoT消息队列
目前阿里云AIoT队列,也叫服务端订阅,意思就是用户用服务端订阅他们设备消息。为了降低接入成本,用户可以使用AMQP1.0协议接入,符合开源生态。 同时兼容传统队列和新队列,交给用户按场景来选择,用户即可选择使用kafka、mq,也可以选用iot队列,甚至组合模式,比如按消息特征规则来配置流转队列。 阿里云AIoT的场景队列实践,在现有mq队列、kafka队列融合之外,加了种自有的实时优先队列实现,同时,加入了队列网关代理,既能让用户选择普通消息队列,也可以选择轻便的IoT消息队列。