版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/pandajava/article/details/50833878
1.
rocketMQ3.2.6安装 https://github.com/alibaba/RocketMQ执行nohup sh mqbroker -n 127.0.0.1:9876 &
报错 java.lang.ExceptionInInitializerError
Caused by: java.net.UnknownHostException: xxxxxxx: 未知的名称或服务
解决方案:http://www.easytd.com/yichangjijin/d_16012708270.html
vi /proc/sys/kernel/hostname
或者
sysctl kernel.hostname=localhost
2.
nohup sh mqnamesrv -n 127.0.0.1:9876 &
The Name Server boot success.
nohup sh mqbroker -n 127.0.0.1:9876 &
The broker[localhost, 192.xxxx.xxx.128:10911] boot success. and name server is 127.0.0.1:9876
进程:
root 16148 15609 0 12:46 pts/1 00:00:00 sh mqnamesrv -n 127.0.0.1:9876
root 16152 16148 0 12:46 pts/1 00:00:00 sh /usr/local/rocketMq/alibaba-rocketmq/bin/runserver.sh com.alibaba.rocketmq.namesrv.NamesrvStartup -n 127.0.0.1:9876
root 16154 16152 0 12:46 pts/1 00:00:37 /usr/local/jdk/jdk1.7/bin/java -server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:+DisableExplicitGC -verbose:gc -Xloggc:/root/rmq_srv_gc.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -Djava.ext.dirs=/usr/local/rocketMq/alibaba-rocketmq/bin/../lib -cp .:/usr/local/rocketMq/alibaba-rocketmq/bin/../conf:.:/usr/local/jdk/jdk1.7/lib/dt.jar:/usr/local/jdk/jdk1.7/lib/tools.jar com.alibaba.rocketmq.namesrv.NamesrvStartup -n 127.0.0.1:9876
root 16290 15609 0 12:54 pts/1 00:00:00 sh mqbroker -n 127.0.0.1:9876
root 16294 16290 0 12:54 pts/1 00:00:00 sh /usr/local/rocketMq/alibaba-rocketmq/bin/runbroker.sh com.alibaba.rocketmq.broker.BrokerStartup -n 127.0.0.1:9876
root 16297 16294 5 12:54 pts/1 00:13:55 /usr/local/jdk/jdk1.7/bin/java -server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:+DisableExplicitGC -verbose:gc -Xloggc:/root/rmq_bk_gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:-OmitStackTraceInFastThrow -Djava.ext.dirs=/usr/local/rocketMq/alibaba-rocketmq/bin/../lib -cp .:/usr/local/rocketMq/alibaba-rocketmq/bin/../conf:.:/usr/local/jdk/jdk1.7/lib/dt.jar:/usr/local/jdk/jdk1.7/lib/tools.jar com.alibaba.rocketmq.broker.BrokerStartup -n 127.0.0.1:9876
3
package com.pjm.mqstudy.recoetmq;
import java.util.concurrent.TimeUnit;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
public class Producer {
public static void main(String[] args) throws MQClientException,
InterruptedException {
/**
* 一个应用创建一个Producer,由应用来维护此对象,可以设置为全局对象或者单例<br>
* 注意:ProducerGroupName需要由应用来保证唯一<br>
* ProducerGroup这个概念发送普通的消息时,作用不大,但是发送分布式事务消息时,比较关键,
* 因为服务器会回查这个Group下的任意一个Producer
*/
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("192.1xx.xxx.1xx:9876");
producer.setInstanceName("Producer");
/**
* Producer对象在使用之前必须要调用start初始化,初始化一次即可<br>
* 注意:切记不可以在每次发送消息时,都调用start方法
*/
producer.start();
/**
* 下面这段代码表明一个Producer对象可以发送多个topic,多个tag的消息。
* 注意:send方法是同步调用,只要不抛异常就标识成功。但是发送成功也可会有多种状态,<br>
* 例如消息写入Master成功,但是Slave不成功,这种情况消息属于成功,但是对于个别应用如果对消息可靠性要求极高,<br>
* 需要对这种情况做处理。另外,消息可能会存在发送失败的情况,失败重试由应用来处理。
*/
for (int i = 0; i < 100; i++) {
try {
{
Message msg = new Message("TopicTest1",// topic
"TagA",// tag
"OrderID001",// key
("Hello MetaQ").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest2",// topic
"TagB",// tag
"OrderID0034",// key
("Hello MetaQ").getBytes());// body
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
{
Message msg = new Message("TopicTest3",// topic
"TagC",// tag
"OrderID061",// key
("Hello MetaQ").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();
}
}
package com.pjm.mqstudy.recoetmq;
import java.util.List;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
public class PushConsumer {
/**
* 当前例子是PushConsumer用法,使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br>
* 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息,然后再回调用户Listener方法<br>
*/
public static void main(String[] args) throws InterruptedException,
MQClientException {
/**
* 一个应用创建一个Consumer,由应用来维护此对象,可以设置为全局对象或者单例<br>
* 注意:ConsumerGroupName需要由应用来保证唯一
*/
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
"ConsumerGroupName");
consumer.setNamesrvAddr("192.1xx.xxx.xxx:9876");
consumer.setInstanceName("Consumber");
/**
* 订阅指定topic下tags分别等于TagA或TagC或TagD
*/
consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
/**
* 订阅指定topic下所有消息<br>
* 注意:一个consumer对象可以订阅多个topic
*/
consumer.subscribe("TopicTest2", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
/**
* 默认msgs里只有一条消息,可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
*/
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 && msg.getTags().equals("TagA")) {
// 执行TagA的消费
System.out.println(new String(msg.getBody()));
} else if (msg.getTags() != null
&& msg.getTags().equals("TagC")) {
// 执行TagC的消费
} else if (msg.getTags() != null
&& msg.getTags().equals("TagD")) {
// 执行TagD的消费
}
} else if (msg.getTopic().equals("TopicTest2")) {
System.out.println(new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
/**
* Consumer对象在使用之前必须要调用start初始化,初始化一次即可<br>
*/
consumer.start();
System.out.println("Consumer Started.");
}
}
创建topic
[root@PjmLinuixCentOs6 bin]# sh mqadmin updateTopic -n 192.168.164.128:9876 -c DefaultCluster -t TopicTest1
create topic to 192.xxx.xxx.xxx:10911 success.
TopicConfig [topicName=TopicTest1, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false]
运行Producer程序,系统报错 no route info of this topic rocketmq
原因linux的系统防火墙没有对外开放相关端口,停用防火墙
1) 重启后生效
开启: chkconfig iptables on
关闭: chkconfig iptables off
2) 即时生效,重启后失效
开启: service iptables start
关闭: service iptables stop
再次运行Producer程序:
8:04:36.789 [NettyClientWorkerThread_1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.leakDetectionLevel: simple
18:04:37.001 [NettyClientSelector_1] DEBUG io.netty.util.internal.Cleaner0 - java.nio.ByteBuffer.cleaner(): available
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F0000000000000000, messageQueue=MessageQueue [topic=TopicTest1, brokerName=localhost, queueId=0], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F0000000000000094, messageQueue=MessageQueue [topic=TopicTest2, brokerName=localhost, queueId=0], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F0000000000000129, messageQueue=MessageQueue [topic=TopicTest3, brokerName=localhost, queueId=0], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F00000000000001BD, messageQueue=MessageQueue [topic=TopicTest1, brokerName=localhost, queueId=1], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F0000000000000251, messageQueue=MessageQueue [topic=TopicTest2, brokerName=localhost, queueId=1], queueOffset=0]
SendResult [sendStatus=SEND_OK, msgId=C0A8A48000002A9F00000000000002E6, messageQueue=MessageQueue [topic=TopicTest3, brokerName=localhost, queueId=1], queueOffset=0]
.....
运行Consumer程序:
Consumer Started.
ConsumeMessageThread_1 Receive New Messages: 1
ConsumeMessageThread_2 Receive New Messages: 1
Hello MetaQ
Hello MetaQ
ConsumeMessageThread_3 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_4 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_5 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_6 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_7 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_8 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_9 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_10 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_11 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_12 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_13 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_17 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_19 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_14 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_16 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_18 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_15 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_1 Receive New Messages: 1
Hello MetaQ
ConsumeMessageThread_20 Receive New Messages: 1
Hello MetaQ
5搭建web监控界面
https://github.com/rocketmq/rocketmq-console
config.properties配置 mq地址端口
Tomcat启动找不到maven里的类 抛ClassNotFoundException参考 http://www.yihaomen.com/article/java/471.htm
1. 右键单击工程项目 ->点击 properties
2. 选择 Deployment Assembly
3. 点击 Add -> Java Build Path Entries -> Next
4. 选择 Maven Dependencies -> Finish -> Apply -> OK
5. Clean project and server. 重启server