RocketMQ在Windows平台下环境搭建.docx
《RocketMQ在Windows平台下环境搭建.docx》由会员分享，可在线阅读，更多相关《RocketMQ在Windows平台下环境搭建.docx（17页珍藏版）》请在冰点文库上搜索。
RocketMQ在Windows平台下环境搭建
一.环境搭建
需要jdk1.6（以上）64bit，maven，eclipse
二.RocketMQ项目下载
项目地址：
三.将RocketMQ-master导入到eclipse中
1将项目导入eclipse，如下图
2.在我下载的RocketMQ-master的pom.xml文件的parent有个问题，默认如下：
2
3com.taobao
4parent
51.0.2
6
7
8
--
9org.sonatype.oss
10oss-parent
117
12-->
13
但是编译时总报错parent找不到，而用下面的parent，则编译通过。
14
--
15com.taobao
16parent
171.0.2
18-->
19
20
21org.sonatype.oss
22oss-parent
237
24
3.由于我用的是jdk1.7，故修改RocketMQ-master.pom的jdk版本
25
26UTF-8
27
--mavenproperties-->
28true
29true
30true
31
--compilersettingsproperties-->
321.7
331.7
34UTF-8
35
四.编译RocketMQ项目
1.在命令行执行在RocketMQ-master文件夹下的install.bat批处理
2.该命令会编译整个项目，并在RocketMQ-master目录下生成一个target文件夹
3.进入刚生成的target文件夹下的bin目录，在命令行中执行：
startmqnamesrv.exe，会弹出一个信息窗口，显示ThenameServerbootsuccess说明启动成功了，接着启动borker，在命令行中执行：
startmqbroker.exe-n127.0.0.1:
9876同样的弹出一个窗口，看到success表示成功了。
五.启动Producer和Customer
1.在RocketMQ-example项目中加入Producer.java
36publicclassProducer{
37publicstaticvoidmain（String[]args）throwsMQClientException,
38InterruptedException{
39/**
40*一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例
41*注意：
ProducerGroupName需要由应用来保证唯一
42*ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
43*因为服务器会回查这个Group下的任意一个Producer
44*/
45finalDefaultMQProducerproducer=newDefaultMQProducer（"ProducerGroupName"）;
46producer.setNamesrvAddr（"127.0.0.1:
9876"）;
47producer.setInstanceName（"Producer"）;
48
49/**
50*Producer对象在使用之前必须要调用start初始化，初始化一次即可
51*注意：
切记不可以在每次发送消息时，都调用start方法
52*/
53producer.start（）;
54
55/**
56*下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
57*注意：
send方法是同步调用，只要不抛异常就标识成功。
但是发送成功也可会有多种状态，
58*例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，
59*需要对这种情况做处理。
另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
60*/
61for（inti=0;i<10;i++）{
62try{
63{
64Messagemsg=newMessage（"TopicTest1",//topic
65"TagA",//tag
66"OrderID001",//key
67（"HelloMetaQA"）.getBytes（））;//body
68SendResultsendResult=producer.send（msg）;
69System.out.println（sendResult）;
70}
71
72{
73Messagemsg=newMessage（"TopicTest2",//topic
74"TagB",//tag
75"OrderID0034",//key
76（"HelloMetaQB"）.getBytes（））;//body
77SendResultsendResult=producer.send（msg）;
78System.out.println（sendResult）;
79}
80
81{
82Messagemsg=newMessage（"TopicTest3",//topic
83"TagC",//tag
84"OrderID061",//key
85（"HelloMetaQC"）.getBytes（））;//body
86SendResultsendResult=producer.send（msg）;
87System.out.println（sendResult）;
88}
89}catch（Exceptione）{
90e.printStackTrace（）;
91}
92TimeUnit.MILLISECONDS.sleep（1000）;
93}
94
95/**
96*应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
97*注意：
我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
98*/
99//producer.shutdown（）;
100Runtime.getRuntime（）.addShutdownHook（newThread（newRunnable（）{
101publicvoidrun（）{
102producer.shutdown（）;
103}
104}））;
105System.exit（0）;
106}
107}
2.加入Customer.java
108publicclassConsumer{
109/**
110*当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。
111*但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法
112*/
113publicstaticvoidmain（String[]args）throwsInterruptedException,
114MQClientException{
115/**
116*一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例
117*注意：
ConsumerGroupName需要由应用来保证唯一
118*/
119DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer（
120"ConsumerGroupName"）;
121consumer.setNamesrvAddr（"127.0.0.1:
9876"）;
122consumer.setInstanceName（"Consumber"）;
123
124/**
125*订阅指定topic下tags分别等于TagA或TagC或TagD
126*/
127consumer.subscribe（"TopicTest1","TagA||TagC||TagD"）;
128/**
129*订阅指定topic下所有消息
130*注意：
一个consumer对象可以订阅多个topic
131*/
132consumer.subscribe（"TopicTest2","*"）;
133
134consumer.registerMessageListener（newMessageListenerConcurrently（）{
135
136publicConsumeConcurrentlyStatusconsumeMessage（
137Listmsgs,ConsumeConcurrentlyContextcontext）{
138
139System.out.println（Thread.currentThread（）.getName（）
140+"ReceiveNewMessages:
"+msgs.size（））;
141
142MessageExtmsg=msgs.get（0）;
143if（msg.getTopic（）.equals（"TopicTest1"））{
144//执行TopicTest1的消费逻辑
145if（msg.getTags（）!
=null&&msg.getTags（）.equals（"TagA"））{
146//执行TagA的消费
147System.out.println（newString（msg.getBody（）））;
148}elseif（msg.getTags（）!
=null
149&&msg.getTags（）.equals（"TagC"））{
150//执行TagC的消费
151System.out.println（newString（msg.getBody（）））;
152}elseif（msg.getTags（）!
=null
153&&msg.getTags（）.equals（"TagD"））{
154//执行TagD的消费
155System.out.println（newString（msg.getBody（）））;
156}
157}elseif（msg.getTopic（）.equals（"TopicTest2"））{
158System.out.println（newString（msg.getBody（）））;
159}
160
161returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
162
163}
164}）;
165
166/**
167*Consumer对象在使用之前必须要调用start初始化，初始化一次即可
168*/
169consumer.start（）;
170
171System.out.println（"ConsumerStarted."）;
172}
173}
3.运行Producer
17419:
50:
57.233[main]DEBUGi.n.u.i.l.InternalLoggerFactory-UsingSLF4Jasthedefaultloggingframework
17519:
50:
57.259[main]DEBUGi.n.c.MultithreadEventLoopGroup--Dty.eventLoopThreads:
4
17619:
50:
57.274[main]DEBUGi.n.util.internal.PlatformDependent0-java.nio.Buffer.address:
available
17719:
50:
57.274[main]DEBUGi.n.util.internal.PlatformDependent0-sun.misc.Unsafe.theUnsafe:
available
17819:
50:
57.274[main]DEBUGi.n.util.internal.PlatformDependent0-sun.misc.Unsafe.copyMemory:
available
17919:
50:
57.275[main]DEBUGi.n.util.internal.PlatformDependent0-java.nio.Bits.unaligned:
true
18019:
50:
57.275[main]DEBUGi.n.util.internal.PlatformDependent-Platform:
Windows
18119:
50:
57.276[main]DEBUGi.n.util.internal.PlatformDependent-Javaversion:
7
18219:
50:
57.276[main]DEBUGi.n.util.internal.PlatformDependent--Dty.noUnsafe:
false
18319:
50:
57.276[main]DEBUGi.n.util.internal.PlatformDependent-sun.misc.Unsafe:
available
18419:
50:
57.277[main]DEBUGi.n.util.internal.PlatformDependent--Dty.noJavassist:
false
18519:
50:
57.506[main]DEBUGi.n.util.internal.PlatformDependent-Javassist:
available
18619:
50:
57.507[main]DEBUGi.n.util.internal.PlatformDependent--Dty.tmpdir:
C:
\Users\tannj\AppData\Local\Temp（java.io.tmpdir）
18719:
50:
57.507[main]DEBUGi.n.util.internal.PlatformDependent--Dty.bitMode:
64（sun.arch.data.model）
18819:
50:
57.507[main]DEBUGi.n.util.internal.PlatformDependent--Dty.noPreferDirect:
false
18919:
50:
57.544[main]DEBUGty.channel.nio.NioEventLoop--Dty.noKeySetOptimization:
false
19019:
50:
57.544[main]DEBUGty.channel.nio.NioEventLoop--Dty.selectorAutoRebuildThreshold:
512
19119:
50:
57.875[main]DEBUGi.n.util.internal.ThreadLocalRandom--Dty.initialSeedUniquifier:
0xfaf4f653b3d8be50（took52ms）
19219:
50:
57.927[main]DEBUGty.buffer.ByteBufUtil--Dty.allocator.type:
unpooled
19319:
50:
57.927[main]DEBUGty.buffer.ByteBufUtil--Dty.threadLocalDirectBufferSize:
65536
19419:
50:
57.958[NettyClientSelector_1]DEBUGi.n.u.i.JavassistTypeParameterMatcherGenerator-Generated:
ty.util.internal.__matchers__.com.alibaba.rocketmq.remoting.protocol.RemotingCommandMatcher
19519:
50:
58.006[main]DEBUGty.util.Recycler--Dty.recycler.maxCapacity.default:
262144
19619:
50:
58.027[NettyClientWorkerThread_1]DEBUGty.util.ResourceLeakDetector--Dty.leakDetectionLevel:
simple
19719:
50:
58.255[NettyClientSelector_1]DEBUGty.util.internal.Cleaner0-java.nio.ByteBuffer.cleaner（）:
available
198SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A36A,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=0],queueOffset=73578]
199SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A3F2,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=1],queueOffset=73577]
200SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A47A,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=2],queueOffset=73577]
201SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A502,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=3],queueOffset=73576]
202SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A58A,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=0],queueOffset=73579]
203SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A612,messageQueue=MessageQueue[topic=TopicTest,brokerName=tannj-PC,queueId=1],queueOffset=73578]
204SendResult[sendStatus=SEND_OK,msgId=0A016F9600002A9F000000000272A69A,messageQueue=MessageQueue[
升级会员 