概念

1、消息发送方式

Rocketmq提供三种方式可以发送普通消息：同步、异步、和单向发送。

• 同步：发送方发送消息后，收到服务端响应后才发送下一条消息
• 异步：发送一条消息后，不等服务端返回就可以继续发送消息或者后续任务处理。发送方通过回调接口接收服务端响应，并处理响应结果。
• OneWay：发送方发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不需要应答。

发送方式对比：发送吞吐量,单向>异步>同步。但单向发送可靠性差存在丢失消息可能，选型根据实际需求确定。

2、消息类型

消息客户端提供多种SDK： 普通、顺序、事务、延时消息

• 普通消息：MQ生产者客户端对象是线程安全的，可以在多线程之间共享使用。同时也可以用多线程并发发送消息可以增加消息TPS，一般项目中创建一个Peoducer实例就好。

 DefaultMQProducer producer = new DefaultMQProducer("arch-rocketmq");
 producer.setNamesrvAddr("localhost:9876");
 producer.start();   
 try {
     Message msg = new Message("mq-test4","*", ("Hello RocketMQ").getBytes("UTF-8"));
     SendResult sendResult = producer.send(msg);
     System.out.printf("%s%n", sendResult);
 }  

• 顺序消息：提供一种严格按照顺序来发送和消费的消息类型。指定一个topic后，所有消息Message根据sharding key进行分区，相同key的消息在同一个分区内完全按照FIFO进行发送和消费。

MQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.start();
for (int i = 0; i < 100; i++) {
    int orderId = i % 10;
    Message msg = new Message("Topic-test", "*", "KEY" + i,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
           Integer id = (Integer) arg;
          int index = id % mqs.size();
          return mqs.get(index);
       }
    }, orderId);
    System.out.printf("%s%n", sendResult);
}

• 延时消息：用于指定消息发送到消息队列后，延时一段时间才会被客户端进行消费，使用于任务延时场景。开源项目支持18个延迟级别 delayTimeLevel: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。

 DefaultMQProducer producer = new DefaultMQProducer("arch-rocketmq");
 producer.setNamesrvAddr("localhost:9876");
 producer.start();   
 try {
     Message msg = new Message("mq-test4","*", ("Hello RocketMQ").getBytes("UTF-8"));
     msg.setDelayTimeLevel(3);  // 延迟10s
     SendResult sendResult = producer.send(msg);
     System.out.printf("%s%n", sendResult);
 } catch (Exception e) { }

• 事务消息：RocketMQ提供类似XA分布式事务能力(XA则是一种分布式事务协议，包含事务管理器和本地资源两部分)，主要解决了消息发送和数据库事务带来不一致问题。整个事务消息交互如下图

image.jpeg

1. 应用程序在事务内完成落库，同步调用RocketMQ消息发送，发送状态为Prepare并设置事件监听。
2. RocketMQ侧收到消息后，将消息存储在RMQ_SYS_Trans_Half_topic消息消费队列，这样消费不会被立即消费到。
3. RocketMQ则开启定时任务，消息RMQ_SYS_Trans_Half_topic向发送端发起消息事务状态回查，应用程序根据事务状态回馈服务器(提交、回滚)如果提交或回滚，消息服务器则提交或回滚消息。RocketMQ允许设置消息回查间隔和回查次数。超过则回滚消息。

 // TransactionListenerImpl需要实现TransactionListener 
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = new TransactionMQProducer("my_group_name");
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            public Thread newThread(Runnable r) { new Thread(r);}
});
producer.setExecutorService(executorService);
producer.setTransactionListener(transactionListener);
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 10; i++) {
     try {
           Message msg = new Message("TopicTest1234", tags[i % tags.length], "KEY" + i, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
          SendResult sendResult = producer.sendMessageInTransaction(msg, null);
          System.out.printf("%s%n", sendResult);
  } catch (MQClientException | UnsupportedEncodingException e) {
          e.printStackTrace();
  }
}

3、消息数据结构

Message配置

客户端公共配置

Producer配置

消息发送原理解刨

1、 发送主要包括三个核心流程：

• 获取topic路由信息TopicPublishInfo（路由信息在启动时就已经加载过，在系列一种已经讲解过）
• 根据topic的路由信息选择一个MessageQueue(明确往哪个broker发送)
• 发送消息，成功则返回，失败则更新规避策略，同时进行重试发送默认重试次数是3次通过times进行计数。如果发生消息重试会导致消息发送重复，例如发送消息实际上Broker接收到消息，但客户端接受结果超时会重发消息，所有消费端需要保证幂等性。

private SendResult sendDefaultImpl( Message msg,final CommunicationMode communicationMode,final SendCallback sendCallback,final long timeout){
       TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
       if (topicPublishInfo != null && topicPublishInfo.ok()) {
           int times = 0;
           for (; times < timesTotal; times++) {
               String lastBrokerName = null == mq ? null : mq.getBrokerName();
               MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
               if (mqSelected != null) {
                   mq = mqSelected;
                   brokersSent[times] = mq.getBrokerName();
                   try {
                       beginTimestampPrev = System.currentTimeMillis();
                       if (times > 0) {msg.setTopic(this.defaultMQProducer.withNamespace(msg.getTopic()));}
                       long costTime = beginTimestampPrev - beginTimestampFirst;
                       if (timeout < costTime) {callTimeout = true;break;}
                       sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
                       endTimestamp = System.currentTimeMillis();
                       this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
                       switch (communicationMode) {
                           case ASYNC: return null;
                           case ONEWAY: return null;
                           case SYNC:
                               if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                                   if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                                       continue;}}
                               return sendResult;
                           default: break;
                       }
                   } catch (Exception e) {
                       endTimestamp = System.currentTimeMillis();
                       this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);                       
                       exception = e;
                       continue;
                   }  
           if (sendResult != null) {
               return sendResult;
           }
}

2、 获取topic路由信息

2.1. 一个topic分布在多个Broker，一个Broker包含多个Queue(brokerName、读队列个数、写队列个数、权限、同步或异步)；从缓存中获取topic路由信息；没有则从namesrv获取；没有使用默认topic获取路由配置信息。

2.2. 从NameServer获取配置信息，使用ReentrantLock，设置超时3s；基于Netty从namesrv获取配置信息；然后更新topic本地缓存，需要同步更新发送者和消费者的topic缓存

ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
    this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
    this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
    topicPublishInfo = this.topicPublishInfoTable.get(topic);
}

3、 根据topic负载均衡算法选择一个MessageQueue

1. 是否开启消息失败延迟规避机制

2. 本地变量ThreadLocal 保存上一次发送的消息队列下标，消息发送使用轮询机制获取下一个发送消息队列。同时topic发送有异常延迟，确保选中的消息队列所在broker正常

3. 当前消息队列是否可用

发送消息延迟机制；MQFaultStrategy（latencyMax最大延迟时间 end-start为消息延迟时间，如果失败 则将这个broker的isolation为true，同时这个broker在5分钟内不提供服务等待回复）

if (this.sendLatencyFaultEnable) {
    try {
       int index = tpInfo.getSendWhichQueue().getAndIncrement();
       for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
           int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
           if (pos < 0)  pos = 0;
           MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
           if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {   // 判断broker是否被规避
               if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
                   return mq;
           }
       }
       final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();  //选择
       int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
       if (writeQueueNums > 0) {
       final MessageQueue mq = tpInfo.selectOneMessageQueue();
       if (notBestBroker != null) {
           mq.setBrokerName(notBestBroker);
           mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
       }
       return mq;  
   }          
}
return tpInfo.selectOneMessageQueue(lastBrokerName);

4、故障延迟机制FaultItem

开启故障延迟则会构造FaultItem记录，在某一时刻前都当做故障(brokeName、发送消息异常时间点、这个时间点都为故障)

4.1.首先选择一个broker==lastBrokerName并且可用的一个队列（也就是该队列并没有因为延迟过长而被加进了延迟容错对象latencyFaultTolerance 中）

4.2.如果第一步中没有找到合适的队列，此时舍弃broker==lastBrokerName这个条件，选择一个相对较好的broker来发送

4.3. 选择一个队列来发送，一般都是取模方式来获取

也就是当Producer发送消息时间过长，逻辑上N秒内Broker不可用。例如发送时间超过15000ms，broker则在60000ms内不可用

private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
class FaultItem implements Comparable<FaultItem> {
    private final String name;
    private volatile long currentLatency;
    private volatile long startTimestamp;
}
public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
    FaultItem old = this.faultItemTable.get(name);
    if (null == old) {
        final FaultItem faultItem = new FaultItem(name);
        faultItem.setCurrentLatency(currentLatency);
        faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);

        old = this.faultItemTable.putIfAbsent(name, faultItem);
        if (old != null) {
            old.setCurrentLatency(currentLatency);
            old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
        }
    }
}

5、发送消息

该方法是消息发送核心方法，已经明确往哪个Broker发送消息了，里面设置到消息校验、消息发送前和发送后做的事情(消息轨迹就是在这里处理的后续文章会分析)、构建请求消息体最终调用remotingClient.invoke()并完成netty的网络请求（具体就是创建channel并将数据写入writeAndFlush，通过ChannelFutureListener进行监听返回结果）。

private SendResult sendKernelImpl(Message msg, MessageQueue mq, CommunicationMode communicationMode, SendCallback sendCallback, TopicPublishInfo topicPublishInfo, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        long beginStartTime = System.currentTimeMillis();
        String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
        SendMessageContext context = null;
        if (brokerAddr != null) {
            // 是否使用broker vip通道。broker会开启两个端口对外服务。
            brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
            byte[] prevBody = msg.getBody();
            SendResult var38;
            try {
                if (!(msg instanceof MessageBatch)) {  //批量消息
                    MessageClientIDSetter.setUniqID(msg);
                }
                if (this.hasCheckForbiddenHook()) {  // 发送消息校验
                    // ....
                }
                if (this.hasSendMessageHook()) {
                     //... 构造发送消息前 上下文
                    this.executeSendMessageHookBefore(context);
                }
                SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
                // 发送前请求参数构造
                if (requestHeader.getTopic().startsWith("%RETRY%")) {
                     // 判断是否是消息重试
                }
                SendResult sendResult = null;
                switch(communicationMode) {
                case ASYNC:
                    Message tmpMessage = msg;
                    boolean messageCloned = false;
                    sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), tmpMessage, requestHeader, timeout - costTimeAsync, communicationMode, sendCallback, topicPublishInfo, this.mQClientFactory, this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), context, this);
                    break;
                case ONEWAY:
                case SYNC:                
                    sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), msg, requestHeader, timeout - costTimeSync, communicationMode, context, this);
                    break;
                default: assert false;
                }
                // 发送消息后执行
                if (this.hasSendMessageHook()) {
                    context.setSendResult(sendResult);
                    this.executeSendMessageHookAfter(context);
                }
                var38 = sendResult;
            } catch (RemotingException var30) {
                if (this.hasSendMessageHook()) {
                    context.setException(var30);
                    this.executeSendMessageHookAfter(context);
                }
                throw var30;
            }  finally {
                msg.setBody(prevBody);
                msg.setTopic(NamespaceUtil.withoutNamespace(msg.getTopic(), this.defaultMQProducer.getNamespace()));
            }
            return var38;
        }  
    }

常见问题

NameServer挂了

如果Namesrv挂了，当新加入的生产消费则获取不到topic路由信息会报MQExecption；如果生产消费缓存了生产者有缓存 Topic 的路由信息，如果NameServer 全部挂掉，并且，此时依然可以发送消息。

Broker挂机

消息生产者每隔30s从nameser获取存活的broker，broker每隔30s向nameser发送存活情况。

如果Broker挂了分两种情况：

• sendLatencyFaultEnable：使用了故障延迟机制，通过获取一个MessageQueue发送失败，Broker 进行标记，标记该 Broker 在未来的某段时间内不会被选择到，默认为（5分钟，不可改变）

• 不启用sendLatencyFaultEnable ： procuder 每次发送消息，会采取轮询机制取下一个 MessageQueue,由于可能该 Message 所在的Broker挂掉，会抛出异常。因为一个 Broker 默认为一个 topic 分配4个 messageQueue,由于默认只重试2次