【RocketMQ】RocketMQ 5.1.0版本Proxy集群模式部署实践

布满荆棘的人生 2023-09-30 21:19 2阅读 0赞

为了支持长远的云原生发展，RocketMQ引入了一个全新的模块：Proxy，官方对RocketMQ客户端提供了独立的开源项目：https://github.com/apache/rocketmq-clients，如果要使用这个新的客户端，必须要使用Proxy作为endpoint。

**Proxy有两种搭建方式：**

*  LOCAL：本地模式，顾名思义，通过追加参数，在broker本地启动
 *  CLUSTER：集群模式，作为独立的集群启动，搭建完nameserver和broker后，独立部署

本地模式更适合非正式的场景，如调试、开发，线上环境还是推荐集群模式，本文基于集群模式进行部署验证。

**启动proxy使用mqproxy命令：**

[root@XXGL-T-TJSYZ-REDIS-01 bin]# ./mqproxy -help
    usage: mqproxy [-bc <arg>] [-h] [-n <arg>] [-pc <arg>] [-pm <arg>]
     -bc,--brokerConfigPath <arg>   Broker config file path for local mode
     -h,--help                      Print help
     -n,--namesrvAddr <arg>         Name server address list, eg: '192.168.0.1:9876;192.168.0.2:9876'
     -pc,--proxyConfigPath <arg>    Proxy config file path
     -pm,--proxyMode <arg>          Proxy run in local or cluster mode
    [root@XXGL-T-TJSYZ-REDIS-01 bin]#

*  \-bc：使用本地模式时，指定broker的配置文件路径
 *  \-h：输出帮助信息；
 *  \-n：nameserver路径，也可以通过在配置文件中配置namesrvAddr指定；
 *  \-pc：proxy配置文件路径；
 *  \-pm：代理模式：LOCAL / CLUSTER，默认为CLUSTER（集群模式）

需要编辑的文件有：

*  bin/runserver.sh：修改GC日志目录和JVM参数；（非必须）
 *  conf/rmq-proxy.json：  
    主要是设置集群名、自定义端口

{
        
      "rocketMQClusterName": "littleCat",
      "remotingListenPort":28080,
      "grpcServerPort":28081
    }

完整参数见源码：org.apache.rocketmq.proxy.config.ProxyConfig

**启动脚本：**

#!/bin/bash
    . /etc/profile
    
    nohup sh /neworiental/rocketmq-5.1.0/rocketmq-proxy/bin/mqproxy -n '172.24.30.192:19876;172.24.30.193:19876;172.24.30.194:19876' -pc /neworiental/rocketmq-5.1.0/rocketmq-proxy/conf/rmq-proxy.json >/dev/null 2>&1 &
    echo "deploying rocketmq-proxy..."

**停止脚本：**

#!/bin/bash
    . /etc/profile
    
    PID=`ps -ef | grep '/neworiental/rocketmq-5.1.0/rocketmq-proxy' | grep -v grep | awk '{print $2}'`
    if [[ "" !=  "$PID" ]]; then
      echo "killing rocketmq-proxy : $PID"
      kill $PID
    fi

启动成功

![3448ab7a90729d5ef7afd92dc9cb6386.png][]

**客户端测试：**

pom依赖：

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client-java</artifactId>
                <version>5.0.4</version>
            </dependency>

生产者：

package cn.xdf.xadd.rmq.test.newclient;
    
    import org.apache.rocketmq.client.apis.ClientConfiguration;
    import org.apache.rocketmq.client.apis.ClientException;
    import org.apache.rocketmq.client.apis.ClientServiceProvider;
    import org.apache.rocketmq.client.apis.message.Message;
    import org.apache.rocketmq.client.apis.producer.Producer;
    import org.apache.rocketmq.client.apis.producer.SendReceipt;
    
    import java.io.IOException;
    import java.nio.charset.StandardCharsets;
    
    public class ProducerNormalMessageExample {
        
    
        public static void main(String[] args) throws ClientException, IOException {
        
            final ClientServiceProvider provider = ClientServiceProvider.loadService();
    
            String endpoints = "172.24.30.192:28080";
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                    .setEndpoints(endpoints)
                    .build();
            String topic = "zhurunhua-test";
            // In most case, you don't need to create too many producers, singleton pattern is recommended.
            final Producer producer = provider.newProducerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    // Set the topic name(s), which is optional but recommended. It makes producer could prefetch the topic
                    // route before message publishing.
                    .setTopics(topic)
                    // May throw {@link ClientException} if the producer is not initialized.
                    .build();
            // Define your message body.
            byte[] body = "This is a normal message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
            String tag = "new-client-test";
            final Message message = provider.newMessageBuilder()
                    // Set topic for the current message.
                    .setTopic(topic)
                    // Message secondary classifier of message besides topic.
                    .setTag(tag)
                    // Key(s) of the message, another way to mark message besides message id.
                    .setKeys("test")
                    .setBody(body)
                    .build();
            try {
        
                final SendReceipt sendReceipt = producer.send(message);
                System.out.println("Send message successfully, messageId=" + sendReceipt.getMessageId());
            } catch (Throwable t) {
        
                System.err.println(t);
            }
            // Close the producer when you don't need it anymore.
            producer.close();
        }
    }

![e241cf8bd8491922d2a3c34658f04196.png][]

消费者：

package cn.xdf.xadd.rmq.test.newclient;
    
    import org.apache.rocketmq.client.apis.ClientConfiguration;
    import org.apache.rocketmq.client.apis.ClientException;
    import org.apache.rocketmq.client.apis.ClientServiceProvider;
    import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
    import org.apache.rocketmq.client.apis.consumer.FilterExpression;
    import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
    import org.apache.rocketmq.client.apis.consumer.PushConsumer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.IOException;
    import java.util.Collections;
    
    public class PushConsumerExample {
        
        private static final Logger log = LoggerFactory.getLogger(PushConsumerExample.class);
    
        public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        
            final ClientServiceProvider provider = ClientServiceProvider.loadService();
    
            String endpoints = "172.24.30.192:28080";
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                    .setEndpoints(endpoints)
                    .build();
            String tag = "new-client-test";
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
            String consumerGroup = "new-client-test-group";
            String topic = "zhurunhua-test";
            PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    // Set the consumer group name.
                    .setConsumerGroup(consumerGroup)
                    // Set the subscription for the consumer.
                    .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                    .setMessageListener(messageView -> {
        
                        // Handle the received message and return consume result.
                        log.info("Consume message={}", messageView);
                        return ConsumeResult.SUCCESS;
                    })
                    .build();
            // Block the main thread, no need for production environment.
            Thread.sleep(Long.MAX_VALUE);
            // Close the push consumer when you don't need it anymore.
            pushConsumer.close();
        }
    }

![19402f7c28ffb549443b7c40d1926614.png][]

以上，搭建成功，消息生产消费成功~

**遇到的问题：**

![b5aa7d6d56142b7ce9ba19808f96bd0e.png][]

找到rmq.proxy,logback.xml，批量修改对应的路径：

![c206afe33c7107a03a6377e8fd0b564e.png][]

将$\{brokerLogDir\}批量替换成自定义的路径即可。

dashboard目前还不能采集到客户端的信息：

![d7a66943085c8221ea6326d4f7ec1bbf.png][]

[3448ab7a90729d5ef7afd92dc9cb6386.png]: https://img-blog.csdnimg.cn/img_convert/3448ab7a90729d5ef7afd92dc9cb6386.png
[e241cf8bd8491922d2a3c34658f04196.png]: https://img-blog.csdnimg.cn/img_convert/e241cf8bd8491922d2a3c34658f04196.png
[19402f7c28ffb549443b7c40d1926614.png]: https://img-blog.csdnimg.cn/img_convert/19402f7c28ffb549443b7c40d1926614.png
[b5aa7d6d56142b7ce9ba19808f96bd0e.png]: https://img-blog.csdnimg.cn/img_convert/b5aa7d6d56142b7ce9ba19808f96bd0e.png
[c206afe33c7107a03a6377e8fd0b564e.png]: https://img-blog.csdnimg.cn/img_convert/c206afe33c7107a03a6377e8fd0b564e.png
[d7a66943085c8221ea6326d4f7ec1bbf.png]: https://img-blog.csdnimg.cn/img_convert/d7a66943085c8221ea6326d4f7ec1bbf.png