声明：本文转载自https://my.oschina.net/grkj/blog/1596142，转载目的在于传递更多信息，仅供学习交流之用。如有侵权行为，请联系我，我会及时删除。
最近公司的终端机需要接通推送，然后查看对比了一下，觉得rocketMq更合适一点，所以选择了rocketMq进行测试。
a、git（yum intsall git）；b、jdk1.8； c、maven； d、rocketmq
1、首先下载rocketMq3.5.8
2、tar -zxvf rocketmq-3.5.8.tar.gz
3、进入rocketmq-3.5.8
4、sh install.sh，初始化rocketmq（这是初始化和下载东西，是必须项目）
5、下载完毕，执行 cd devenv(进入这个文件夹devenv)
6、编写rocket环境 （1）"ROCKETMQ_HOME=pwd" >> ~/.bash_profile
（2）source ~/.bash_profile
7、修改内存大小bin文件夹下面的runserver.sh文件夹的内存，从4G修改成128m和256m(因为我的测试服务器内存不大)
8、修改内存大小bin文件夹下面的runbroker.sh件夹的内存，从4G修改成128m和256m(因为我的测试服务器内存不大) 9、初始化并且授予权限
（1）cd /usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/bin
（2）授予执行权限 chmod +x mqadmin mqbroker mqfiltersrv mqnamesrv mqshutdown
10、运行namesrv服务（同时将日归类）
（1）nohup mqnamesrv 1>/usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/ng.log 2>/usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/ng-err.log &
（2）查看ng.log，如果出现 The Name Server boot success. serializeType=JSON，则表示nameServer开启成功
11、启动mqbroker
（1）nohup sh mqbroker -n ip:9876 autoCreateTopicEnable=true > /usr/local/rocketmq/rocketmq3.5.8/target/alibaba-rocketmq-broker/alibaba-rocketmq/log/broker.log 2>&1 &
（2）查看broker.log，如果出现 The Name Server boot success. serializeType=JSON，则表示nameServer开启成功
12、查看是否运行成功 ps aux | grep java
如果代码里面生产者和消费者加上这句话，就走端口10909，不加就走vip的10911，看服务器日子启动哪个端口而定加不加 producer.setVipChannelEnabled(false);
1、消费者代码
package cn.shopin.ssm.testRocketMq; 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.consumer.ConsumeFromWhere; import com.alibaba.rocketmq.common.message.MessageExt; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; /** * Created by Administrator on 2017/12/26. */ public class ConsumerTest01 { /** * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br> * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法<br> */ public static void main(String[] args) { /** * 注意：ConsumerGroupName需要由应用来保证唯一 */ DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("ProducerGroupName"); //pushConsumer.setNamesrvAddr("192.168.180.1:9876"); pushConsumer.setNamesrvAddr("ip:9876"); pushConsumer.setInstanceName("Consumer"); /** * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br> * 如果非第一次启动，那么按照上次消费的位置继续消费 */ pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); try { /** * 订阅指定topic下tags分别等于TagA或TagC或TagD * 两个参数：第一个参数是topic第二个参数是tags */ pushConsumer.subscribe("TopicTest1", "TagA || TagC || TagD"); pushConsumer.subscribe("TopicTest2", "*"); pushConsumer.subscribe("TopicTest3", "*"); /** * 订阅指定topic下所有消息<br> * 注意：一个consumer对象可以订阅多个topic */ //pushConsumer.subscribe("TopicTest2", "*"); final AtomicInteger count = new AtomicInteger(); pushConsumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) { System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size()); MessageExt messageExt = msgs.get(0); count.getAndIncrement(); if ("TopicTest1".equals(messageExt.getTopic())) { // 执行TopicTest1的消费逻辑 if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) { // 执行TagA的消费 System.out.println("TopicTest1的TagA:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) { System.out.println("TopicTest1的TagB:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) { System.out.println("TopicTest1的TagC:"+new String(messageExt.getBody())); } } else if ("TopicTest2".equals(messageExt.getTopic())) { // 执行TopicTest1的消费逻辑 if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) { // 执行TagA的消费 System.out.println("TopicTest2的TagA:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) { System.out.println("TopicTest2的TagB:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) { System.out.println("TopicTest2的TagC:"+new String(messageExt.getBody())); } } else if ("TopicTest3".equals(messageExt.getTopic())) { // 执行TopicTest1的消费逻辑 if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) { // 执行TagA的消费 System.out.println("TopicTest3的TagA:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) { System.out.println("TopicTest3的TagB:"+new String(messageExt.getBody())); } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) { System.out.println("TopicTest3的TagC:"+new String(messageExt.getBody())); } } System.out.println("当前的count的值为："+count); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); } catch (MQClientException e) { e.printStackTrace(); } /** * Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br> */ try { pushConsumer.start(); } catch (MQClientException e) { e.printStackTrace(); } System.out.println("Consumer Started."); } } 2、生产者代码
package cn.shopin.ssm.testRocketMq; import com.alibaba.rocketmq.client.exception.MQBrokerException; 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; import com.alibaba.rocketmq.remoting.exception.RemotingException; import java.util.concurrent.TimeUnit; /** * Created by Administrator on 2017/12/26. */ public class ProducerTest01 { public static void main(String[] args) { /** * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br> * 注意：ProducerGroupName需要由应用来保证唯一<br> * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键， * 因为服务器会回查这个Group下的任意一个Producer */ DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); //producer.setNamesrvAddr("192.168.180.1:9876"); producer.setNamesrvAddr("ip:9876"); producer.setInstanceName("Producer"); //producer.setVipChannelEnabled(false); /** * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br> * 注意：切记不可以在每次发送消息时，都调用start方法 */ try { producer.start(); } catch (MQClientException e) { e.printStackTrace(); } for (int i = 0; i < 10; i++) { try { /** * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。 * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br> * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br> * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。 */ { Message msg = new Message("TopicTest1",// topic "TagA",// tag "OrderID001",// key ("按一下").getBytes());// body SendResult sendResult = producer.send(msg); System.out.println("按一下:"+sendResult); } { Message msg = new Message("TopicTest2", "TagB", "OrderID001", ("Hello MetaQ TagB".getBytes())); SendResult sendResult = producer.send(msg); System.out.println("Hello MetaQ TagB"+sendResult); } { Message msg = new Message("TopicTest3", "TagC", "OrderID001", ("Hello MetaQ TagC").getBytes()); SendResult sendResult = producer.send(msg); System.out.println("Hello MetaQ TagC"+sendResult); } TimeUnit.MILLISECONDS.sleep(1000); } catch (MQClientException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (RemotingException e) { e.printStackTrace(); } catch (MQBrokerException e) { e.printStackTrace(); } } /** * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己 * 注意：我们建议应用在JBOSS、Tomcat等容器的退出销毁方法里调用shutdown方法 */ producer.shutdown(); } } 
本文发表于2017年12月28日 18:32
(c)注：本文转载自https://my.oschina.net/grkj/blog/1596142，转载目的在于传递更多信息，并不代表本网赞同其观点和对其真实性负责。如有侵权行为，请联系我们，我们会及时删除.
阅读 3163 讨论 0 喜欢 0
Spring Cloud Bus + RabbitMq 自动刷新
selenium中如何测试360等基于chrome内核的浏览器
分布式环境中使用基于共享存储的LVM块设备时的元数据同步方案对比
避坑！MotionGo就是一款诈骗软件！千万不要用ChatPPT，浪费时间！
宝塔面板使用WWW用户执行计划任务命令 解决laravel日志权限问题 宝塔设置计划任务执行用户
手机微信一键制作透明底文字表情 教你如何让微信表情包背景为透明 自制微信表情包动图 文字表情在线制作
ssh目录权限配置指南 .ssh文件夹及rsa_id.pub等文件正确权限规则
Mac连接亚马逊AWS EC2 远程连接登录不上去 有pem私钥文件依然要输密码 默认密码 教程
PHP工程师必知必会 - Linux系统常用命令 - Linux中的网络管理命令（下）
| 抢先体验 |
|---|
扫码体验 |
| 闪念胶囊 |
|---|
|
万稳万当，不如一默。任何一句话，你不说出来便是那句话的主人，你说了出来，便是那句话的奴隶。 18:22 04月20日 查看详情 |
|
你要过得好哇，这样我才能恨你啊，你要是过得不好，我都不知道该恨你还是拥抱你啊。 17:21 2021年04月19日 查看详情 |
|
直抵黄龙府，与诸君痛饮尔。 18:17 2021年03月28日 查看详情 |
|
那时陪伴我的人啊，你们如今在何方。 16:28 2021年03月19日 查看详情 |
|
不出意外的话，我们再也不会见了，祝你前程似锦。 18:05 2021年03月17日 查看详情 |
| Contact |
|---|
Y2lvbkBjaGluYWNpb24uY24= |