metaq最佳实践

1. 前言

本文档旨在描述RocketMQ使用过程当中的一些最佳实践，建议用户这样作，可是非必须。

2. Producer最佳实践

2.1 发送消息注意事项

1. 一个应用尽量用一个Topic，消息子类型用tags来标识，tags能够由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才能够利用tags在broker作消息过滤。message.setTags(“TagA”);

2. 每一个消息在业务层面的惟一标识码，要设置到keys字段，方便未来定位消息丢失问题。服务器会为每一个消息建立索引（哈希索引），应用能够经过topic，key来查询这条消息内容，以及消息被谁消费。因为是哈希索引，请务必保证key尽量惟一，这样能够避免潜在的哈希冲突。
   // 订单Id
   String orderId = “20034568923546”;
   message.setKeys(orderId);

3. 消息发送成功或者失败，要打印消息日志，务必要打印sendresult和key字段。

4. send消息方法，只要不抛异常，就表明发送成功。可是发送成功会有多个状态，在sendResult里定义。

• SEND_OK
  消息发送成功
• FLUSH_DISK_TIMEOUT
  消息发送成功，可是服务器刷盘超时， 消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
• FLUSH_SLAVE_TIMEOUT
  消息发送成功，可是服务器同步到Slave时超时， 消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
• SLAVE_NOT_AVAILABLE
  消息发送成功，可是此时slave不可用， 消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失

对于精卫发送顺序消息的应用，因为顺序消息的局限性，可能会涉及到主备自动切换问题，因此若是sendresult中的status字段不等于SEND_OK，就应该尝试重试。对于其余应用，则没有必要这样。

5. 对于消息不可丢失应用，务必要有消息重发机制
   例如若是消息发送失败，存储到数据库，能有定时程序尝试重发，或者人工触发重发。

2.2消息发送失败如何处理

Producer的send方法自己支持内部重试，重试逻辑以下：

1. 至多重试3次。
2. 若是发送失败，则轮转到下一个Broker。
3. 这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s。因此，若是自己向broker发送消息产生超时异常，就不会再作重试。

以上策略仍然不能保证消息必定发送成功，为保证消息必定成功，建议应用这样作若是调用send同步方法发送失败， 则尝试将消息存储到db，由后台线程定时重试，保证消息必定到达Broker。

上述db重试方式为何没有集成到MQ客户端内部作，而是要求应用本身去完成，咱们基于如下几点考虑

1. MQ的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是cpu、内存、网络。
2. 若是MQ客户端内部集成一个KV存储模块，那么数据只有同步落盘才能较可靠，而同步落盘自己性能开销较大，因此一般会采用异步落盘，又因为应用关闭过程不受MQ运维人员控制，可能常常会发生kill -9这样暴力方式关闭，形成数据没有及时落盘而丢失。
3. Producer所在机器的可靠性较低，通常为虚拟机，不适合存储重要数据。

综上，建议重试过程交由应用来控制。

2.3选择oneway形式发送

一个RPC调用，一般是这样一个过程

1. 客户端发送请求到服务器
2. 服务器处理该请求
3. 服务器向客户端返回应答因此一个RPC的耗时时间是上述三个步骤的总和，而某些场景要求耗时很是短，可是对可靠性要求并不高，例如日志收集类应用，此类应用能够采用oneway形式调用，oneway形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个os系统调用的开销，即将数据写入客户端的socket缓冲区，此过程耗时一般在微秒级。

3 Consumer最佳实践

3.1消费过程要作到幂等（即消费端去重）

如《RocketMQ 原理简介》中所述，RocketMQ没法避免消息重复，因此若是业务对消费重复很是敏感，务必要在业务层面去重，有如下几种去重方式

1. 将消息的惟一键，能够是msgId，也能够是消息内容中的惟一标识字段，例如订单Id等，消费以前判断是否在Db或Tair(全局KV存储)中存在，若是不存在则插入，并消费，不然跳过。（实际过程要考虑原子性问题，判断是否存在能够尝试插入，若是报主键冲突，则插入失败，直接跳过）
   msgId必定是全局惟一标识符，可是可能会存在一样的消息有两个不一样msgId的状况（有多种缘由），这种状况可能会使业务上重复消费，建议最好使用消息内容中的惟一标识字段去重。
2. 使用业务层面的状态机去重

3.3消费速度慢处理方式

3.3.1提升消费并行度

绝大部分消息消费行为属于IO密集型，便可能是操做数据库，或者调用RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，经过增长消费并行度，能够提升总的消费吞吐量，可是并行度增长到必定程度，反而会降低，如图所示，呈现抛物线形式。因此应用必需要设置合理的并行度。CPU密集型应用除外。

3.3.2批量方式消费

某些业务流程若是支持批量方式消费，则能够很大程度上提升消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时1秒钟，一次处理10个订单可能也只耗时2秒钟，这样便可大幅度提升消费的吞吐量，经过设置consumer的consumeMessageBatchMaxSize这个参数，默认是1，即一次只消费一条消息，例如设置为N，那么每次消费的消息数小于等于N。

3.3.3跳过非重要消息

发生消息堆积时，若是消费速度一直追不上发送速度，能够选择丢弃不重要的消息
如何判断消费发生了堆积？

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { longoffset = msgs.get(0).getQueueOffset(); String maxOffset = msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET); longdiff = Long.parseLong(maxOffset) - offset; if (diff > 100000) { // TODO消息堆积状况的特殊处理 returnConsumeConcurrentlyStatus.CONSUME_SUCCESS; } // TODO正常消费过程 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { longoffset = msgs.get(0).getQueueOffset(); String maxOffset = msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET); longdiff = Long.parseLong(maxOffset) - offset; if (diff > 100000) { // TODO消息堆积状况的特殊处理 returnConsumeConcurrentlyStatus.CONSUME_SUCCESS; } // TODO正常消费过程 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }

如以上代码所示，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或所有消息，这样就能够快速追上发送消息的速度。

3.3.4优化每条消息消费过程

举例以下，某条消息的消费过程以下
1.根据消息从DB查询数据1
2.根据消息从DB查询数据2
3.复杂的业务计算
4.向DB插入数据3
5.向DB插入数据4
这条消息的消费过程与DB交互了4次，若是按照每次5ms计算，那么总共耗时20ms，假设业务计算耗时5ms，那么总过耗时25ms，若是能把4次DB交互优化为2次，那么总耗时就能够优化到15ms，也就是说整体性能提升了40%。

对于Mysql等DB，若是部署在磁盘，那么与DB进行交互，若是数据没有命中cache，每次交互的RT会直线上升，若是采用SSD，则RT上升趋势要明显好于磁盘。个别应用可能会遇到这种状况：在线下压测消费过程当中，db表现很是好，每次RT都很短，可是上线运行一段时间，RT就会变长，消费吞吐量直线降低。

主要缘由是线下压测时间太短，线上运行一段时间后，cache命中率降低，那么RT就会增长。建议在线下压测时，要测试足够长时间，尽量模拟线上环境，压测过程当中，数据的分布也很重要，数据不一样，可能cache的命中率也会彻底不一样。3.4消费打印日志若是消息量较少，建议在消费入口方法打印消息，方便后续排查问题。

3.4 消费打印日志

若是消息量较少，建议在消费入口方法打印消息，方便后续排查问题。

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { log.info("RECEIVE_MSG_BEGIN: " + msgs.toString()); // TODO正常消费过程 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} } public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { log.info("RECEIVE_MSG_BEGIN: " + msgs.toString()); // TODO正常消费过程 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} }

若是能打印每条消息消费耗时，那么在排查消费慢等线上问题时，会更方便。