M02.MQ之腾讯开源消息中间件TubeMQ

随着分布式技术的发展，MQ技术产品也出现井喷。目前除了各类常用的MQ，比如Apache的ActiveMQ，Kafka，Pulsar，RocketMQ（既是Apache，也是阿里的，头条也是基于RocketMQ），以及RabbitMQ（美团、汽车之家大量使用）外，各大厂商都自研了自己的产品，腾讯的CMQ和TubeMQ，京东的JMQ，去哪儿的QMQ，滴滴的DDMQ（基于RocketMQ），其中不少都开源了。这里说一下今年开源的TubeMQ。

腾讯开源的TubeMQ

官方介绍如下：
https://github.com/Tencent/TubeMQ/blob/master/docs/tubemq_basic_introduction_cn.md

TubeMQ是腾讯大数据在2013年开始研发的分布式消息中间件系统（MQ），专注服务大数据场景下海量数据的高性能存储和传输。经过近7年上万亿的海量数据沉淀，较之于众多的开源MQ组件，TubeMQ在海量实践（稳定性+性能）和低成本方面有一定的优势，近期我们在开源TubeMQ的相关代码及设计，更多资料正在陆续整理。

TubeMQ集群架构：

经过多年演变，TubeMQ集群分为如下5个部分：

• Portal： 负责对外交互和运维操作的Portal部分，包括API和Web两块，API对接集群之外的管理系统，Web是在API基础上对日常运维功能做的页面封装；
• Master： 负责集群控制的Control部分，该部分由1个或多个Master节点组成，Master HA通过Master节点间心跳保活、实时热备切换完成（这是大家使用TubeMQ的Lib时需要填写对应集群所有Master节点地址的原因），主Master负责管理整个集群的状态、资源调度、权限检查、元数据查询等；
• Broker： 负责实际数据存储的Store部分，该部分由相互之间独立的Broker节点组成，每个Broker节点对本节点内的Topic集合进行管理，包括Topic的增、删、改、查，Topic内的消息存储、消费、老化、分区扩容、数据消费的offset记录等，集群对外能力，包括Topic数目、吞吐量、容量等，通过水平扩展Broker节点来完成；
• Client： 负责数据生产和消费的Client部分，该部分我们以Lib形式对外提供，大家用得最多的是消费端，相比之前，消费端现支持Push、Pull两种数据拉取模式，数据消费行为支持顺序和过滤消费两种。对于Pull消费模式，支持业务通过客户端重置精确offset以支持业务extractly-once消费，同时，消费端新推出跨集群切换免重启的BidConsumer客户端；
• Zookeeper： 负责offset存储的zk部分，该部分功能已弱化到仅做offset的持久化存储，考虑到接下来的多节点副本功能该模块暂时保留。

比较常规的分布式MQ结构，broker功能比较重。

相比Kafka,TubeMQ的系统特点：

1. 纯Java实现语言： TubeMQ采用纯Java语言开发，便于开发人员快速熟悉项目及问题处理；
2. 引入Master协调节点： 相比Kafka依赖于Zookeeper完成元数据的管理和实现HA保障不同，TubeMQ系统采用的是自管理的元数据仲裁机制方式进行，Master节点通过采用内嵌数据库BDB完成集群内元数据的存储、更新以及HA热切功能，负责TubeMQ集群的运行管控和配置管理操作，对外提供接口等；通过Master节点，TubeMQ集群里的Broker配置设置、变更及查询实现了完整的自动化闭环管理，减轻了系统维护的复杂度；
3. 服务器侧消费负载均衡： TubeMQ采用的是服务侧负载均衡的方案，而不是客户端侧操作，提升系统的管控能力同时简化客户端实现，更便于均衡算法升级；
4. 系统行级锁操作： 对于Broker消息读写中存在中间状态的并发操作采用行级锁，避免重复问题；
5. Offset管理调整： Offset由各个Broker独自管理，ZK只作数据持久化存储用（最初考虑完全去掉ZK依赖，考虑到后续的功能扩展就暂时保留）；
6. 消息读取机制的改进： 相比于Kafka的顺序块读,TubeMQ采用的是消息随机读取模式,同时为了降低消息时延又增加了内存缓存读写,对于带SSD设备的机器,增加消息滞后转SSD消费的处理，解决消费严重滞后时吞吐量下降以及SSD磁盘容量小、刷盘次数有限的问题，使其满足业务快速生产消费的需求（后面章节详细介绍）；
7. 消费者行为管控： 支持通过策略实时动态地控制系统接入的消费者行为，包括系统负载高时对特定业务的限流、暂停消费，动态调整数据拉取的频率等；
8. 服务分级管控： 针对系统运维、业务特点、机器负载状态的不同需求，系统支持运维通过策略来动态控制不同消费者的消费行为，比如是否有权限消费、消费时延分级保证、消费限流控制，以及数据拉取频率控制等；
9. 系统安全管控： 根据业务不同的数据服务需要，以及系统运维安全的考虑，TubeMQ系统增加了TLS传输层加密管道，生产和消费服务的认证、授权，以及针对分布式访问控制的访问令牌管理，满足业务和系统运维在系统安全方面的需求；
10. 资源利用率提升改进： 相比于Kafka，TubeMQ采用连接复用模式，减少连接资源消耗；通过逻辑分区构造，减少系统对文件句柄数的占用，通过服务器端过滤模式，减少网络带宽资源使用率；通过剥离对Zookeeper的使用，减少Zookeeper的强依赖及瓶颈限制；
11. 客户端改进： 基于业务使用上的便利性以，我们简化了客户端逻辑，使其做到最小的功能集合，我们采用基于响应消息的接收质量统计算法来自动剔出坏的Broker节点，基于首次使用时作连接尝试来避免大数据量发送时发送受阻（具体内容见后面章节介绍）。

这一块基本上说清楚了特点，以及与其他MQ的一些特色的地方，其实可以猜到，一直在和kafka做对比，很多地方参与并改进了kafka，在管理能力上做了不少思考和新的实现。

TubeMQ客户端的演进：

业务与TubeMQ接触得最多的是消费侧，怎样更适应业务特点、更方便业务使用我们在这块做了比较多的改进：
数据拉取模式支持Push、Pull：

• Push客户端： TubeMQ最初消费端版本只提供Push模式的消费，这种模式能比较快速地消费数据，减轻服务端压力，但同时也带来一个问题，业务使用的时候因为无法控制拉取频率，从而容易形成数据积压数据处理不过来；
  – 带消费中止/继续的Push客户端： 在收到业务反馈能否控制Push拉取动作的需求后，我们增加了resumeConsume()/pauseConsume()函数对，让业务可以模拟水位线控制机制，状态比较繁忙时调用pauseConsume()函数来中止Lib后台的数据拉取，在状态恢复后，再调用resumeConsume()通知Lib后台继续拉取数据；
  – Pull客户端： 我们后来版本里增加了Pull客户端，该客户端有别于 – Push客户端，是由业务而非Lib主动的拉取消息并对数据处理的结果进行成功与否的确认，将数据处理的主动权留给业务。这样处理后，虽然服务端压力有所提升，但业务消费时积压情况可大大缓解。
• 数据消费行为支持顺序和过滤消费： 在TubeMQ设计初我们考虑是不同业务使用不同的Topic，实际运营中我们发现不少业务实际上是通过代理模式上报的数据，数据通过Topic下的文件ID或者表ID属性来区分，业务为了消费自己的一份数据是需要全量消费该Topic下的所有数据。我们通过tid字段支持指定属性的过滤消费模式，将数据过滤放到服务端来做，减少出流量以及客户端的数据处理压力。
• 支持业务extractly-once消费： 为了解决业务处理数据时需要精确回档的需求，在客户端版本里提供了通过客户端重置精确offset功能，业务重启系统时，只需通过客户端提供待回拨时间点的消费上下文，TubeMQ即可按照指定的精确位置接续消费。该特性目前已在Flink这类实时计算框架使用，依托Flink基于checkpoint机制进行extractly-once数据处理。

推和拉 是消息处理的两个最基础模式。推对服务器处理来说更简单，推出去就不管了，broker变轻，但是可能单位时间推太多，导致消费端积压，压垮了client端系统。拉则意味着，你随时来拿数据，broker都要保持状态而且会产生积压，还需要处理重试策略等。有了offset则意味着可以随时回溯消息，但是这样可能会导致重复，如果没有内置的去重其实不是extractly once，而是atleast once，消息会重复。

其他几个mq

滴滴的DDMQ：
https://github.com/didi/DDMQ/blob/master/README_CN.md
​
去哪儿网的QMQ：
https://github.com/qunarcorp/qmq

有意思的几个点

TubeMQ跟 kafka，rocketmq，pulsar等主流的MQ架构上有什么差别？
官方给出的意见是：
Kafka按照顺序写 + 顺序块读的模式实现，单实例下性能数据很强，但随着实例数增多，它的性能就呈现不稳定下降状态；TubeMQ采用顺序写 + 随机读的模式，即使在最大限制下系统仍可以做到长期稳定的1G以上的入流量，同时，结合服务端过滤过滤消费非常顺畅。
个人对这个持保留意见，大量创建topic不适合kafka的设计原则（一般我们建议单集群的topic数量在100以内，过多的小topic造成随机读写，但是可以合并，然后区分和路由消息即可），同时如果改成SSD盘也可以提升吞吐和延迟，几千个topic问题不大。而且kafka的延迟也不像上面的文档里对比说的250ms，我们实际使用大概在10-40ms之间。

TubeMQ看了一下，整体设计跟pulsar有点像，主要是broker和storage做了分离；消息处理模式上跟ActiveMQ到底有些许接近。

几个有意思的地方：
1、TubeMQ不支持多副本，这样的话单机有可能还是在极端情况下丢失数据，但多副本是目前的各种分布式消息队列的标配（看了一下腾讯云上的商业版本CMQ是支持的。）
2、服务器侧消费负载均衡，早期版本的kafka是这样的，问题挺多
3、消息随机读，这样需要加内存缓存和依赖SSD，挺诡异，为了并发又加了锁，这一块很复杂，ActiveMQ就是因为内存的处理太复杂，导致量一大，谁都用不好
4、同时支持推和拉，这一点也挺有意思，跟第一条一条有关系，要是支持推的话，服务端肯定需要有状态
5、支持服务器端的消息过滤，现在一般的MQ都是客户端过滤，也同理。

MQ发现到现在，一共经历了三代，分别以ActiveMQ，Kafka/RocketMQ，Pulsar为代表，从趋势上来看，越来越分布式、趋向对云原生的支持，越来越无状态，broker越来越轻薄。
总之这个方案看起来是综合了传统和现在的各个MQ的一些特点，但是实现的很重。

还有个tip，TubeMQ里的组件名称有点乱，叫master的东西，实际上是broker，叫broker的东西，实际上是storage（在pulsar里是bookie）。
：）