RocketMQ 系列 - 架构设计之消息

// 使用GroupName初始化Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 指定NameSrv的地址: 也可以通过环境变量NAMESRV_ADDR来指定，则不需要下面这一行。
producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
// 启动实例
producer.start();
try {
    // 创建消息实例,指定 topic, tag, message body.
    Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        ("Hello RocketMQ !!").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    // 发送消息给Broker
    SendResult sendResult = producer.send(msg);
    System.out.printf("%s%n", sendResult);
} catch (Exception e) {
    e.printStackTrace();
    Thread.sleep(1000);
}
// 关闭生产者
producer.shutdown();

/**
 * 延迟容错
 *
 * @param <T>
 */
public interface LatencyFaultTolerance<T> {
    /**
     * 更新失败的Broker。
     *
     * @param name                 brokerName
     * @param currentLatency       本次消息发送的时长
     * @param notAvailableDuration 规避broker的时长
     */
    void updateFaultItem(final T name, final long currentLatency, final long notAvailableDuration);

    /**
     * 判断Broker是否可以用
     *
     * @param name broker的名称
     * @return
     */
    boolean isAvailable(final T name);

    /**
     * 从Broker故障列表中移除
     * @param name
     */
    void remove(final T name);

    /**
     * 弹出最后一个 Broker
     * @return
     */
    T pickOneAtLeast();
}

@Override
public boolean isAvailable(final String name) {
    // faultItemTable 是一个 ConcurrentHashMap
    final FaultItem faultItem = this.faultItemTable.get(name);
    if (faultItem != null) {
        // 判断该Broker(messageQueue)是否在规避时间内。
        return faultItem.isAvailable();
    }
    return true;
}

public boolean isAvailable() {
    return (System.currentTimeMillis() - startTimestamp) >= 0;
}

 // 更新topic的一个MessageQueue 是否可用
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);

/**
  * 延迟机制 - 本次消息发送时长区间，用于映射计算出需要规避Broker的时长。
  */
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
/**
  * 延迟机制 - 规避Broker的时长
  */
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

public boolean isAvailable() {
  return (System.currentTimeMillis() - startTimestamp) >= 0;
}

response = asyncProcessRequest(ctx, request).get();

public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final SendMessageContext mqtraceContext;
        switch (request.getCode()) {
            // 消费者发送回来的消息(可能是消费失败之后的?)
            case RequestCode.CONSUMER_SEND_MSG_BACK:
                return this.asyncConsumerSendMsgBack(ctx, request);
            default:
                // 处理生产者发送来的消息
                SendMessageRequestHeader requestHeader = parseRequestHeader(request);
                if (requestHeader == null) {
                    return CompletableFuture.completedFuture(null);
                }
                mqtraceContext = buildMsgContext(ctx, requestHeader);
                // 消息发送前的钩子。在消息发送之前，执行一些操作。
                this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
                if (requestHeader.isBatch()) {
                    // 如果是批量消息
                    return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);
                } else {
                    // 不是批量消息
                    return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);
                }
        }
    }

public MappedFile getLastMappedFile() {
    MappedFile mappedFileLast = null;
    while (!this.mappedFiles.isEmpty()) {
        try {
            mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
            break;
        } catch (IndexOutOfBoundsException e) {
            //continue;
        } catch (Exception e) {
            log.error("getLastMappedFile has exception.", e);
            break;
        }
    }
    return mappedFileLast;
}

public MappedFile getLastMappedFile(final long startOffset) {
    return getLastMappedFile(startOffset, true);
}

/** 每10ms从NameServer拉取一下 TopicRouteInfo*/
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            MQClientInstance.this.updateTopicRouteInfoFromNameServer();
        } catch (Exception e) {
            log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
        }
    }
}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

// 本机 启动时=>其他的消费者会受到 consumerIds发生变更的请求。
/** 5.启动负载均衡服务 org.apache.rocketmq.client.impl.factory.MQClientInstance#start*/
this.rebalanceService.start();
// 其他的消息者启动或者关闭时,即 consumerId 发生变动的时候 org.apache.rocketmq.client.impl.ClientRemotingProcessor#processRequest
case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
    return this.notifyConsumerIdsChanged(ctx, request);
// 控制端 重置消费偏移量时 org.apache.rocketmq.client.impl.ClientRemotingProcessor#processRequest
case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
    return this.resetOffset(ctx, request);

// 消费者的模式: 广播还是集群模式
switch (messageModel) {
    // 广播方式。
    case BROADCASTING: {
        // topicSubscribeInfoTable: 存储topic 对应的 MessageQueue 列表.
        Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
        if (mqSet != null) {
            // 更新 消费者消费订阅表。返回是否发生了变动。
            boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
            if (changed) {
                // 如果发生了变动，则根据不同的消费消费方式处理：
                // push,则需要通知给Broker，更新各个消费者的订阅关系。
                // pull,则会通知消费者，执行消费逻辑。
                this.messageQueueChanged(topic, mqSet, mqSet);
                log.info("messageQueueChanged {} {} {} {}", consumerGroup, topic, mqSet, mqSet);
            }
        } else {
            log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
        }
        break;
    }
    case ....:
    // 省略..
}

// 当前消费者 需要消费哪几个消息队列中消息的分配策略。
AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

List<MessageQueue> allocateResult = null;
try {
    // mqAll:当前topic和group下的消费队列列表.   cidAll: 所有消费队列的id.
    // 两者一一对应。
    allocateResult = strategy.allocate(this.consumerGroup, this.mQClientFactory.getClientId(),
        mqAll, cidAll);
}catch(Throwable e){}

// cidAll: 消费者id列表
int index = cidAll.indexOf(currentCID);
// mqAll: 消息队列的列表
// mod: 表示 有多个队列无法平均分配。
int mod = mqAll.size() % cidAll.size();

// 4 <= 5 ? 1 : (4 > 0 && 1 < 4 ? 4 / 5 + 1 : 4 / 5)
int averageSize;
if (mqAll.size() <= cidAll.size()) {
    // 队列比消费者少，每个消费者最大消费1个队列。
    averageSize = 1;
} else {
    if (mod > 0 && index < mod) {
        // 不能平均分, 并且消费者的id比余数小。那么该消费者要多消费一个MessageQueue.
        averageSize = mqAll.size() / cidAll.size() + 1;
    } else {
        // 正好平均分 或者 当前消费者的id大于mod, 那么该消费者就平均分.
        // 会有两种情况: 1、mq有5个,consumer有5个， cid=5,则每个消费者都正好订阅一个mq
        // 2、mq有5个，consumer有6个,cid=6, 则该消费者不会订阅mq
        averageSize = mqAll.size() / cidAll.size();
    }
}
//        int averageSize =
//            mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
//                + 1 : mqAll.size() / cidAll.size());

// 订阅mq 的起始 索引. 比如5个mq，3个consumer, 那么,c1=>m1,m2; c2=>m3,m4; c3=>m5
int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
int range = Math.min(averageSize, mqAll.size() - startIndex);
for (int i = 0; i < range; i++) {
    result.add(mqAll.get((startIndex + i) % mqAll.size()));
}
return result;

// index: 表示第index个消费者
int index = cidAll.indexOf(currentCID);
for (int i = index; i < mqAll.size(); i++) {
    // 循环分配给消费者。
    if (i % cidAll.size() == index) {
        result.add(mqAll.get(i));
    }
}

// 哈希环的节点
Collection<ClientNode> cidNodes = new ArrayList<>();
for (String cid : cidAll) {
    cidNodes.add(new ClientNode(cid));
}

// 创建一致性哈希的Router
final ConsistentHashRouter<ClientNode> router;
if (customHashFunction != null) {
    router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt, customHashFunction);
} else {
    router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
}

// 将MQf通过Router分配给消费者
List<MessageQueue> results = new ArrayList<>();
for (MessageQueue mq : mqAll) {
    ClientNode clientNode = router.routeNode(mq.toString());
    if (clientNode != null && currentCID.equals(clientNode.getKey())) {
        results.add(mq);
    }
}

return results;

public void setMessageQueueList(List<MessageQueue> messageQueueList) {
    this.messageQueueList = messageQueueList;
}

@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
    List<String> cidAll) {
    return this.messageQueueList;
}

List<MessageQueue> premqAll = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
    String[] temp = mq.getBrokerName().split("@");
    if (temp.length == 2 && consumeridcs.contains(temp[0])) {
        premqAll.add(mq);
    }
}

int mod = premqAll.size() / cidAll.size();
int rem = premqAll.size() % cidAll.size();
int startIndex = mod * currentIndex;
int endIndex = startIndex + mod;
for (int i = startIndex; i < endIndex; i++) {
    result.add(premqAll.get(i));
}
if (rem > currentIndex) {
    result.add(premqAll.get(currentIndex + mod * cidAll.size()));
}
return result;

//  根据机房分组 broker,
Map<String/*machine room */, List<MessageQueue>> mr2Mq = new TreeMap<String, List<MessageQueue>>();
for (MessageQueue mq : mqAll) {
    String brokerMachineRoom = machineRoomResolver.brokerDeployIn(mq);
    if (StringUtils.isNoneEmpty(brokerMachineRoom)) {
        if (mr2Mq.get(brokerMachineRoom) == null) {
            mr2Mq.put(brokerMachineRoom, new ArrayList<MessageQueue>());
        }
        mr2Mq.get(brokerMachineRoom).add(mq);
    } else {
        throw new IllegalArgumentException("Machine room is null for mq " + mq);
    }
}

//根据机房分组consumer
Map<String/*machine room */, List<String/*clientId*/>> mr2c = new TreeMap<String, List<String>>();
for (String cid : cidAll) {
    // 根据cid判断出所属的机房==> 相同机房的消费需要有相同的命名格式.
    String consumerMachineRoom = machineRoomResolver.consumerDeployIn(cid);
    if (StringUtils.isNoneEmpty(consumerMachineRoom)) {
        if (mr2c.get(consumerMachineRoom) == null) {
            mr2c.put(consumerMachineRoom, new ArrayList<String>());
        }
        mr2c.get(consumerMachineRoom).add(cid);
    } else {
        throw new IllegalArgumentException("Machine room is null for consumer id " + cid);
    }
}

List<MessageQueue> allocateResults = new ArrayList<MessageQueue>();

// 计算当前consumer可以消费的MQ. 当前机房有多个消费者的时候，则使用真正分配策略(其他的分配策略)进行分配。
// 1.allocate the mq that deploy in the same machine room with the current consumer
String currentMachineRoom = machineRoomResolver.consumerDeployIn(currentCID);
List<MessageQueue> mqInThisMachineRoom = mr2Mq.remove(currentMachineRoom);
List<String> consumerInThisMachineRoom = mr2c.get(currentMachineRoom);
if (mqInThisMachineRoom != null && !mqInThisMachineRoom.isEmpty()) {
    allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mqInThisMachineRoom, consumerInThisMachineRoom));
}

// 2.allocate the rest mq to each machine room if there are no consumer alive in that machine room
// 如果当前机房没有存活的消费者，那么当前机房的MQ会被所有的消费者共享。根据真实的分配策略进行分配
for (String machineRoom : mr2Mq.keySet()) {
    if (!mr2c.containsKey(machineRoom)) { // no alive consumer in the corresponding machine room, so all consumers share these queues
        allocateResults.addAll(allocateMessageQueueStrategy.allocate(consumerGroup, currentCID, mr2Mq.get(machineRoom), cidAll));
    }
}

return allocateResults;

public void putMessagePositionInfo(DispatchRequest dispatchRequest) {
    ConsumeQueue cq = this.findConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
    cq.putMessagePositionInfoWrapper(dispatchRequest);
}

@Override
public void run() {

    while (!this.isStopped()) {
        try {
            // 出队
            PullRequest pullRequest = this.pullRequestQueue.take();
            // 拉取消息
            this.pullMessage(pullRequest);
        }
    }
}

/**
* 顺序消费
* 每秒 定时多实例消费的所有队列，上锁成功将 ProcessQueue的lock属性设置为true
*/
@Override
public void start() {
    if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) { /// 只支持集群模式
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                ConsumeMessageOrderlyService.this.lockMQPeriodically();
            }
        }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
    }
}

public class ConsumeMessageOrderlyService implements ConsumeMessageService {
    class ConsumeRequest implements Runnable {
        private final ProcessQueue processQueue;
        private final MessageQueue messageQueue;

        public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
            this.processQueue = processQueue;
            this.messageQueue = messageQueue;
        }

        public ProcessQueue getProcessQueue() {
            return processQueue;
        }

        public MessageQueue getMessageQueue() {
            return messageQueue;
        }

        @Override
        public void run() {
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock) {
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())|| (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    for (boolean continueConsume = true; continueConsume; ) {
                        // 每次获取消息的数量
                        final int consumeBatchSize =
                            ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                        // 拿到可以消费的消息列表。默认为 1 条。
                        List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);
                            ConsumeOrderlyStatus status = null;
                            ConsumeMessageContext consumeMessageContext = null;
                            try {
                                // 消费者加锁
                                this.processQueue.getConsumeLock().lock();
                                // 回调消费者消息
                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                            } finally {
                                this.processQueue.getConsumeLock().unlock();
                            }
                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } else {
                            continueConsume = false;
                        }
                    }
                } else {
                    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
                }
            }
        }
    }
}