快捷導(dǎo)航

spring?boot項(xiàng)目中集成rocketmq詳細(xì)步驟

更新時(shí)間：2023年09月13日 09:07:39 作者：orton777

這篇文章主要給大家介紹了關(guān)于spring?boot項(xiàng)目中集成rocketmq的相關(guān)資料,springboot集成rocketmq的方法非常簡單,文中直接上代碼,需要的朋友可以參考下

集成Spring Boot和RocketMQ

在現(xiàn)代的微服務(wù)架構(gòu)中，消息隊(duì)列已經(jīng)成為一種常見的異步處理模式，它能解決服務(wù)間的同步調(diào)用、耦合度高、流量高峰等問題。RocketMQ是阿里巴巴開源的一款消息中間件，性能優(yōu)秀，功能齊全，被廣泛應(yīng)用在各種業(yè)務(wù)場景。

本文將詳細(xì)介紹如何在Spring Boot項(xiàng)目中集成RocketMQ，實(shí)現(xiàn)消息的生產(chǎn)和消費(fèi)。

開發(fā)環(huán)境

JDK 1.8 或更高
RocketMQ 4.8.0 或更高
Spring Boot 2.3.1.RELEASE 或更高
Maven 3.0 或更高

RocketMQ服務(wù)器部署

首先，我們需要在本地或服務(wù)器上部署RocketMQ。具體的部署步驟可以參考RocketMQ官方文檔。為了簡化部署，我們可以使用Docker進(jìn)行部署。

Spring Boot項(xiàng)目創(chuàng)建

我們使用Spring Initializr創(chuàng)建一個(gè)新的Spring Boot項(xiàng)目，選擇Web、Lombok和RocketMQ Spring Boot Starter為項(xiàng)目依賴。

pom.xml示例：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.0</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

配置RocketMQ

在application.properties文件中配置RocketMQ的服務(wù)器地址和其他相關(guān)參數(shù)。

rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=my-group

在這里，rocketmq.name-server是RocketMQ服務(wù)器的地址，rocketmq.producer.group是生產(chǎn)者的組名。

消息生產(chǎn)者

接下來，我們創(chuàng)建一個(gè)消息生產(chǎn)者。在Spring Boot項(xiàng)目中，我們可以使用RocketMQTemplate來發(fā)送消息。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class ProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GetMapping("/send")
    public String send(String message) {
        rocketMQTemplate.convertAndSend("test-topic", message);
        return "Message: '" + message + "' sent.";
    }
}

上述代碼中，我們創(chuàng)建了一個(gè)RESTful接口/send，當(dāng)接口被調(diào)用時(shí)，它將發(fā)送一個(gè)消息到test-topic主題。

消息消費(fèi)者

接下來，我們創(chuàng)建一個(gè)消息消費(fèi)者。在Spring Boot項(xiàng)目中，我們可以使用@RocketMQMessageListener注解來定義一個(gè)消息消費(fèi)者。

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer_test-topic")
public class ConsumerService implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.printf("------- StringConsumer received: %s \n", message);
    }
}

上述代碼中，我們定義了一個(gè)消息消費(fèi)者，它將監(jiān)聽test-topic主題的消息，當(dāng)有新的消息時(shí)，它將打印消息內(nèi)容。

測試

至此，我們已經(jīng)完成了Spring Boot集成RocketMQ的所有代碼。接下來，我們就可以運(yùn)行Spring Boot項(xiàng)目，并通過訪問/send接口來發(fā)送消息，查看控制臺(tái)的輸出來驗(yàn)證消息消費(fèi)者是否可以正常接收消息。

這就是Spring Boot集成RocketMQ的全過程。RocketMQ作為一款功能強(qiáng)大的消息中間件，不僅支持基本的消息生產(chǎn)和消費(fèi)，還支持許多高級(jí)特性，如事務(wù)消息、順序消息、延遲消息等。在實(shí)際的項(xiàng)目開發(fā)中，我們可以根據(jù)業(yè)務(wù)需求選擇合適的消息模型，提高系統(tǒng)的可用性和可靠性。

事務(wù)消息

RocketMQ支持發(fā)送事務(wù)消息，也就是說，在發(fā)送消息的同時(shí)，我們可以執(zhí)行本地的數(shù)據(jù)庫操作，只有當(dāng)本地的數(shù)據(jù)庫操作成功時(shí)，消息才會(huì)真正被發(fā)送出去。

下面是一個(gè)發(fā)送事務(wù)消息的例子：

import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.*;
@RestController
public class TransactionProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GetMapping("/sendTransaction")
    public String sendTransaction(String message) {
        ExecutorService executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5000), r -> {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        });
        TransactionListener transactionListener = new TransactionListenerImpl();
        TransactionMQProducer producer = rocketMQTemplate.createAndStartTransactionMQProducer("transaction-group",transactionListener,executor);
        producer.sendMessageInTransaction("test-topic", "TagA", message, null);
        return "Transaction Message: '" + message + "' sent.";
    }
}

在上述代碼中，我們創(chuàng)建了一個(gè)TransactionMQProducer，并設(shè)置了一個(gè)TransactionListener來處理事務(wù)的提交和回滾。當(dāng)發(fā)送事務(wù)消息時(shí)，我們需要調(diào)用sendMessageInTransaction方法。

順序消息

RocketMQ支持發(fā)送順序消息，也就是說，消息會(huì)按照發(fā)送的順序被消費(fèi)。

下面是一個(gè)發(fā)送順序消息的例子：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.messaging.support.MessageBuilder;
@RestController
public class OrderlyProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GetMapping("/sendOrderly")
    public String sendOrderly(String message) {
        for (int i = 0; i < 100; i++) {
            rocketMQTemplate.syncSendOrderly("orderly_topic", MessageBuilder.withPayload(message + i).build(), "hashkey");
        }
        return "Orderly Message: '" + message + "' sent.";
    }
}

在上述代碼中，我們調(diào)用syncSendOrderly方法發(fā)送順序消息。該方法的第三個(gè)參數(shù)是hashkey，RocketMQ會(huì)根據(jù)這個(gè)key來決定消息發(fā)送到哪個(gè)隊(duì)列，具有相同hashkey的消息會(huì)發(fā)送到同一個(gè)隊(duì)列。

延遲消息

RocketMQ支持發(fā)送延遲消息，也就是說，消息不會(huì)立即被消費(fèi)，而是會(huì)在指定的時(shí)間后被消費(fèi)。

下面是一個(gè)發(fā)送延遲消息的例子：

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.messaging.support.MessageBuilder;
@RestController
public class DelayProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @GetMapping("/sendDelay")
    public String sendDelay(String message) {
        rocketMQTemplate.syncSend("delay_topic", MessageBuilder.withPayload(message).build(), 1000, 4);
        return "Delay Message: '" + message + "' sent.";
    }
}

在上述代碼中，我們調(diào)用syncSend方法發(fā)送延遲消息。該方法的第三個(gè)參數(shù)是延遲時(shí)間，第四個(gè)參數(shù)是延遲級(jí)別。