概述
ApacheRocketMQ是一个低延时、高性能、可靠、海量并且灵活扩展性的分布式消息和流平台,于2017年9月25日成为Apache基金会顶级开源项目。它由4个部分组成:name servers、brokers、producers、consumers。每个部分都能水平扩展防止单点故障。
NameServer Cluster
Name servers提供轻量级的服务发现和路由功能。每个Name Server记录完整的路由信息,提供一致的读写服务并且支持快速的容量扩展。
Broker Cluster
Brokers关注消息存储通过轻量的TOPIC和QUEUE机制。它们支持Push和Pull模式,包含容错机制(2 copies or 3 copies),并且提供strong padding of peaks and capacity of accumulating hundreds of billion messages in their original time order.另外,Brokers提供灾难恢复、丰富的指标统计和警告机制,所有这些是传统的消息系统缺失的。
Producter Cluster
Producters支持分布式部署。分布式Producters通过多种负载均衡机制发送消息到Broker Cluster。发送进程支持快速失败并且低延时。
Consumer Cluster
Consumers支持Push和Pull两种分布式部署模式。也支持集群消费和消息广播。它提供实时消息订阅机制并且满足大多数消费场景。RocketMQ的网站为感兴趣的读者提供1个快速开始指南。
NameServer
NameServer是一个全功能的服务器,包括有2个功能:
- Broker管理:NameServer接受Broker cluster的注册,并且提供心跳机制检查broker是否alive
- 路由管理:每个NameServer拥有broker cluster的完整路由信息和供客户端查询的队列信息
据我们所知,RocketMQ客户端(Producer/Consumer)将会从NameServer查询队列路由信息,但是客户端是怎么发现NameServer的地址的呢?
有四种方式提供NameServer地址列表给客户端:
- 编程方式,例如
producer.setNamesrvAddr("ip:port")
- Java Options,使用
rocketmq.namesrv.addr
- 环境变量,使用
NAMESRV_ADDR
- HTTP Endpoint
Broker Server
Broker Server是消息存储、传递、消息查询、高可用等的可靠保证。
下图显示的是Broker Server几个重要的子模块
- Remoting Module,broker的入口,处理客户端请求
- Client Manager,管理客户端(Producer/Consumer)并且维护Consumer的topic订阅
- Store Service,提供简单的API在物理磁盘存储或查询消息
- HA Service,提供master broker和slave broker之间的数据同步功能
- Index Service,通过制定的key构建消息索引和消息快速查找
其他功能
- 异步发送不处理结果,调用SendOneway方法即可,性能高,即使发失败也不会抛异常。但这种情况下,发送方对消息是否发成功全然不知,适用于量大但允许丢消息的场景。
- 异步发送,异步处理结果,producer的Send方法允许传入一个回调接口,此时,Send方法不阻塞直接返回,消息发送成功或失败时,会触发传入接口中的方法
- 广播消费,如果希望消息在同一个group的每台机器上都消费一次,可以使用广播消费。
- Tag过滤,被滤掉的消息会直接被这个consumer的group丢弃,不会再通过网络发送
- 拉模式
- 延时消息
最佳实践
Producer Group
- 同一个group的生产者一般尽量只创建一个,每次发消息时重复使用。
- 使用日志记录消息ID,便于出问题时排查。每条消息都有一个唯一的ID,消息发成功时,返回的结束里能拿到发成功的消息的ID。消费消息时,也能拿到此消息的ID。一般比较好的做法是在日志里把这个消息ID打出来,便于今后追踪问题。使用这个ID,还可以到后台查出消息内容。
- 注意消费方法是并发执行的。用户手册中有说明,消费时请留意线程安全问题。另外,一般没必要在消费方法里另外开个线程去处理消息,调整消费线程池的大小在大多数情况下就能达到目的。
- 用消息队列本身的重试机制。消息队列本身对消息的可靠消费做了一定的保证。如果消费时抛了异常,或返回了失败,消息会进入一个重试队列,定期重试消费,重试的间隔会逐次延长(1s、5s、10s、30s……最后一直是2小时)。如果你的消费方法里要做插数据库、调其它系统的接口等可能失败的操作,但又要保证消息最终要消费成功,可以利用这个特性,但要注意重试是会延时的,要留意这个延时对业务的影响。
- 如果对重复消费的情况零容忍,则一定要做幂等处理。消息系统保证消息可靠消费,但相应的,就不能保证消息不重复。大多数情况下一条消息在一个消费者组里只消费一次。但在网络抖动、消费者挂掉等异常情况下,可能会有少量的重复消费。如果重复消费会导致业务上不能容忍的错误(比如重复下单、重复扣款之类的),就一定要做去重处理。去重的方法根据业务逻辑可能各不相同,这里不能给出一个统一的方法。
- 注意发消息是可能失败抛异常的。网络故障、消息服务挂掉的情况下都会抛异常(虽然挂掉的机率是非常低的)。这时消息没发出去,也不会再重试了。如果业务对此零容忍,则需要做处理。
- 善用后台排查问题。后台可以看很多东西:消费者的在线情况、消息的消费进度、消息内容等等。很多调试问题都可以借助后台来排查