RocketMQ简单实例搭建

系统管理员 2021-12-18 05:21 325阅读 0赞

1、安装RocketMQ（在Linux环境下，我用的是Centos6.5）  
安装RocketMQ需要jdk1.6以上, maven,git环境，以上环境自行百度命令安装。

git clone https://github.com/alibaba/RocketMQ.git \#\#从github上下载RocketMQ开源项目

cd RocketMQ \#\#进入文件夹

sh install.sh \#\#开始安装  
1  
2  
3  
4  
5  
安装完之后可以看到下图这样：其中可以看到一个符号链接devenv如红框所示  
  
然后

cd devenv/bin \#\#进入链接的目录下的bin目录

nohup sh mqnamesrv -n "121.42.179.195:9876" & \#\#配置nameserver，121.42.179.195是本机ip，也就是服务器外网地址

nohup sh mqbroker -n "121.42.179.195:9876" & \#\#配置broker，121.42.179.195同上  
1  
2  
3  
4  
5  
之后

cat nohup.out  
1  
在输出的最低端，可以看到红框中的两句话则说明nameserver和broker启动成功。  
  
如果服务器内存不够，你就会启动失败，可以修改runbroker.sh脚本（mqbroker文件中通过runbroker.sh脚本调用Broker的主函数com.alibaba.rocketmq.broker.BrokerStartup启动Broker）的JAVA\_OPT参数

vi runbroker.sh  
1  
我阿里云内存小，我就改成这样

JAVA\_OPT="$\{JAVA\_OPT\} -server -Xms128m -Xmx128m -Xmn128m -XX:PermSize=128m -XX:MaxPermSize=128m"  
1  
2、编写Consumer和Producer测试类  
首先需要的jar包如下：

<?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>groupId</groupId>  
<artifactId>RocketMQ</artifactId>  
<version>1.0-SNAPSHOT</version>  
<build>  
<plugins>  
<plugin>  
<groupId>org.apache.maven.plugins</groupId>  
<artifactId>maven-compiler-plugin</artifactId>  
<configuration>  
<source>1.6</source>  
<target>1.6</target>  
</configuration>  
</plugin>  
</plugins>  
</build>

<name>RocketMQTest</name>  
<url>http://maven.apache.org</url>

<dependencies>  
<dependency>  
<groupId>com.alibaba.rocketmq</groupId>  
<artifactId>rocketmq-client</artifactId>  
<version>3.0.10</version>  
</dependency>  
<dependency>  
<groupId>com.alibaba.rocketmq</groupId>  
<artifactId>rocketmq-all</artifactId>  
<version>3.0.10</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>  
<dependency>  
<groupId>junit</groupId>  
<artifactId>junit</artifactId>  
<version>4.10</version>  
<scope>test</scope>  
</dependency>  
</dependencies>  
</project>  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
36  
37  
38  
39  
40  
41  
42  
43  
44  
45  
46  
47  
48  
49  
50  
51  
52  
53  
54  
55  
56  
57  
58  
59  
Consumer类：

public class Consumer \{  
public static void main(String\[\] args) throws MQClientException \{  
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(  
"rmq-group");

consumer.setNamesrvAddr("121.42.179.195:9876");  
consumer.setInstanceName("consumer");  
consumer.subscribe("TopicA-test", "TagA");

consumer.registerMessageListener(new MessageListenerConcurrently() \{  
@Override  
public ConsumeConcurrentlyStatus consumeMessage(  
List<MessageExt> msgs, ConsumeConcurrentlyContext context) \{  
for (MessageExt msg : msgs) \{  
System.out.println(new String(msg.getBody()));  
\}  
return ConsumeConcurrentlyStatus.CONSUME\_SUCCESS;  
\}  
\});  
consumer.start();  
System.out.println("Consumer Started.");  
\}  
\}  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
Producer类：

public class Producer \{  
public static void main(String\[\] args) throws MQClientException \{  
DefaultMQProducer producer = new DefaultMQProducer("rmq-group");  
producer.setNamesrvAddr("121.42.179.195:9876");  
producer.setInstanceName("producer");  
producer.start();  
try \{  
for (int i = 0; i < 10; i++) \{  
Thread.sleep(1000); //每秒发送一次MQ  
Message msg = new Message("TopicA-test",// topic  
"TagA",// tag  
(new Date() + "Hello RocketMQ ,QuickStart" + i)  
.getBytes()// body  
);  
SendResult sendResult = producer.send(msg);  
\}  
\} catch (Exception e) \{  
e.printStackTrace();  
\}  
producer.shutdown();  
\}  
\}  
  
之后分别启动，Consumer效果如下  
  
这里需要注意，  
默认情况下，一台服务器只能启动一个Producer或Consumer实例，所以如果需要在一台服务器启动多个实例，需要设置实例的名称，如要再建一个producer：  
producer.setNamesrvAddr(“121.42.179.195:9876”);  
producer.setInstanceName(“Producer2”);  
  
原文：https://blog.csdn.net/coffeesweet/article/details/52451860

转载于:https://www.cnblogs.com/dywen/p/10450296.html