rocketmq集成boot

野性酷女 2021-12-17 04:21 267阅读 0赞

# 3.1.启动 #

3.1.1.启动namesrv  
nohup sh mqnamesrv > /dev/null 2>&1 &

3.1.2.修改broker.conf  
vim …/conf/broker.conf  
最下面添加两行

namesrvAddr = 192.168.5.128:9876
    brokerIP1 = 192.168.5.128

Ip是虚拟机的ip

3.1.3.启动broker  
nohup sh mqbroker -n 192.168.5.128:9876 autoCreateTopicEnable=true -c …/conf/broker.conf /dev/null 2>&1 &

只有这样启动，外网才能正常收发消息。

# 3.2.项目 #

3.2.1.配置文件  
3.2.1.1.Pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.study</groupId>
        <artifactId>rocketmq-boot</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-parent</artifactId>
            <version>1.5.3.RELEASE</version>
        </parent>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.1.0-incubating</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-common</artifactId>
                <version>4.1.0-incubating</version>
            </dependency>
        </dependencies>
    </project>

3.2.1.2.application.properties

# 消费者的组名
    apache.rocketmq.consumerGroup=consumerGroup
    
    # 生产者的组名
    apache.rocketmq.producerGroup=producerGroup
    
    # NameServer地址
    apache.rocketmq.namesrvAddr=192.168.5.128:9876

3.2.2.Java文件  
读取属性

package com.study.rocketmq.boot.config.property;
    
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.stereotype.Component;
    
    /**
     * Created by Administrator on 2018/4/27.
     */
    @Component
    @ConfigurationProperties(prefix = "apache.rocketmq")
    public class RocketmqProperty {
        private String consumerGroup;
        private String producerGroup;
        private String namesrvAddr;
    
        public String getConsumerGroup() {
            return consumerGroup;
        }
    
        public void setConsumerGroup(String consumerGroup) {
            this.consumerGroup = consumerGroup;
        }
    
        public String getProducerGroup() {
            return producerGroup;
        }
    
        public void setProducerGroup(String producerGroup) {
            this.producerGroup = producerGroup;
        }
    
        public String getNamesrvAddr() {
            return namesrvAddr;
        }
    
        public void setNamesrvAddr(String namesrvAddr) {
            this.namesrvAddr = namesrvAddr;
        }
    }

常量

package com.study.rocketmq.boot.util;
    
    /**
     * Created by Administrator on 2018/4/27.
     */
    public class ConstantUtil {
        public static final String PUSH_TOPIC = "PushTopic";    //topic
        public static final String TAG = "push";    //tag
    
    }

发送

package com.study.rocketmq.boot.service;
    
    import com.study.rocketmq.boot.config.property.RocketmqProperty;
    import com.study.rocketmq.boot.util.ConstantUtil;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.PostConstruct;
    
    /**
     * Created by Administrator on 2018/4/27.
     */
    @Service
    public class Producer {
        @Autowired
        private RocketmqProperty rocketmqProperty;
    
        @PostConstruct
        private void init() {
            //生产者的组名
            DefaultMQProducer producer = new DefaultMQProducer(rocketmqProperty.getProducerGroup());
            //指定NameServer地址，多个地址以 ; 隔开
            producer.setNamesrvAddr(rocketmqProperty.getNamesrvAddr());
    
            try {
                /**
                 * Producer对象在使用之前必须要调用start初始化，初始化一次即可
                 * 注意：切记不可以在每次发送消息时，都调用start方法
                 */
                producer.start();
                System.out.println("producer started");
    
                for (int i = 0; i < 3; i++) {
                    String messageBody = "我是消息内容:" + i;
                    String message = new String(messageBody.getBytes(), "utf-8");
    
                    //构建消息
                    Message msg = new Message(ConstantUtil.PUSH_TOPIC //PushTopic
                            ,ConstantUtil.TAG //Tag
                            ,"key_" + i //keys
                            ,message.getBytes() //body
                            );
                    //发送消息
                    SendResult result = producer.send(msg);
    
                    System.out.println("send success. MsgId = " + result.getMsgId() +
                        ", status = " + result.getSendStatus());
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                producer.shutdown();
            }
        }
    }

接收

package com.study.rocketmq.boot.service;
    
    import com.study.rocketmq.boot.config.property.RocketmqProperty;
    import com.study.rocketmq.boot.util.ConstantUtil;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.PostConstruct;
    import java.io.UnsupportedEncodingException;
    import java.util.List;
    
    /**
     * Created by Administrator on 2018/4/27.
     */
    @Service
    public class Consumer {
        @Autowired
        private RocketmqProperty rocketmqProperty;
    
        @PostConstruct
        private void init() {
            //消费者的组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketmqProperty.getConsumerGroup());
    
            //指定NameServer地址，多个地址以 ; 隔开
            consumer.setNamesrvAddr(rocketmqProperty.getNamesrvAddr());
    
            try {
                //订阅PushTopic下Tag为push的消息
                consumer.subscribe(ConstantUtil.PUSH_TOPIC, ConstantUtil.TAG);
    
                //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
                //如果非第一次启动，那么按照上次消费的位置继续消费
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                consumer.registerMessageListener(new MessageListenerConcurrently() {
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        try {
                            for (MessageExt messageExt :
                                    list) {
                                System.out.println("messageExt = " + messageExt);
                                String messageBody = new String(messageExt.getBody(), "utf-8");
                                System.out.println("consumer MsgId = " + messageExt.getMsgId() + ", msgBody = " + messageBody);
                            }
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                });
                consumer.start();
                System.out.println("consumer started");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

启动类

package com.study.rocketmq.boot;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /**
     * Created by Administrator on 2018/4/27.
     */
    @SpringBootApplication
    public class RocketmqBootMain {
        public static void main(String[] args) {
            SpringApplication.run(RocketmqBootMain.class, args);
        }
    }

3.2.3.成功  
![在这里插入图片描述][2019070815173629.png]  
消费消息，是乱序的。

consumer started
    producer started
    send success. MsgId = C0A803CA2AFC18B4AAC2892531CC0000, status = SEND_OK
    send success. MsgId = C0A803CA2AFC18B4AAC2892531E10001, status = SEND_OK
    send success. MsgId = C0A803CA2AFC18B4AAC2892531E50002, status = SEND_OK
    messageExt = MessageExt [queueId=2, storeSize=193, queueOffset=1, sysFlag=0, bornTimestamp=1524812916173, bornHost=/192.168.5.1:58664, storeTimestamp=1524812915586, storeHost=/192.168.5.128:10911, msgId=C0A8058000002A9F000000000002AAB6, commitLogOffset=174774, bodyCRC=126356130, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=PushTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=2, KEYS=key_0, CONSUME_START_TIME=1524812916246, UNIQ_KEY=C0A803CA2AFC18B4AAC2892531CC0000, WAIT=true, TAGS=push}, body=20]]
    consumer MsgId = C0A803CA2AFC18B4AAC2892531CC0000, msgBody = 我是消息内容:0
    messageExt = MessageExt [queueId=0, storeSize=193, queueOffset=0, sysFlag=0, bornTimestamp=1524812916197, bornHost=/192.168.5.1:58664, storeTimestamp=1524812915604, storeHost=/192.168.5.128:10911, msgId=C0A8058000002A9F000000000002AC38, commitLogOffset=175160, bodyCRC=1770417038, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=PushTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1, KEYS=key_2, CONSUME_START_TIME=1524812916246, UNIQ_KEY=C0A803CA2AFC18B4AAC2892531E50002, WAIT=true, TAGS=push}, body=20]]
    consumer MsgId = C0A803CA2AFC18B4AAC2892531E50002, msgBody = 我是消息内容:2
    messageExt = MessageExt [queueId=3, storeSize=193, queueOffset=1, sysFlag=0, bornTimestamp=1524812916193, bornHost=/192.168.5.1:58664, storeTimestamp=1524812915599, storeHost=/192.168.5.128:10911, msgId=C0A8058000002A9F000000000002AB77, commitLogOffset=174967, bodyCRC=1888434740, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=PushTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=2, KEYS=key_1, CONSUME_START_TIME=1524812916245, UNIQ_KEY=C0A803CA2AFC18B4AAC2892531E10001, WAIT=true, TAGS=push}, body=20]]
    consumer MsgId = C0A803CA2AFC18B4AAC2892531E10001, msgBody = 我是消息内容:1
    2018-04-27 15:08:36.393  INFO 11004 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Registering beans for JMX exposure on startup
    2018-04-27 15:08:36.411  INFO 11004 --- [           main] c.study.rocketmq.boot.RocketmqBootMain   : Started RocketmqBootMain in 4.215 seconds (JVM running for 4.797)

[2019070815173629.png]: /images/20211213/edd469c39afa4b81bc65d79505dd3ebf.png