最近在看消息隊列框架 ,alibaba的RocketMQ單機支持1萬以上的持久化隊列,支持諸多特性,java
目前RocketMQ在阿里集團被普遍應用在訂單,交易,充值,流計算,消息推送,日誌流式處理,binglog分發等場景linux
比kafka仍是有過之無不及,其實kafka文檔很豐富git
但RocketMQ網上的文章太少,找不到相關的操做教程github
因而研究了下源碼 作個單機操做的教程,若是你也對此有興趣不妨共同研究sql
下載源碼的地址 https://github.com/alibaba/RocketMQ/releasesmarkdown
首選經過在java項目裏面Maven依賴方式引用RocketMQ Java SDKmybatis
<dependency> <groupId>com.alibaba.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>3.2.6</version> </dependency>
在linux 下用wget 下載源碼而後解壓出來框架
在runserver.sh裏面能夠配置 jvm啓動的參數 JAVA_OPT_1="-server -Xms4g -Xmx4g -Xmn2g -XX:PermSize=128m -XX:MaxPermSize=320m" 異步
能夠 vi runserver.shjvm
分別給 mqnamesrv mqbroker play.sh 執行的權限
chmod +x mqnamersrv
chmod +x mqbroker
chmod +x play.sh
下面紅線框的這段 命令輸入錯誤了,忽略不用看
經過 nohup sh mqnamesrv& 啓動 RocketMq
目前沒看到結束的命令,也沒找到相關的介紹,
我這裏用的 ps -ef|grep rocketmq 查到進程pid
而後kill pid號
或則pkill -9 java [慎用]
用jps -v 查看下java進程的參數
rocketmq啓動後監聽 9876端口,這裏仍是在看源碼裏面看到的,資料實在是太少了
在防火牆配置裏面加上 9876端口,設置iptables對外開放
部署Broker
nohup sh mqbroker -n "127.0.0.1:9876" -c ../conf/2m-2s-async/broker-a.properties &
這裏ip換成本機的就是單機實例,若是配置主從 這裏能夠配其餘的ip
Master和Slave的配置文件參考conf目錄下的配置文件
Master與Slave經過指定相同的brokerName參數來配對,Master的BrokerId必須是0,Slave的BrokerId必須是大於0的數
一個Master下面能夠掛載多個Slave,同一Master下的多個Slave經過指定不一樣的BrokerId來區分
部署一Master一Slave,集羣採用異步複製方式:
Master: nohup sh mqbroker -n "192.168.1.23:9876" -c ../conf/2m-2s-async/broker-a.properties &
Slave: nohup sh mqbroker -n "192.168.1.23:9876" -c ../conf/2m-2s-async/broker-a-s.properties &
package com.pgsqlmybatis.common.rocketmq;/* *************************************************************** * 公司名稱 : * 系統名稱 :信用管家專業版 * 類 名 稱 :Ios渠道idfa統計,推廣統計用 * 功能描述 : * 業務描述 : * 做 者 名 :@Author Royal * 開發日期 :2016-05-15 * Created :IntelliJ IDEA *************************************************************** * 修改日期 : * 修 改 者 : * 修改內容 : *************************************************************** */ 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) { DefaultMQProducer producer = new DefaultMQProducer("Producer"); producer.setNamesrvAddr("xxxxxxxxxx:9876"); try { producer.start(); String pushMsg="kafka activeMq rocketMq 消息隊列使用1"; Message msg = new Message("PushTopic","push","1", pushMsg.getBytes("UTF-8")); SendResult result = producer.send(msg); System.out.println("id:" + result.getMsgId() + " result:" + result.getSendStatus()); String pushMsg2="海量級消息記錄單機測試2"; msg = new Message("PushTopic","push","2",pushMsg2.getBytes("UTF-8")); result = producer.send(msg); System.out.println("id:" + result.getMsgId() + " result:" + result.getSendStatus()); String pushMsg3="海量級消息記錄單機測試3"; msg = new Message("PullTopic","pull","1",pushMsg3.getBytes()); result = producer.send(msg); System.out.println("id:" + result.getMsgId() + " result:" + result.getSendStatus()); } catch (Exception e) { e.printStackTrace(); } finally { producer.shutdown(); } } }
啓動生成者
啓動消費者
package com.pgsqlmybatis.common.rocketmq;/* *************************************************************** * 公司名稱 : * 系統名稱 :信用管家專業版 * 類 名 稱 :Ios渠道idfa統計,推廣統計用 * 功能描述 : * 業務描述 : * 做 者 名 :@Author Royal * 開發日期 :2016-05-15 * Created :IntelliJ IDEA *************************************************************** * 修改日期 : * 修 改 者 : * 修改內容 : *************************************************************** */ import java.io.UnsupportedEncodingException; 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.common.consumer.ConsumeFromWhere; import com.alibaba.rocketmq.common.message.Message; import com.alibaba.rocketmq.common.message.MessageExt; public class Consumer { public static void main(String[] args){ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("PushConsumer"); consumer.setNamesrvAddr("xxxxxxxxxxxx:9876"); try { consumer.subscribe("PushTopic", "push"); /** * 設置Consumer第一次啓動是從隊列頭部開始消費仍是隊列尾部開始消費<br> * 若是非第一次啓動,那麼按照上次消費的位置繼續消費 */ consumer.setConsumeFromWhere( ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer.registerMessageListener( new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage( List<MessageExt> list, ConsumeConcurrentlyContext Context) { Message msg = list.get(0); System.out.println(msg.toString()); String recString= null; try { recString = new String(msg.getBody() ,"UTF-8"); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println(recString); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } } ); consumer.start(); } catch (Exception e) { e.printStackTrace(); } } }
以上爲單機實例配置
若是你遇到什麼問題能夠私信我,若是以爲此文對你頗有幫助,點下贊推薦下額^_^
參考:http://blog.csdn.net/a19881029/article/details/34446629
http://sofar.blog.51cto.com/353572/1540874
http://blog.csdn.net/loongshawn/article/details/51086876