本文快速入门,RocketMQ消息系统的安装部署,发送,和接收消息,监控消息，的详细说明。

环境需要

64位操作系统，建议使用Linux / Unix /

• CentOs7.3

• 64bit JDK 1.8+

• Maven 3.2.x

• Git 1.8.3.1

环境安装

请参考我的另一篇文章

搭建 Apache RocketMQ 单机环境

http://www.ymq.io/2018/02/01/RocketMQ-install

新加项目

新建一个 maven 项目，这里就不详细操作了，大家都会的

不过也可以下载我的示例源码，下载地址如下

GitHub 源码：https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-rocketmq

添加依赖

在POM 中添加如下依赖

1. <!-- RocketMq客户端相关依赖 -->

2. <dependency>

3.    <groupId>org.apache.rocketmq</groupId>

4.    <artifactId>rocketmq-client</artifactId>

5.    <version>4.1.0-incubating</version>

6. </dependency>

8. <dependency>

9.    <groupId>org.apache.rocketmq</groupId>

10.    <artifactId>rocketmq-common</artifactId>

11.    <version>4.1.0-incubating</version>

12. </dependency>

配置文件

在配置文件 application.properties 添加一下内容

1. # 消费者的组名

2. apache.rocketmq.consumer.PushConsumer=PushConsumer

4. # 生产者的组名

5. apache.rocketmq.producer.producerGroup=Producer

7. # NameServer地址

8. apache.rocketmq.namesrvAddr=192.168.252.121:9876

消息生产者

1. @Component

2. public class Producer {

4.    /**

5.     * 生产者的组名

6.     */

7.    @Value("${apache.rocketmq.producer.producerGroup}")

8.    private String producerGroup;

10.    /**

11.     * NameServer 地址

12.     */

13.    @Value("${apache.rocketmq.namesrvAddr}")

14.    private String namesrvAddr;

16.    @PostConstruct

17.    public void defaultMQProducer() {

19.        //生产者的组名

20.        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);

22.        //指定NameServer地址，多个地址以 ; 隔开

23.        producer.setNamesrvAddr(namesrvAddr);

25.        try {

27.            /**

28.             * Producer对象在使用之前必须要调用start初始化，初始化一次即可

29.             * 注意：切记不可以在每次发送消息时，都调用start方法

30.             */

31.            producer.start();

33.            for (int i = 0; i < 100; i++) {

35.                String messageBody = "我是消息内容:" + i;

37.                String message = new String(messageBody.getBytes(), "utf-8");

39.                //构建消息

40.                Message msg = new Message("PushTopic" /* PushTopic */, "push"/* Tag  */, "key_" + i /* Keys */, message.getBytes());

42.                //发送消息

43.                SendResult result = producer.send(msg);

45.                System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());

47.            }

49.        } catch (Exception e) {

50.            e.printStackTrace();

51.        } finally {

52.            producer.shutdown();

53.        }

55.    }

56. }

消息消费者

1. @Component

2. public class Consumer {

4.    /**

5.     * 消费者的组名

6.     */

7.    @Value("${apache.rocketmq.consumer.PushConsumer}")

8.    private String consumerGroup;

10.    /**

11.     * NameServer地址

12.     */

13.    @Value("${apache.rocketmq.namesrvAddr}")

14.    private String namesrvAddr;

16.    @PostConstruct

17.    public void defaultMQPushConsumer() {

19.        //消费者的组名

20.        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

22.        //指定NameServer地址，多个地址以 ; 隔开

23.        consumer.setNamesrvAddr(namesrvAddr);

24.        try {

25.            //订阅PushTopic下Tag为push的消息

26.            consumer.subscribe("PushTopic", "push");

28.            //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费

29.            //如果非第一次启动，那么按照上次消费的位置继续消费

30.            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

31.            consumer.registerMessageListener(new MessageListenerConcurrently() {

33.                @Override

34.                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {

35.                    try {

36.                        for (MessageExt messageExt : list) {

38.                            System.out.println("messageExt: " + messageExt);//输出消息内容

40.                            String messageBody = new String(messageExt.getBody(), "utf-8");

42.                            System.out.println("消费响应：Msg: " + messageExt.getMsgId() + ",msgBody: " + messageBody);//输出消息内容

44.                        }

45.                    } catch (Exception e) {

46.                        e.printStackTrace();

47.                        return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试

48.                    }

49.                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功

50.                }

53.            });

54.            consumer.start();

55.        } catch (Exception e) {

56.            e.printStackTrace();

57.        }

58.    }

60. }

启动服务

1. @SpringBootApplication

2. public class SpringBootRocketmqApplication {

4.    public static void main(String[] args) {

5.        SpringApplication.run(SpringBootRocketmqApplication.class, args);

6.    }

7. }

控制台会有响应

1. 发送响应：MsgId:0AFF015E556818B4AAC208A0504F0063，发送状态:SEND_OK

3. messageExt: MessageExt [queueId=0, storeSize=195, queueOffset=113824, sysFlag=0, bornTimestamp=1517559124047, bornHost=/192.168.252.1:62165, storeTimestamp=1517559135052, storeHost=/192.168.252.121:10911, msgId=C0A8FC7900002A9F00000000056F499C, commitLogOffset=91179420, bodyCRC=1687852546, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=PushTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=113825, KEYS=key_99, CONSUME_START_TIME=1517559124049, UNIQ_KEY=0AFF015E556818B4AAC208A0504F0063, WAIT=true, TAGS=push}, body=21]]

5. 消费响应：Msg: 0AFF015E556818B4AAC208A0504F0063,msgBody: 我是消息内容:99

7. ...

监控服务

RocketMQ web界面监控RocketMQ-Console-Ng部署

https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console

下载并且编译

下载并且 maven 编译

1. git clone https://github.com/apache/rocketmq-externals.git

2. rocketmq-externals/rocketmq-console/

3. mvn clean package -Dmaven.test.skip=true

启动监控服务

rocketmq.config.namesrvAddr NameServer 地址,默认启动端口8080

1. nohup java -jar target/rocketmq-console-ng-1.0.0.jar --rocketmq.config.namesrvAddr=127.0.0.1:9876

访问监控服务

GitHub 源码：https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-rocketmq

Gitee 源码：https://gitee.com/souyunku/spring-boot-examples/tree/master/spring-boot-rabbitmq

Contact

• 作者：鹏磊

• 出处：http://www.ymq.io/2018/02/02/spring-boot-rocketmq-example

• Email：admin@souyunku.com

• 版权归作者所有，转载请注明出处

• Wechat：关注公众号，搜云库，专注于开发技术的研究与知识分享