spring?boot項(xiàng)目中集成rocketmq詳細(xì)步驟
集成Spring Boot和RocketMQ
在現(xiàn)代的微服務(wù)架構(gòu)中,消息隊(duì)列已經(jīng)成為一種常見(jiàn)的異步處理模式,它能解決服務(wù)間的同步調(diào)用、耦合度高、流量高峰等問(wèn)題。RocketMQ是阿里巴巴開(kāi)源的一款消息中間件,性能優(yōu)秀,功能齊全,被廣泛應(yīng)用在各種業(yè)務(wù)場(chǎng)景。
本文將詳細(xì)介紹如何在Spring Boot項(xiàng)目中集成RocketMQ,實(shí)現(xiàn)消息的生產(chǎn)和消費(fèi)。
開(kā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官方文檔。為了簡(jiǎn)化部署,我們可以使用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)者
接下來(lái),我們創(chuàng)建一個(gè)消息生產(chǎn)者。在Spring Boot項(xiàng)目中,我們可以使用RocketMQTemplate來(lái)發(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)者
接下來(lái),我們創(chuàng)建一個(gè)消息消費(fèi)者。在Spring Boot項(xiàng)目中,我們可以使用@RocketMQMessageListener注解來(lái)定義一個(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)聽(tīng)test-topic主題的消息,當(dāng)有新的消息時(shí),它將打印消息內(nèi)容。
測(cè)試
至此,我們已經(jīng)完成了Spring Boot集成RocketMQ的所有代碼。接下來(lái),我們就可以運(yùn)行Spring Boot項(xiàng)目,并通過(guò)訪問(wèn)/send接口來(lái)發(fā)送消息,查看控制臺(tái)的輸出來(lái)驗(yàn)證消息消費(fèi)者是否可以正常接收消息。
這就是Spring Boot集成RocketMQ的全過(guò)程。RocketMQ作為一款功能強(qiáng)大的消息中間件,不僅支持基本的消息生產(chǎn)和消費(fèi),還支持許多高級(jí)特性,如事務(wù)消息、順序消息、延遲消息等。在實(shí)際的項(xiàng)目開(kāi)發(fā)中,我們可以根據(jù)業(yè)務(wù)需求選擇合適的消息模型,提高系統(tǒng)的可用性和可靠性。
事務(wù)消息
RocketMQ支持發(fā)送事務(wù)消息,也就是說(shuō),在發(fā)送消息的同時(shí),我們可以執(zhí)行本地的數(shù)據(jù)庫(kù)操作,只有當(dāng)本地的數(shù)據(jù)庫(kù)操作成功時(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來(lái)處理事務(wù)的提交和回滾。當(dāng)發(fā)送事務(wù)消息時(shí),我們需要調(diào)用sendMessageInTransaction方法。
順序消息
RocketMQ支持發(fā)送順序消息,也就是說(shuō),消息會(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來(lái)決定消息發(fā)送到哪個(gè)隊(duì)列,具有相同hashkey的消息會(huì)發(fā)送到同一個(gè)隊(duì)列。
延遲消息
RocketMQ支持發(fā)送延遲消息,也就是說(shuō),消息不會(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í)別。
總結(jié)
到此這篇關(guān)于spring boot項(xiàng)目中集成rocketmq詳細(xì)步驟的文章就介紹到這了,更多相關(guān)springboot集成rocketmq內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
springcloud項(xiàng)目里application.yml不加載的坑及解決
這篇文章主要介紹了springcloud項(xiàng)目里application.yml不加載的坑及解決,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-07-07
ConcurrentHashMap線程安全及實(shí)現(xiàn)原理實(shí)例解析
這篇文章主要介紹了ConcurrentHashMap線程安全及實(shí)現(xiàn)原理實(shí)例解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-11-11
解決SpringBoot 測(cè)試類無(wú)法自動(dòng)注入@Autowired的問(wèn)題
這篇文章主要介紹了解決SpringBoot 測(cè)試類無(wú)法自動(dòng)注入@Autowired的問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-03-03
IDEA無(wú)法識(shí)別SpringBoot項(xiàng)目的簡(jiǎn)單解決辦法
今天使用idea的時(shí)候,遇到idea無(wú)法啟動(dòng)springboot,所以這篇文章主要給大家介紹了關(guān)于IDEA無(wú)法識(shí)別SpringBoot項(xiàng)目的簡(jiǎn)單解決辦法,需要的朋友可以參考下2023-08-08
談?wù)凧ava中try-catch-finally中的return語(yǔ)句
我們知道return語(yǔ)句用在某一個(gè)方法中,一是用于返回函數(shù)的執(zhí)行結(jié)果,二是用于返回值為void類型的函數(shù)中,僅僅是一個(gè)return語(yǔ)句(return ;),此時(shí)用于結(jié)束方法的執(zhí)行,也即此return后的語(yǔ)句將不會(huì)被執(zhí)行,當(dāng)然,這種情況下return語(yǔ)句后不能再有其它的語(yǔ)句了2016-01-01
鎖超時(shí)發(fā)現(xiàn)parallelStream并行流線程上下文坑解決
這篇文章主要為大家介紹了鎖超時(shí)發(fā)現(xiàn)parallelStream并行流線程上下文坑解決,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-08-08

