RocketMQ(4.6.1) 系列教程 -- Producer 篇

RocketMQ 消息发送

三种发送方式

• 同步

生产者发送消息，须要等到消息服务器返回结果

• 异步

生产者发送消息， 不须要等到消息服务器返回结果，生产者线程不阻塞，只需注册监听回调函数便可。

• 单向

生产者只管发，无论成功与否。

消息结构

org.apache.rocketmq.common.message.Message

private String topic;
private int flag;
// 属性
private Map<String, String> properties;
// 消息体
private byte[] body;
// 事务 ID
private String transactionId;

properties 存放的扩展属性主要有：

• tags

用于过滤消息

• keys

Message 索引键，多个用空格隔开，RocketMQ 可根据这些 key 快速检索消息

• waitStoreMsgOk

消息发送时，是否等到消息存储完，再返回

• delayTimeLevel

消息延迟级别，用于定时消息或消息重试

生产者

DefaultMQProducer 核心属性

// 生成者组，消息服务器在回查事务状态时，会随机选择该组中的任何一个生成者发起事务回查请求
private String producerGroup;

// 默认的 topic key
private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC; // TBW102

// 默认的主题队列数量
private volatile int defaultTopicQueueNums = 4;

// 默认发送超时时间
private int sendMsgTimeout = 3000;

// 消息体超过 4kb 时，压缩消息
private int compressMsgBodyOverHowmuch = 1024 * 4;

// 同步发送失败，最大尝试次数
private int retryTimesWhenSendFailed = 2;

// 异步发送失败，最大尝试次数
private int retryTimesWhenSendAsyncFailed = 2;

// 消息重试时选择另一个 broker 时，是否不等待存储结果就返回。默认 false
private boolean retryAnotherBrokerWhenNotStoreOK = false;

// 容许发送的最大消息长度
private int maxMessageSize = 1024 * 1024 * 4; // 4M

生产者启动

流程代码位于：DefaultMQProducerImpl#start

1. 检查 productGroup，默认为 CLIENT_INNER_PRODUCER。
2. 将 instanceName 修改成 PID。即 clientId 格式为 ip@pid
3. 建立 MQClientInstance 实例，给交给 MQClientManager 管理
4. 建立默认的 Topic（TBW102），并放到 topicPublishInfoTable 中。

消息发送流程

流程代码位于：DefaultMQProducerImpl#sendDefaultImpl

验证消息

查找路由

1. Producer 优先从缓存中查找，存在而且正常直接返回
2. 若是没有，则向 namesrv 发起请求， namesrv 返回 TopicRouteData。
3. 判断返回的 TopicRouteData 与旧的是否被改变过；若是是，则TopicPublishInfo 更新 queueDatas，brokerDatas 信息。

重要的类：TopicPublishInfo

// 是不是顺序消息
private boolean orderTopic = false;
private boolean haveTopicRouterInfo = false;

// 主题的消息队列
private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
// 用于选择消息队列。每次选择一个队列，会自增1，若是等于 Integer.MAX_VALUE，重置为0
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
private TopicRouteData topicRouteData;

// TopicRouteData
private String orderTopicConf;
private List<QueueData> queueDatas;
private List<BrokerData> brokerDatas;
private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

代码位于：DefaultMQProducerImpl#tryToFindTopicPublishInfo(final String topic)

选择消息队列

不启用故障延迟机制

sendLatencyFaultEnable = false 默认不启用 broker 故障延迟机制。

RocketMQ 会避开上一次发送失败的 broker

代码入口：TopicPublishInfo#selectOneMessageQueue(final String lastBrokerName)

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
  if (lastBrokerName == null) {
    return selectOneMessageQueue();
  } else {
    int index = this.sendWhichQueue.getAndIncrement();
    for (int i = 0; i < this.messageQueueList.size(); i++) {
      int pos = Math.abs(index++) % this.messageQueueList.size();
      if (pos < 0)
        pos = 0;
      MessageQueue mq = this.messageQueueList.get(pos);
      // 避开上一次发送失败的 broker
      if (!mq.getBrokerName().equals(lastBrokerName)) {
        return mq;
      }
    }
    return selectOneMessageQueue();
  }
}

启用故障延迟机制

代码入口：MQFaultStrategy#selectOneMessageQueue

消息发送

代码入口：DefaultMQProducerImpl#sendKernelImpl

1. 找到所要发送的 broker 地址
2. 为消息分配全局惟一ID，消息体超过4kb，则采用ZIP压缩。
3. 若是注册了<font color="red">消息发送钩子函数</font>，则执行消息发送以前的加强逻辑。经过 <font color="red">DefaultMQProducerImpl#registerSendMessageHook() </font>注册钩子处理类，且能够注册多个
4. 构造消息发送请求包
5. 根据消息发送方式，进行发送（同步、异步、单向）
6. 若是注册了消息发送钩子函数，执行after 逻辑

若是发送失败，则继续重复 选择消息队列，而后发送

同步发送

1. 检查消息发送是否合理
2. 消息重试次数超过最大重试次数，消息将进入 DLD 延迟队列。主题为：%DLD%
3. 消息存储

异步方式

重试的调用入口是在收到服务端响应包时进行的。所以网络异常、网络超时，将不会触发重试。

单向发送

异步发送，不须要回调，没有重试机制

上篇疑问解答

Q: 路由发现与删除机制存在一个问题，Namesrv 须要等 Broker 失效 至少 120s 才能将 Broker 从路由表中删除。那么若是，Producer 获取到的 broker 是已经宕机的信息，那么是否会形成消息发送失败

单机环境确定没救了，集群环境下没有影响的。

rocketMQ 每次发送消息时，会选择消息队列。第一次发送失败后，会进行失败的尝试。此时会将上一次发送失败的 broker 排除。从新选择一个 broker。

Q: namesrv 之间数据不共享，那么会形成消息的发送失败吗？

若是是当前保持链接的 namesrv 保存的 broker 信息，都挂掉了，那么必定是会失败的。（若是是这个缘由，那么我以为，全部的 broker 应该挂掉了）

若是是此 namesrv 没有 topic 对应的 broker ，那么 RocketMQ 会选择其余的 namesrv 保持链接，因此，不会发送失败。

相关代码：NettyRemotingClient#getAndCreateNameserverChannel()

若是只是 namesrv 中的某个 broker 挂掉了，可是由于 Producer 有消息重试机制，会选择其余的 broker。所以，消息不会发送失败。除非全部的 broker 都失败了。

结论：除非全部 broker 挂掉，否则我以为不会形成消息发送失败。

Q: 集群中的 namesrv 挂掉了一个，是否会形成链接它的 broker 发送消息失败？

答案是不会。

缘由是，每次再向 namesrv 发送消息时，须要判断与 namesrv 的 channel 是否有效。若是无效，则会尝试从剩下的 namesrv 查找一个有效的，并与之保持链接。

相关代码位于： NettyRemotingClient#getAndCreateNameserverChannel()