Linux的centos7.2部署rocketMq3.5.8

背景

最近公司的终端机需要接通推送，然后查看对比了一下，觉得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、消费者代码

1package cn.shopin.ssm.testRocketMq;
2
3import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
4import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
5import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
6import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
7import com.alibaba.rocketmq.client.exception.MQClientException;
8import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
9import com.alibaba.rocketmq.common.message.MessageExt;
10
11import java.util.List;
12import java.util.concurrent.atomic.AtomicInteger;
13
14/**
15 * Created by Administrator on 2017/12/26.
16 */
17public class ConsumerTest01 {
18    /**
19     * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。<br>
20     * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法<br>
21     */
22    public static void main(String[] args) {
23
24        /**
25         * 注意：ConsumerGroupName需要由应用来保证唯一
26         */
27        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("ProducerGroupName");
28        //pushConsumer.setNamesrvAddr("192.168.180.1:9876");
29        pushConsumer.setNamesrvAddr("ip:9876");
30        pushConsumer.setInstanceName("Consumer");
31        /**
32         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
33         * 如果非第一次启动，那么按照上次消费的位置继续消费
34         */
35        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
36        try {
37            /**
38             * 订阅指定topic下tags分别等于TagA或TagC或TagD
39             * 两个参数：第一个参数是topic第二个参数是tags
40             */
41            pushConsumer.subscribe("TopicTest1", "TagA || TagC || TagD");
42            pushConsumer.subscribe("TopicTest2", "*");
43            pushConsumer.subscribe("TopicTest3", "*");
44            /**
45             * 订阅指定topic下所有消息<br>
46             * 注意：一个consumer对象可以订阅多个topic
47             */
48            //pushConsumer.subscribe("TopicTest2", "*");
49            final AtomicInteger count = new AtomicInteger();
50            pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
51                @Override
52                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
53                                                                ConsumeConcurrentlyContext consumeConcurrentlyContext) {
54                    System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs.size());
55                    MessageExt messageExt = msgs.get(0);
56                    count.getAndIncrement();
57                    if ("TopicTest1".equals(messageExt.getTopic())) {
58                        // 执行TopicTest1的消费逻辑
59                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
60                            // 执行TagA的消费
61                            System.out.println("TopicTest1的TagA:"+new String(messageExt.getBody()));
62                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
63                            System.out.println("TopicTest1的TagB:"+new String(messageExt.getBody()));
64                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
65                            System.out.println("TopicTest1的TagC:"+new String(messageExt.getBody()));
66                        }
67                    } else if ("TopicTest2".equals(messageExt.getTopic())) {
68                        // 执行TopicTest1的消费逻辑
69                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
70                            // 执行TagA的消费
71                            System.out.println("TopicTest2的TagA:"+new String(messageExt.getBody()));
72                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
73                            System.out.println("TopicTest2的TagB:"+new String(messageExt.getBody()));
74                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
75                            System.out.println("TopicTest2的TagC:"+new String(messageExt.getBody()));
76                        }
77                    } else if ("TopicTest3".equals(messageExt.getTopic())) {
78                        // 执行TopicTest1的消费逻辑
79                        if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
80                            // 执行TagA的消费
81                            System.out.println("TopicTest3的TagA:"+new String(messageExt.getBody()));
82                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
83                            System.out.println("TopicTest3的TagB:"+new String(messageExt.getBody()));
84                        } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagC")) {
85                            System.out.println("TopicTest3的TagC:"+new String(messageExt.getBody()));
86                        }
87                    }
88                    System.out.println("当前的count的值为："+count);
89                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
90                }
91            });
92        } catch (MQClientException e) {
93            e.printStackTrace();
94        }
95        /**
96         * Consumer对象在使用之前必须要调用start初始化，初始化一次即可<br>
97         */
98        try {
99            pushConsumer.start();
100        } catch (MQClientException e) {
101            e.printStackTrace();
102        }
103        System.out.println("Consumer Started.");
104    }
105}

2、生产者代码

1package cn.shopin.ssm.testRocketMq;
2
3import com.alibaba.rocketmq.client.exception.MQBrokerException;
4import com.alibaba.rocketmq.client.exception.MQClientException;
5import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
6import com.alibaba.rocketmq.client.producer.SendResult;
7import com.alibaba.rocketmq.common.message.Message;
8import com.alibaba.rocketmq.remoting.exception.RemotingException;
9
10import java.util.concurrent.TimeUnit;
11
12/**
13 * Created by Administrator on 2017/12/26.
14 */
15public class ProducerTest01 {
16    public static void main(String[] args) {
17        /**
18         * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
19         * 注意：ProducerGroupName需要由应用来保证唯一<br>
20         * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
21         * 因为服务器会回查这个Group下的任意一个Producer
22         */
23        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
24        //producer.setNamesrvAddr("192.168.180.1:9876");
25        producer.setNamesrvAddr("ip:9876");
26        producer.setInstanceName("Producer");
27        //producer.setVipChannelEnabled(false);
28        /**
29         * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
30         * 注意：切记不可以在每次发送消息时，都调用start方法
31         */
32        try {
33            producer.start();
34        } catch (MQClientException e) {
35            e.printStackTrace();
36        }
37        for (int i = 0; i < 10; i++) {
38            try {
39                /**
40                 * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
41                 * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br>
42                 * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br>
43                 * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
44                 */
45                {
46                    Message msg = new Message("TopicTest1",// topic
47                            "TagA",// tag
48                            "OrderID001",// key
49                            ("按一下").getBytes());// body
50                    SendResult sendResult = producer.send(msg);
51                    System.out.println("按一下:"+sendResult);
52                }
53
54                {
55                    Message msg = new Message("TopicTest2",
56                            "TagB",
57                            "OrderID001",
58                            ("Hello MetaQ TagB".getBytes()));
59
60                    SendResult sendResult = producer.send(msg);
61                    System.out.println("Hello MetaQ TagB"+sendResult);
62                }
63
64                {
65                    Message msg = new Message("TopicTest3",
66                            "TagC",
67                            "OrderID001",
68                            ("Hello MetaQ TagC").getBytes());
69
70                    SendResult sendResult = producer.send(msg);
71                    System.out.println("Hello MetaQ TagC"+sendResult);
72                }
73
74                TimeUnit.MILLISECONDS.sleep(1000);
75
76            } catch (MQClientException e) {
77                e.printStackTrace();
78            } catch (InterruptedException e) {
79                e.printStackTrace();
80            } catch (RemotingException e) {
81                e.printStackTrace();
82            } catch (MQBrokerException e) {
83                e.printStackTrace();
84            }
85        }
86        /**
87         * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
88         * 注意：我们建议应用在JBOSS、Tomcat等容器的退出销毁方法里调用shutdown方法
89         */
90        producer.shutdown();
91    }
92}

Linux的centos7.2部署rocketMq3.5.8

背景

软件清单

操作步骤

代码区

评论区

相关推荐