揭秘RocketMQ5源码：新架构下的普通消息收发

2023-04-23 17:01:32

RocketMQ 5 中普通消息收发的机制详解

RocketMQ 5 带来了一套全新的架构，其中对普通消息收发的机制进行了重大改造。本文将深入探讨新架构下的普通消息收发过程，了解它如何提升消息传递的效率、降低延迟并增强可靠性。

gRPC 客户端：从头到尾的请求过程

RocketMQ 5 的 gRPC 客户端是向 Broker 发送消息的主要接口。客户端承担着将消息封装成 gRPC 请求并在网络上传输到指定 Broker 的职责。为了确保消息可靠地到达目标 Broker，gRPC 客户端采用了以下策略：

重试机制： 如果发送消息失败，客户端将自动重试，直到消息成功发送为止。
死信队列： 如果消息无法成功发送，客户端会将消息放入死信队列中。死信队列中的消息可以由人工处理。

// 示例代码（Java）：
import io.openmessaging.api.Message;
import io.openmessaging.api.Producer;
import io.openmessaging.connector.rocketmq.producer.RocketmqProducer;

public class ProducerExample {

    public static void main(String[] args) {
        Producer producer = new RocketmqProducer();
        producer.start();

        Message message = producer.createMessage("test_topic", "hello_world");
        producer.send(message);

        producer.shutdown();
    }
}

Broker 代理：承上启下，消息的转接枢纽

Broker 代理是 RocketMQ 新架构中的关键组件，负责接收来自 gRPC 客户端的消息请求并将其转发给对应的 Broker。为了确保消息能够可靠地到达目标 Broker，代理采用了以下策略：

重试机制： 如果转发消息失败，代理会自动重试，直到消息成功转发为止。
死信队列： 如果消息无法成功转发，代理会将消息放入死信队列中。死信队列中的消息可以由人工处理。

普通消息收发：串联起客户端和代理的流程

普通消息的收发过程是 gRPC 客户端和 Broker 代理之间密切配合的结果。客户端将消息发送到代理，代理再将消息转发给 Broker。这个看似简单的过程背后其实涉及大量细节和优化。

// 示例代码（Java）：
import com.google.protobuf.ByteString;
import io.openmessaging.connector.rocketmq.domain.MessageQueue;
import io.openmessaging.connector.rocketmq.domain.TopicConfig;
import io.openmessaging.connector.rocketmq.producer.protocol.SendRequest;
import io.openmessaging.connector.rocketmq.producer.protocol.SendResponse;
import io.openmessaging.connector.rocketmq.producer.transport.TransportProducer;

public class CustomProducerExample {

    public static void main(String[] args) {
        TransportProducer producer = new TransportProducer();
        producer.start();

        MessageQueue messageQueue = new MessageQueue();
        messageQueue.setTopic("test_topic");
        messageQueue.setBrokerName("broker-a");
        messageQueue.setQueueId(0);

        ByteString body = ByteString.copyFromUtf8("hello_world");

        SendRequest request = new SendRequest();
        request.setMessageQueue(messageQueue);
        request.setBody(body);

        SendResponse response = producer.send(request);

        producer.shutdown();
    }
}