RocketMQ 4.2.0 在windows上安装和开发使用

小灰灰 2022-05-21 13:41 189阅读 0赞

# Apache RocketMQ 4.2.0 在windows上安装和开发使用 #

2018年06月15日 19:06:17

阅读数：83

# 1.概述 #

RocketMQ是alibaba公司开源的一个纯java的开源消息中间件。

文档：[http://rocketmq.apache.org/docs/quick-start/][http_rocketmq.apache.org_docs_quick-start]

# 2.开发测试环境搭建 #

## 2.1 下载安装文件 ##

## 2.1.1 下载源码文件 ##

源码下载：[https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.2.0/rocketmq-all-4.2.0-source-release.zip][https_www.apache.org_dyn_closer.cgi_path_rocketmq_4.2.0_rocketmq-all-4.2.0-source-release.zip]

源码包编译生成二进制文件命令：

进入到RocketMQ下载包解压的路径下

1.  > cdrocketmq-all-4.2.0/
2.  > mvn -Prelease-all -DskipTests clean install -U
3.  > cddistribution/target/apache-rocketmq

接下来把编译好的项目copy出来   D:\\java\\soft\\rocketmq-all-4.2.0\\distribution\\target\\apache-rocketmq

放到D:\\java\\soft\\apache-rocketmq路径。

## 2.1.2 下载编译后的二进制文件 ##

编译后的二进制文件（下载此文件则不需要编译）：[http://rocketmq.apache.org/release\_notes/release-notes-4.2.0/][http_rocketmq.apache.org_release_notes_release-notes-4.2.0]

直接解压，D:\\java\\soft\\rocketmq-all-4.2.0-bin-release

## 2.2.设置环境变量 ##

源码： ROCKETMQ\_HOME : D:\\java\\soft\\apache-rocketmq

二进制文件： ROCKETMQ\_HOME : D:\\java\\soft\\rocketmq-all-4.2.0-bin-release

## 2.3进入bin目录 ##

![70][]

## 2.4 启动name server ##

双击mqnamesrv.exe，启动name server，保持mqnamesrv.exe运行，不要关闭这个终端。

![70 1][]

## 2.5 启动broker ##

方法一：开启另一个windows终端cmd，进入解压的bin目录，先输入`set NAMESRV_ADDR=127.0.0.1:9876`设置环境变量，输入`mqbroker.exe`启动broker，保持mqbroker.exe运行，不要关闭这个终端。   
方法二：开启另一个windows终端cmd，进入解压的bin目录，也可一步输入`mqbroker -n 127.0.0.1:9876`启动broker，保持mqbroker.exe运行，不要关闭这个终端。

![70 2][]

## 2.6.项目实例 ##

项目下载地址：[https://github.com/liyanlei58/rocketmq\_demo][https_github.com_liyanlei58_rocketmq_demo]

1.      创建maven项目

![70 3][]

Pom.xml文件如下

<?xml version="1.0" encoding="UTF-8"?>
 
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 
 <groupId>com.lyl.mq</groupId>
 <artifactId>rocketmq_demo</artifactId>
 <version>1.0-SNAPSHOT</version>
 
 <name>rocketmq_demo</name>
 
 <url>http://www.example.com</url>
 
 <properties>
 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 <maven.compiler.source>1.7</maven.compiler.source>
 <maven.compiler.target>1.7</maven.compiler.target>
 </properties>
 
 <dependencies>
 <dependency>
 <groupId>junit</groupId>
 <artifactId>junit</artifactId>
 <version>4.11</version>
 <scope>test</scope>
 </dependency>
 
 <dependency>
 <groupId>org.apache.rocketmq</groupId>
 <artifactId>rocketmq-client</artifactId>
 <version>4.2.0</version>
 </dependency>
 <dependency>
 <groupId>org.apache.rocketmq</groupId>
 <artifactId>rocketmq-all</artifactId>
 <version>4.2.0</version>
 <type>pom</type>
 </dependency>
 <dependency>
 <groupId>ch.qos.logback</groupId>
 <artifactId>logback-classic</artifactId>
 <version>1.1.1</version>
 </dependency>
 <dependency>
 <groupId>ch.qos.logback</groupId>
 <artifactId>logback-core</artifactId>
 <version>1.1.1</version>
 </dependency>
 </dependencies>
 
 <build>
 <pluginManagement>
 <plugins>
 <plugin>
 <artifactId>maven-clean-plugin</artifactId>
 <version>3.0.0</version>
 </plugin>
 
 <plugin>
 <artifactId>maven-resources-plugin</artifactId>
 <version>3.0.2</version>
 </plugin>
 <plugin>
 <artifactId>maven-compiler-plugin</artifactId>
 <version>3.7.0</version>
 </plugin>
 <plugin>
 <artifactId>maven-surefire-plugin</artifactId>
 <version>2.20.1</version>
 </plugin>
 <plugin>
 <artifactId>maven-jar-plugin</artifactId>
 <version>3.0.2</version>
 </plugin>
 <plugin>
 <artifactId>maven-install-plugin</artifactId>
 <version>2.5.2</version>
 </plugin>
 <plugin>
 <artifactId>maven-deploy-plugin</artifactId>
 <version>2.8.2</version>
 </plugin>
 </plugins>
 </pluginManagement>
 </build>
 </project>

2.      编写消息产生者Producer

package com.lyl.mq;
 
 import java.util.concurrent.TimeUnit;
 
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.message.Message;
 
 public class Producer {
 public static void main(String[] args) throws MQClientException, InterruptedException {
 /** * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例 * 注意：ProducerGroupName需要由应用来保证唯一 * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键， * 因为服务器会回查这个Group下的任意一个Producer */ final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
 producer.setNamesrvAddr("127.0.0.1:9876");
 producer.setInstanceName("Producer");
 
 /** * Producer对象在使用之前必须要调用start初始化，初始化一次即可 * 注意：切记不可以在每次发送消息时，都调用start方法 */ producer.start();
 
 /** * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。 * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态， * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高， * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。 */ for (int i = 0; i < 10; i++) {
 try {
 {
 Message msg = new Message("TopicTest1",// topic
 "TagA",// tag
 "OrderID001",// key
 ("Hello MetaQA").getBytes());// body
 SendResult sendResult = producer.send(msg);
 System.out.println(sendResult);
 }
 
 {
 Message msg = new Message("TopicTest2",// topic
 "TagB",// tag
 "OrderID0034",// key
 ("Hello MetaQB").getBytes());// body
 SendResult sendResult = producer.send(msg);
 System.out.println(sendResult);
 }
 
 {
 Message msg = new Message("TopicTest3",// topic
 "TagC",// tag
 "OrderID061",// key
 ("Hello MetaQC").getBytes());// body
 SendResult sendResult = producer.send(msg);
 System.out.println(sendResult);
 }
 } catch (Exception e) {
 e.printStackTrace();
 }
 TimeUnit.MILLISECONDS.sleep(1000);
 }
 
 /** * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己 * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法 */ // producer.shutdown();
 Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
 public void run() {
 producer.shutdown();
 }
 }));
 System.exit(0);
 }
 }

3.      编写消息消费者Consumer

package com.lyl.mq ; 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 ; public class PushConsumer \{ */\*\** *\** *当前例子是* *PushConsumer* *用法，使用方式给用户感觉是消息从* *RocketMQ* *服务器推到了应用客户端。* * * *\** *但是实际* *PushConsumer* *内部是使用长轮询* *Pull* *方式从* *MetaQ* *服务器拉消息，然后再回调用户* *Listener* *方法* * * *\*/* public static void main(String\[\] args) throws InterruptedException , MQClientException \{ */\*\** *\** *一个应用创建一个* *Consumer* *，由应用来维护此对象，可以设置为全局对象或者单例* * * *\** *注意：* *ConsumerGroupName* *需要由应用来保证唯一* *\*/* DefaultMQPushConsumer consumer = new DefaultMQPushConsumer( "ConsumerGroupName") ; consumer.setNamesrvAddr( "127.0.0.1:9876") ; consumer.setInstanceName( "Consumber") ; */\*\** *\** *订阅指定* *topic* *下* *tags* *分别等于* *TagA* *或* *TagC* *或* *TagD* *\*/* consumer.subscribe( "TopicTest1" , "TagA || TagC || TagD") ; */\*\** *\** *订阅指定* *topic* *下所有消息* * * *\** *注意：一个* *consumer* *对象可以订阅多个* *topic* *\*/* consumer.subscribe( "TopicTest2" , "\*") ; consumer.registerMessageListener( new MessageListenerConcurrently() \{ public ConsumeConcurrentlyStatus consumeMessage( List<MessageExt> msgs , ConsumeConcurrentlyContext context) \{ System. *out*.println(Thread. *currentThread*().getName() + " Receive New Messages: " \+ msgs.size()) ; MessageExt msg = msgs.get( 0) ; if (msg.getTopic().equals( "TopicTest1")) \{ // 执行 TopicTest1 的消费逻辑 if(msg.getTags() != null)\{ if( "TagA".equals(msg.getTags()))\{ // 执行 TagA 的消费 System. *out*.println( new String(msg.getBody())) ; \} else if ( "TagC".equals(msg.getTags()))\{ // 执行 TagC 的消费 System. *out*.println( new String(msg.getBody())) ; \} else if ( "TagD".equals(msg.getTags()))\{ // 执行 TagD 的消费 System. *out*.println( new String(msg.getBody())) ; \} \} \} else if (msg.getTopic().equals( "TopicTest2")) \{ System. *out*.println( new String(msg.getBody())) ; \} return ConsumeConcurrentlyStatus. *CONSUME\_SUCCESS* ; \} \}) ; */\*\** *\* Consumer* *对象在使用之前必须要调用* *start* *初始化，初始化一次即可* * * *\*/* consumer.start() ; System. *out*.println( "ConsumerStarted.") ; \}\}

[http_rocketmq.apache.org_docs_quick-start]: http://rocketmq.apache.org/docs/quick-start/
[https_www.apache.org_dyn_closer.cgi_path_rocketmq_4.2.0_rocketmq-all-4.2.0-source-release.zip]: https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.2.0/rocketmq-all-4.2.0-source-release.zip
[http_rocketmq.apache.org_release_notes_release-notes-4.2.0]: http://rocketmq.apache.org/release_notes/release-notes-4.2.0/
[70]: /images/20220521/96cacb8a0d0749279c239f53f9b6a372.png
[70 1]: /images/20220521/b1c5ac04db2346f6b71a7f5b65d641e7.png
[70 2]: /images/20220521/9df311e2d1d6493d98f5060263e5d225.png
[https_github.com_liyanlei58_rocketmq_demo]: https://github.com/liyanlei58/rocketmq_demo
[70 3]: /images/20220521/4d3b82eeb87d437eaa1506fa52cf3623.png