Centos 安装 RocketMq

官网和网上的各种教程都有坑，这里提供楼主亲测安装教程

1、下载安装

前提已经安装java环境，这里不再赘述 http://rocketmq.apache.org/docs/quick-start/

从官网下载安装包

wget http://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.4.0/rocketmq-all-4.4.0-source-release.zip

安装maven

wget http://repos.fedorapeople.org/repos/dchen/apache-maven/epel-apache-maven.repo -O /etc/yum.repos.d/epel-apache-maven.repo yum -y install apache-maven

安装unzip

yum install unzip

安装RocketMq

unzip rocketmq-all-4.4.0-source-release.zip cd rocketmq-all-4.4.0/ mvn -Prelease-all -DskipTests clean install -U cd distribution/target/apache-rocketmq

2、启动 Name Server

nohup sh mqnamesrv -n localhost:9876 > ./srv.log & #或使用外网ip，配置日志到srv.log

调低Rocketmq使用内存，默认8g，内存不够会报错

vim bin/runbroker.sh #修改对应的内存配置 #-server -Xms8g -Xmx8g -Xmn4g -server -Xms256m -Xmx256m -Xmn128m

配置Broker,

vim conf/brokder.conf brokerClusterName = DefaultCluster brokerName = broker-a brokerId = 0 deleteWhen = 04 fileReservedTime = 48 brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSH namesrvAddr = 127.0.0.1:9876 brokerIP1 = 公网IP #使用内网IP，无法远程访问 autoCreateTopicEnable = true

启动Broker

nohup sh bin/mqbroker -c conf/broker.conf -n 127.0.0.1:9876 > ./broker.log &

2、配置监控平台 rocketmq_console

#使用git命令下载项目源码，由于我们仅需要rocketmq-console，故下载此项目对应分支即可。 git clone -b release-rocketmq-console-1.0.0 https://github.com/apache/rocketmq-externals.git

进入项目文件夹并修改配置文件

$ cd rocketmq-externals/rocketmq-console/ $ vi src/main/resources/application.properties #管理后台访问上下文路径，默认为空，如果填写，一定要前面加“/”，后面不要加，否则启动报错 server.contextPath=/rocketmq #访问端口 server.port=8080 #spring.application.index=true spring.application.name=rocketmq-console spring.http.encoding.charset=UTF-8 spring.http.encoding.enabled=true spring.http.encoding.force=true #logback配置文件路径 logging.config=classpath:logback.xml #if this value is empty,use env value rocketmq.config.namesrvAddr NAMESRV_ADDR | now, you can set it in ops page.default localhost:9876 #Name Server地址，修改成你自己的服务地址 rocketmq.config.namesrvAddr=10.0.74.198:9876 #集群10.0.74.198:9876 ;10.0.74.198:9876 #if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true rocketmq.config.isVIPChannel= #rocketmq-console's data path:dashboard/monitor rocketmq.config.dataPath=/tmp/rocketmq-console/data #set it false if you don't want use dashboard.default true rocketmq.config.enableDashBoardCollect=true

Name Server地址默认为空，注释说可以在启动项目后在后台配置，经测试，后台配置切换失败，有报错，所以打包前需修改配置文件明确给出Name Server地址，或者启动服务的时候给出rocketmq.config.namesrvAddr参数值。

将项目打成jar包，并运行jar文件。

$ mvn clean package -Dmaven.test.skip=true $ java -jar target/rocketmq-console-ng-1.0.0.jar #如果配置文件没有填写Name Server $ java -jar target/rocketmq-console-ng-1.0.0.jar --rocketmq.config.namesrvAddr='10.0.74.198:9876'

启动成功后，访问地址http://localhost:8080/rocketmq, 即可进入管理后台操作。

3、java测试代码

maven:

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

生产者SyncProducer

public class SyncProducer { public static void main(String[] args) throws Exception { //Instantiate with a producer group name. DefaultMQProducer producer = new DefaultMQProducer("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(); } }

消费者AsyncProducer

public class AsyncProducer { public static void main(String[] args) throws Exception { //Instantiate with a producer group name. DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); // Specify name server addresses. producer.setNamesrvAddr("localhost:9876"); //Launch the instance. producer.start(); producer.setRetryTimesWhenSendAsyncFailed(0); for (int i = 0; i < 100; i++) { final int index = i; //Create a message instance, specifying topic, tag and message body. Message msg = new Message("TopicTest", "TagA", "OrderID188", "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET)); producer.send(msg, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId()); } @Override public void onException(Throwable e) { System.out.printf("%-10d Exception %s %n", index, e); e.printStackTrace(); } }); } //Shut down once the producer instance is not longer in use. producer.shutdown(); } }