spring boot 整合rocketmq

柔光的暖阳◎ 2022-10-07 06:48 218阅读 0赞

1、linux安装rocketmq，官网下载安装包解压到制定目录，本次使用4.8.0版本

![20210616184825858.png][]

修改堆内存大小

vim ./bin/runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
    choose_gc_options
    JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
    JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
    JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib:${JAVA_HOME}/lib/ext"
    #JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
    JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
    JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"

修改broker.conf配置

vim ./conf/broker.conf

#broker 监听端口
    listenPort = 10911
    brokerIP1 = 192.168.81.200
    #namesrvAddr地址
    namesrvAddr = 192.168.81.200:9876
    brokerClusterName = TestCluster
    brokerName = broker-test
    #brokerId = 0表示主节点，大于0表示从节点
    brokerId = 0
    deleteWhen = 04
    fileReservedTime = 48
    #同步发行消息配置SYNC_MASTER
    brokerRole = SYNC_MASTER
    #同步刷盘配置SYNC_MASTER
    flushDiskType = SYNC_FLUSH
    #发送异步消息配置
    #brokerRole = ASYNC_MASTER
    #异步刷盘消息配置
    #flushDiskType = ASYNC_FLUSH
    #自动创建topic
    autoCreateTopicEnable=true
    storePathCommitLog = /home/master/rocketmq4.8.0/store/commitlog/
    storePathConsumerQueue = /home/master/rocketmq4.8.0/store/consumequeue/

2、自定义启动脚本start\_rocketmq.sh

#!/bin/bash
    nid=`jps| grep NamesrvStartup | awk '{print $1}'`
    if [[ -z $nid ]]; then
        echo "没有正在运行的NamesrvStartup...直接启动"
    else
        echo "扫描到正在运行的NamesrvStartup...执行关闭"
        kill  $nid
    fi
    nohup sh  ./bin/mqnamesrv -n 192.168.81.200:9876  >>  ./logs/nameserver.log  2>&1 &
    nid=`jps| grep NamesrvStartup | awk '{print $0}'`
    echo "$nid ... 启动成功"
    
    bid=`jps| grep BrokerStartup | awk '{print $1}'`
    if [[ -z $bid ]]; then
        echo "没有正在运行的BrokerStartup...直接启动"
    else
        echo "扫描到正在运行的BrokerStartup执行关闭"
        kill  $bid
    fi
    nohup  sh ./bin/mqbroker -n 192.168.81.200:9876   -c   ./conf/broker.conf  autoCreateTopicEnable=true  >> ./logs/broker.log 2>&1 &
    bid=`jps| grep BrokerStartup | awk '{print $0}'`
    echo "$bid ...启动成功"

3、整合spring boot

pom.xml文件导入依赖

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.2.0</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.apache.rocketmq</groupId>
                        <artifactId>rocketmq-client</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.8.0</version>
            </dependency>

appliction.yml配置

rocketmq:
     #ip为虚拟机ip
      name-server: http://192.168.81.200:9876
      producer:
        group: rocketmq_test
        send-message-timeout: 3000

消息生产者

@Service
    public class RocketMqProducer {
    
        private final  static Logger LOGGER = LoggerFactory.getLogger(RocketMqProducer.class);
    
        @Autowired
        private RocketMQTemplate rocketMqTemplate;
        
        /**
         * 发送普通消息
         */
        public void sendMsg(String msgBody) {
            rocketMqTemplate.syncSend("rocketmq_test_topic", MessageBuilder.withPayload(msgBody).build());
        }
    
        /**
         * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
         */
        public void sendAsyncMsg(String msgBody) {
            rocketMqTemplate.asyncSend("rocketmq_test_topic", MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    // 处理消息发送成功逻辑
                    LOGGER.info("RocketMqProducer---->"+sendResult.getMessageQueue().toString());
                }
                @Override
                public void onException(Throwable e) {
                    // 处理消息发送异常逻辑
                    LOGGER.info("RocketMqProducer--sendAsyncMsg-->e"+e.toString());
                }
            });
        }
    
        /**
         * 发送延时消息<br/>
         * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h<br/>
         */
        public void sendDelayMsg(String msgBody, Integer delayLevel) {
            rocketMqTemplate.syncSend("rocketmq_test_topic", MessageBuilder.withPayload(msgBody).build(), 3000, delayLevel);
        }
    
        /**
         * 发送带tag的消息,直接在topic后面加上":tag"
         */
        public void sendTagMsg(String msgBody) {
            rocketMqTemplate.syncSend("rocketmq_test_topic:tag1", MessageBuilder.withPayload(msgBody).build());
        }
    
    }

消息消费者

@Slf4j
    @Component
    @RocketMQMessageListener(topic = "rocketmq_test_topic",selectorExpression="*",consumerGroup = "rocketmq_test_topic")
    public class RocketMQMsgListener implements RocketMQListener<MessageExt> {
    
        @Override
        public void onMessage(MessageExt message) {
            byte[] body = message.getBody();
            String msg = new String(body);
            log.info("RocketMQMsgListener接收到消息：{}", msg);
        }
    }

4、rocketmq的各个角色

Producer: 消息发送者

Consumer: 消息接收者

Broker: 暂存消息和传输消息

NameServer: 管理Broker

Topic: 区分消息分类，一个消息发送者可以发送给一个或多个Topic,一个接收者可以接收一个或者多个topic消息

MessageQueue:Topic分区，用于发送消息

RocketMq集群架构图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdjaGVuZ3d1MjAxMg_size_16_color_FFFFFF_t_70][]

5、发送事务消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdjaGVuZ3d1MjAxMg_size_16_color_FFFFFF_t_70 1][]       RocketMQ实现事务消息主要分为两个阶段：正常事务的发送及提交、事务信息的补偿流程整体流程为：

(1)正常事务发送与提交阶段

\*生产者发送一个半消息给MQServer（半消息是指消费者暂时不能消费的消息）

\* 服务端响应消息写入结果，半消息发送成功

\* 开始执行本地事务

\* 根据本地事务的执行状态执行Commit或者Rollback操作

(2) 事务信息的补偿流程

\*  如果MQServer长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操作请求

\* 生产者收到确认回查请求后，检查本地事务的执行状态

\* 根据检查后的结果执行Commit或者Rollback操作

\* 补偿阶段主要是用于解决生

@Autowired
        private RocketMQTemplate rocketMqTemplate;
         
        
        public  void sendNativeTxMessage(String  msg,int  id) {
            /// 消息发送和本地事务的原子性操作  半消息
            TransactionMQProducer  producer=new TransactionMQProducer();
            producer.setNamesrvAddr("192.168.81.200:9876");
            producer.setProducerGroup("ptx_group");
            producer.setTransactionListener(new TransactionListener() {
                @Override
                public LocalTransactionState executeLocalTransaction(Message message, Object o) {
       
    
                    String tags = message.getTags();
                    if (StringUtils.equals("TagA",tags)) {
                        ///提交本地事务成功
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }else  if (StringUtils.equals("TagB",tags)) {
                        /// 提交本地事务失败
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                    ///由于网络问题造成的未知提交状态，列入回查状态
                    return LocalTransactionState.UNKNOW;
                }
    
                @Override
                public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                    事务回查
                    log.info("checkLocalTransaction->message {}",messageExt.getTags());
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            });
            try {
                producer.start();
                String[]  tags={"TagA","TagB","TagC"};
                for (int i=0;i<3;i++){
                    Message  message=new Message("tx_topic",tags[i],msg.getBytes());
                    SendResult sendResult = producer.sendMessageInTransaction(message, null);
                    log.info("sendResult--->"+sendResult.toString());
                    TimeUnit.SECONDS.sleep(5);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }

根据以上情况，最终会有两条消息发送出去，tagA和tagC

@Slf4j
    @Conditional(AutoRocketMQ.class)
    @Component
    @RocketMQMessageListener(topic = "tx_topic",consumerGroup = "ctx_group",selectorExpression="*")
    public class TransactionConsumerListener implements RocketMQListener<MessageExt> {
    
    
        @Override
        public void onMessage(MessageExt message) {
            byte[] body = message.getBody();
            String msg = new String(body);
            log.info("tx_topic接收到消息:{}  {}", msg,message.getTags());
        }
    }

[20210616184825858.png]: /images/20221005/426d671eb5dd4812a84ba4f809c534e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdjaGVuZ3d1MjAxMg_size_16_color_FFFFFF_t_70]: /images/20221005/d03efb0eeebe4c138a12f7e34a00db73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlbmdjaGVuZ3d1MjAxMg_size_16_color_FFFFFF_t_70 1]: /images/20221005/ef1071e03257478eabb957ede4366020.png