RocketMQ的5.0新特性Proxy分析解读

后端

2022-12-15 10:14:30

Proxy是什么？

RocketMQ自推出以来一直是消息中间件的佼佼者，其稳定性和高性能备受开发者青睐。在5.0版本中，RocketMQ引入了新的组件——Proxy。这一新增加的功能模块主要设计用于提升系统的可扩展性与灵活性，同时降低系统维护成本。

Proxy的作用

通过引入Proxy，RocketMQ能够实现消息发送与接收的负载均衡。Proxy作为独立的服务节点，不仅可以分担Broker的压力，还增强了系统容错能力。当某个Broker宕机时，其他Proxy可以无缝接管，从而保证服务持续可用。

如何部署Proxy？

在部署Proxy之前，确保环境满足运行条件：Java 8或更高版本、RocketMQ源码已正确编译并安装。

下载最新版的RocketMQ源代码。
编译源代码生成相应的Jar文件：
```
mvn clean install -DskipTests=true
```
配置application.properties以指定Proxy服务所需的参数，如监听端口、日志级别等。
启动Proxy服务。在RocketMQ安装目录下执行以下命令启动Proxy服务：
```
nohup sh bin/mqproxy -c ./conf/rocketmqproxy.conf &
```
确认Proxy服务已成功运行。

配置示例

# 代理服务器监听端口，用于接收来自客户端的请求。
listenPort=9876
# RocketMQ NameServer 地址列表
namesrvAddr=localhost:9874
# 日志级别设置为INFO
logLevel=INFO

使用Proxy进行消息发送

通过在RocketMQ客户端中配置Proxy信息，可以实现消息的透明转发。

修改生产者配置文件，添加Proxy地址。

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 设置代理服务器地址
producer.setBrokerAddr("http://localhost:9876")
producer.start();

消息发送示例：

Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        "OrderID188" /* Key */,
        ("Hello World").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Body */
);

SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);

使用Proxy进行消息接收

消费端配置类似，同样需要设置代理服务器地址来使用Proxy服务。

配置消费者：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_groupName");

// 设置代理服务器地址
consumer.setBrokerAddr("http://localhost:9876")

consumer.subscribe("TopicTest", "*");

实现消息监听器：

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.printf("Receive message %s at queue %d%n", new String(msg.getBody()), msg.getQueueId());
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

consumer.start();