為了支持長遠(yuǎn)的云原生發(fā)展，RocketMQ引入了一個全新的模塊：Proxy，官方對RocketMQ客戶端提供了獨(dú)立的開源項(xiàng)目：https://github.com/apache/rocketmq-clients，如果要使用這個新的客戶端，必須要使用Proxy作為endpoint。
Proxy有兩種搭建方式：
- LOCAL：本地模式，顧名思義，通過追加參數(shù)，在broker本地啟動
- CLUSTER：集群模式，作為獨(dú)立的集群啟動，搭建完nameserver和broker后，獨(dú)立部署
本地模式更適合非正式的場景，如調(diào)試、開發(fā)，線上環(huán)境還是推薦集群模式，本文基于集群模式進(jìn)行部署驗(yàn)證。
啟動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，默認(rèn)為CLUSTER（集群模式）
需要編輯的文件有：
- bin/runserver.sh：修改GC日志目錄和JVM參數(shù)；（非必須）
- conf/rmq-proxy.json：
主要是設(shè)置集群名、自定義端口
{
"rocketMQClusterName": "littleCat",
"remotingListenPort":28080,
"grpcServerPort":28081
}
完整參數(shù)見源碼：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
啟動成功
客戶端測試：
pom依賴：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client-java</artifactId>
<version>5.0.4</version>
</dependency>
生產(chǎn)者：
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();
}
}
消費(fèi)者：
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();
}
}
以上，搭建成功，消息生產(chǎn)消費(fèi)成功~
遇到的問題：
找到rmq.proxy,logback.xml，批量修改對應(yīng)的路徑：
將${brokerLogDir}批量替換成自定義的路徑即可。
dashboard目前還不能采集到客戶端的信息：文章來源：http://www.zghlxwxcb.cn/news/detail-428956.html
文章來源地址http://www.zghlxwxcb.cn/news/detail-428956.html
到了這里，關(guān)于【RocketMQ】RocketMQ 5.1.0版本Proxy集群模式部署實(shí)踐的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
