RocketMQ實戰(zhàn)與原理

---

第一章? 安裝、部署及簡單應用

1. 安裝

1.1 下載RocketMQ

2. 部署

2.1 部署nameserver

2.2 部署broker

2.3 新增Topic

3. 簡單應用

3.1 運行Producer

3.2 運行Consumer

---

第一章? 安裝、部署及簡單應用

1. 安裝

1.1 下載RocketMQ

RocketMQ下載地址：http://rocketmq.apache.org/release_notes/release-notes-4.3.0

選擇rocketmq-all-4.3.0-bin-release.zip下載

?

將其解壓，注意解壓目錄名不能帶有空格

?

2. 部署

命令、參數(shù)、代碼詳解比較復雜，故放后面說明，此處只寫部署步驟及運行代碼，保證可搭建部署簡單的應用

2.1 部署nameserver

進入bin目錄下，運行start mqnamesrv.cmd，出現(xiàn)The Name Server boot success.即代表nameserver啟動成功

?

2.2 部署broker

bin目錄下，運行start mqbroker.cmd -n 127.0.0.1:9876，出現(xiàn)The broker boot success.即代表broker啟動成功

?

2.3 新增Topic

bin目錄下，運行mqadmin.cmd updateTopic -n 127.0.0.1 -b 127.0.0.1:10911 -t TopicTest，出現(xiàn)create topic?success.即代表topic新增成功

3. 簡單應用

3.1 運行Producer

運行程序，將會發(fā)送100條消息

import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper;/*** @Author: Wen-Xueliang* @Date: Created in 2019/12/14 14:04* @Description: Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS notification, SMS marketing system, etc..*/ public class SyncProducer {public static void main(String[] args) throws Exception {//Instantiate with a producer group name.DefaultMQProducer producer = newDefaultMQProducer("please_rename_unique_group_name");// Specify name server addresses.producer.setNamesrvAddr("localhost:9876");//Launch the instance.producer.start();for (int i = 0; i < 100; i++) {//Create a message instance, specifying topic, tag and message body.Message msg = new Message("TopicTest" /* Topic */,"TagA" /* Tag */,("Hello RocketMQ " +i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);//Call send message to deliver message to one of brokers.SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);}//Shut down once the producer instance is not longer in use.producer.shutdown();} }

?

3.2 運行Consumer

運行程序，則會接受100條消息

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;/*** @Author: Wen-Xueliang* @Date: Created in 2019/12/14 14:05* @Description:*/ public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {// Instantiate with specified consumer group name.DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");// Specify name server addresses.consumer.setNamesrvAddr("localhost:9876");// Subscribe one more more topics to consume.consumer.subscribe("TopicTest", "*");// Register callback to execute on arrival of messages fetched from brokers.consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//Launch the consumer instance.consumer.start();System.out.printf("Consumer Started.%n");} }

?

至此，我們就完成了將RocketMQ安裝部署在Windows系統(tǒng)，并運行第一個程序，由生產(chǎn)者生產(chǎn)100條消息，運行第二個程序，由消費者消費100條消息。

后續(xù)，將繼續(xù)闡述RocketMQ的高級應用以及源碼的解析。

總結

以上是生活随笔為你收集整理的RocketMQ实战与原理---安装、部署及简单应用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。