快捷導(dǎo)航

SpringBoot集成RocketMQ實(shí)現(xiàn)消息發(fā)送的三種方式

更新時(shí)間：2023年09月05日 11:57:30 作者：Sea-Man

RocketMQ?支持3?種消息發(fā)送方式：?同步?（sync）、異步（async）、單向（oneway）,本文就將給大家介紹一下SpringBoot集成RocketMQ實(shí)現(xiàn)消息發(fā)送的三種方式文中有詳細(xì)的代碼示例,需要的朋友可以參考下

同步：發(fā)送者向 MQ 執(zhí)行發(fā)送消息API 時(shí)，同步等待，直到消息服務(wù)器返回發(fā)送結(jié)果。
異步：發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時(shí)，指定消息發(fā)送成功后的回調(diào)函數(shù)，然后調(diào)用消息發(fā)送API 后，立即返回，消息發(fā)送者線程不阻塞，直到運(yùn)行結(jié)束，消息發(fā)送成功或失敗的回調(diào)任務(wù)在一個(gè)新的線程中返回。
單向：消息發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時(shí)，直接返回，不等待消息服務(wù)器的結(jié)果，也不注冊回調(diào)函數(shù)，只管發(fā)，不管是否成功存儲(chǔ)在消息服務(wù)器上。

前提：

運(yùn)行項(xiàng)目需要具備RocketMQ環(huán)境

一、pom文件引入依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.2</version>
</dependency>

二、application.yml文件添加內(nèi)容

rocketmq:
  name-server: IP:9876  #IP為rocketmq訪問的地址
  producer:
    group: first1-group  #事務(wù)消息才會(huì)用到

三、創(chuàng)建producer生產(chǎn)者

package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class RocketController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @RequestMapping(value = "/rocket", method = RequestMethod.GET)
    public void noTag() {
        // convertAndSend() 發(fā)送普通字符串消息
        rocketMQTemplate.convertAndSend("sendMessage_topic", "Hello Word");
    }
    @RequestMapping(value = "/tagA", method = RequestMethod.GET)
    public void tagA() {
        rocketMQTemplate.convertAndSend("sendMessage_topic:tagA", "hello world tagA");
    }
    @RequestMapping(value = "/tagB", method = RequestMethod.GET)
    public void tagB() {
        rocketMQTemplate.convertAndSend("sendMessage_topic:tagB", "hello world tagB");
    }
    @RequestMapping(value = "/syncSend", method = RequestMethod.GET)
    public void syncSend() {
        String json = "發(fā)送同步消息";
        SendResult sendResult = rocketMQTemplate.syncSend("sendMessage_topic:1", json);
        System.out.println(sendResult);
    }
    @RequestMapping(value = "/aSyncSend", method = RequestMethod.GET)
    public void aSyncSend() {
        String json = "發(fā)送異步消息";
        SendCallback callback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("發(fā)送消息成功");
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("發(fā)送消息失敗");
            }
        };
        rocketMQTemplate.asyncSend("sendMessage_topic", json, callback);
    }
    @RequestMapping(value = "/sendOneWay", method = RequestMethod.GET)
    public void sendOneWay() {
        rocketMQTemplate.sendOneWay("sendMessage_topic", "發(fā)送單向消息");
    }
}

四、創(chuàng)建Consumer消費(fèi)者（創(chuàng)建兩個(gè)消費(fèi)者，所屬一個(gè)Topic）

Consumer1:

package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
 * topic 是主題
 * consumerGroup 是消費(fèi)者組，一條消息只能被同一個(gè)消費(fèi)者組里的一個(gè)消費(fèi)者消費(fèi)。
 * selectorExpression 是用于消息過濾的，以 TAG 方式為例：
 * 默認(rèn)為 "*"，表示不過濾，消費(fèi)此 topic 下所有消息
 * 配置為 "tagA"，表示只消費(fèi)此 topic 下 TAG = tagA 的消息
 * 配置為 "tagA || tagB"，表示消費(fèi)此 topic 下 TAG = tagA 或  TAG = tagB 的消息，以此類推
 * 消費(fèi)模式：默認(rèn) CLUSTERING （ CLUSTERING：負(fù)載均衡 ）（ BROADCASTING：廣播機(jī)制 ）
 */
@RocketMQMessageListener(topic = "sendMessage_topic",
        consumerGroup = "consumer-group-test1",
//        selectorExpression = "tagA || tagB",
        messageModel = MessageModel.CLUSTERING)
@Component
public class DemoConsumer1 implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("receive message1:" + s);
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("處理完成");
    }
}

Consumer2:

package com.tlxy.lhn.controller.rocketmq;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
 * topic 是主題
 * consumerGroup 是消費(fèi)者組，一條消息只能被同一個(gè)消費(fèi)者組里的一個(gè)消費(fèi)者消費(fèi)。
 * selectorExpression 是用于消息過濾的，以 TAG 方式為例：
 * 默認(rèn)為 "*"，表示不過濾，消費(fèi)此 topic 下所有消息
 * 配置為 "tagA"，表示只消費(fèi)此 topic 下 TAG = tagA 的消息
 * 配置為 "tagA || tagB"，表示消費(fèi)此 topic 下 TAG = tagA 或  TAG = tagB 的消息，以此類推
 * 消費(fèi)模式：默認(rèn) CLUSTERING （ CLUSTERING：負(fù)載均衡 ）（ BROADCASTING：廣播機(jī)制 ）
 */
@RocketMQMessageListener(topic = "sendMessage_topic",
        consumerGroup = "consumer-group-test1",
        messageModel = MessageModel.CLUSTERING)
@Component
public class DemoConsumer2 implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.println("receive message2:" + s);
        try {
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("處理完成");
    }
}

五、啟動(dòng)項(xiàng)目測試

1、發(fā)送同步消息：

以上輸出可以看到：同步消息發(fā)送后，消息發(fā)送到broker后就返回結(jié)果了，消費(fèi)端還未處理完，兩者互互不影響。

2、發(fā)送異步消息：

以上輸出：發(fā)送者向MQ 執(zhí)行發(fā)送消息API 時(shí)，指定消息發(fā)送成功后的回調(diào)函數(shù)，然后調(diào)用消息發(fā)送API 后，立即返回，消息發(fā)送者線程不阻塞，直到運(yùn)行結(jié)束，消息發(fā)送成功或失敗的回調(diào)任務(wù)在一個(gè)新的線程中返回。

3、發(fā)送單向消息：