RocketMQ简单接入
RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。本文简单介绍下RocketMQ接入流程。
下面以服务号为例介绍RocketMQ接入接入
安装
pom.xml引入依赖
1
2
3
4
5
| <dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.0</version>
</dependency>
|
配置
通用配置
1
2
| rocketmq:
name-server: 10.18.23.41:9876;10.18.23.42:9876
|
生产者配置
1
2
3
4
| rocketmq:
producer:
group: weixin-producer-group-dev
send-message-timeout: 10000
|
消费者配置
以服务号为例，配置如下：
1
2
3
| wx:
rocketmq:
weixinTopic: weixinTopic-dev
|
使用
生产者开发示例
以服务号为例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| @Resource
@Value("${wx.rocketmq.weixinTopic}")
private String templateMsgTopic;
@Resource
private RocketMQTemplate rocketMQTemplate;
@PostConstruct
public void init (){
templateMsgTopic = templateMsgTopic + ":templateMsgTag";
}
@PostMapping(value = "/send")
@ApiOperation(value = "MQ消息发送", notes = "MQ异步消息发送")
public @ResponseBody
BaseResponse<WxMessageSendOut> send(@Valid @RequestBody WxMessageSendIn in) {
UUID uuid = Generators.randomBasedGenerator().generate();
Message message = MessageBuilder.withPayload(in).setHeader(RocketMQHeaders.KEYS, UUID.randomUUID()).build();
rocketMQTemplate.syncSend(templateMsgTopic,message);
WxMessageSendOut wxMessageSendOut = new WxMessageSendOut();
}
|
消费者开发示例
以服务号为例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| @Service
@Slf4j
@RocketMQMessageListener(topic = "${wx.rocketmq.weixinTopic}", consumerGroup = "templateMsg_consumer", selectorExpression = "templateMsgTag")
public class TemplateMsgListener implements RocketMQListener<WxMessageSendIn> {
@Resource
WxMessageService wxMessageService;
@Override
public void onMessage(WxMessageSendIn in) {
try {
wxMessageService.send(in);
} catch (Exception e) {
log.error("", e);
}
}
}
|
最佳实践
生产者
- 一个应用尽可能只使用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。
- 每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 Topic，key 来查询返条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。
- 消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段。
- 对于消息不可丢失应用，务必要有消息重发机制。例如：消息发送失败，存储到数据库，能有定时程序尝试重发或者人工触发重发。
消费者
- Consumer 数量要小于等于queue的总数量，由于Topic下的queue会被相对均匀的分配给Consumer，如果 Consumer 超过queue的数量，那多余的 Consumer 将没有queue可以消费消息。
- 消费过程要做到幂等（即消费端去重），RocketMQ为了保证性能并不支持严格的消息去重。
- 尽量使用批量方式消费，RocketMQ消费端采用pull方式拉取消息，通过consumeMessageBatchMaxSize参数可以增加单次拉取的消息数量，可以很大程度上提高消费吞吐量。另外，提高消费并行度也可以通过增加Consumer处理线程的方式，对应参数consumeThreadMin和consumeThreadMax。
- 消息发送成功或者失败，要打印消息日志。
更多的实践经验，请参考最佳实践