參考博客
https://blog.csdn.net/daqi1983/article/details/116144764 8-Spring Boot消息服務(wù)
視頻鏈接
https://www.bilibili.com/video/BV1nz4y1X7jh?p=12

1.消息服務(wù)概述

1.為什么使用消息服務(wù)

在多數(shù)應(yīng)用尤其是分布式系統(tǒng)中，消息服務(wù)是不可或缺的重要部分，它使用起來(lái)比較簡(jiǎn)單，同時(shí)解決了不少難題，例如異步處理、應(yīng)用解耦、流量削鋒、分布式事務(wù)管理等，使用消息服務(wù)可以實(shí)現(xiàn)一個(gè)高性能、高可用、高擴(kuò)展的系統(tǒng)。

a.異步處理

b.應(yīng)用解耦

c.流量削峰

d.分布式事務(wù)管理

2.常用的消息中間件

1.ActiveMQ(性能相對(duì)較弱) 2.RabbitMQ 3.Kafka 4.RocketMQ

2.RabbitMQ 消息中間件

1.RabbitMQ 簡(jiǎn)介

RabbitMQ是基于AMQP協(xié)議的輕量級(jí)、可靠、可伸縮和可移植的消息代理，Spring使用RabbitMQ通過(guò)AMQP協(xié)議進(jìn)行通信，在Spirng Boot中對(duì)RabbitMQ進(jìn)行了集成管理。

2.消息代理過(guò)程

3.RabbitMQ 工作模式介紹

a.Work queues（工作隊(duì)列模式） b.Publish/Subscribe（發(fā)布訂閱模式） c.Routing（路由模式） d.Topics（通配符模式） e.RPC

3.RabbitMQ安裝及整合環(huán)境搭建

Windows下RabbitMQ安裝及配置 https://blog.csdn.net/weixin_42673046/article/details/118442323
RabbitMQ的安裝(超詳細(xì))https://blog.csdn.net/qq_42498815/article/details/116139712
在Win環(huán)境下安裝Erlang https://blog.csdn.net/g6256613/article/details/80191402
Windows下RabbitMQ安裝及配置 https://blog.csdn.net/zhm3023/article/details/82217222 (效果極好)

常用命令 判斷環(huán)境變量是否配置 rabbitmqctl status 執(zhí)行 rabbitmq-service.bat stop 停止服務(wù) 執(zhí)行 rabbitmq-service.bat install 重新安裝服務(wù)，這時(shí)才會(huì)重新加載config文件 執(zhí)行 rabbitmq-service.bat start 開(kāi)啟服務(wù)

4.Spring Boot與RabbitMQ整合實(shí)現(xiàn)

4.1、創(chuàng)建Spring Boot項(xiàng)目，添加Web依賴(lài)以及RabbitMQ依賴(lài)。

4.2、在pplication.properties配置文件中編寫(xiě)需要設(shè)置的配置屬性。

# 配置RabbitMQ消息中間件連接配置 spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest #配置RabbitMQ虛擬主機(jī)路徑/，默認(rèn)可以省略 spring.rabbitmq.virtual-host=/

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 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>demo</artifactId><version>0.0.1-SNAPSHOT</version><name>demo</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><spring-boot.version>2.3.7.RELEASE</spring-boot.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>${spring-boot.version}</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.1</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><version>2.3.7.RELEASE</version><configuration><mainClass>com.example.demo.DemoApplication</mainClass></configuration><executions><execution><id>repackage</id><goals><goal>repackage</goal></goals></execution></executions></plugin></plugins></build> </project>

5.Publish/Subscribe（發(fā)布訂閱模式）

1.基于API的方式及測(cè)試的搭建步驟

a.使用AmqpAdmin定制消息組件。

① 在測(cè)試類(lèi)中先引入AmqpAdmin管理類(lèi),定制Publish/Subscribe工作模式所需的消息組件

@Autowiredprivate AmqpAdmin amqpAdmin;/*** 使用AmqpAdmin管理員API定制消息組件 import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue;*/@Testpublic void amqpAdmin() {// 1、定義fanout類(lèi)型的交換器amqpAdmin.declareExchange(new FanoutExchange("fanout_exchange"));// 2、定義兩個(gè)默認(rèn)持久化隊(duì)列，分別處理email和smsamqpAdmin.declareQueue(new Queue("fanout_queue_email"));amqpAdmin.declareQueue(new Queue("fanout_queue_sms"));// 3、將隊(duì)列分別與交換器進(jìn)行綁定amqpAdmin.declareBinding(new Binding("fanout_queue_email",Binding.DestinationType.QUEUE,"fanout_exchange","",null));amqpAdmin.declareBinding(new Binding("fanout_queue_sms",Binding.DestinationType.QUEUE,"fanout_exchange","",null));} //注意Queue會(huì)報(bào)紅，這時(shí)候要導(dǎo)一下AmQp的包 //運(yùn)行的使用要用rabbitmq-server.bat指令開(kāi)啟服務(wù)，否則會(huì)報(bào)無(wú)法連接rabbitMQ的錯(cuò)誤

② 執(zhí)行上述單元測(cè)試方法amqpAdmin()，驗(yàn)證RabbitMQ消息組件的定制效果，通過(guò)RabbitMQ可視化管理頁(yè)面的Exchanges面板查看效果。

③ 在RabbitMQ可視化管理頁(yè)面的Exchanges面板中新出現(xiàn)了一個(gè)名稱(chēng)為fanout_exchange的交換器,且其類(lèi)型是設(shè)置的fanout類(lèi)型。單擊fanout_exchange交換器進(jìn)入查看。

④ 切換到Queues面板頁(yè)面，查看定制生成的消息隊(duì)列信息

b.消息發(fā)送者發(fā)送消息

① 創(chuàng)建一個(gè)實(shí)體類(lèi)User,將User對(duì)象作為消息的內(nèi)容

public class User {private Integer id;private String username;//自己補(bǔ)充其他//idea右鍵自動(dòng)生成get和set以及toString方法 }

② 在項(xiàng)目測(cè)試類(lèi)中使用Spring框架提供的RabbitTemplate模板類(lèi)實(shí)現(xiàn)消息發(fā)送。

@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 1、Publish/Subscribe工作模式消息發(fā)送端*/@Testpublic void psubPublisher() {User user=new User();user.setId(1);user.setUsername("石頭");rabbitTemplate.convertAndSend("fanout_exchange","",user);}

3、執(zhí)行上述消息發(fā)送的測(cè)試方法psubPublisher()，將出現(xiàn)異常，控制執(zhí)行效果如圖

這個(gè)轉(zhuǎn)換器只支持Spring、byte[]和Serializable序列化后的消息

異常的兩種解決方式：

將實(shí)體類(lèi)實(shí)現(xiàn)JDK自帶的Serializable序列化接口，不推薦，RabbitMQ管理頁(yè)面顯示消息不夠有好
定制其他類(lèi)型的消息轉(zhuǎn)換器
創(chuàng)建一個(gè)RabbitMQ消息配置類(lèi)RabbitMQConfig,并在該配置類(lèi)中通過(guò)@Bean注解自定義了一個(gè)Jackson2JsonMessageConverter類(lèi)型的消息轉(zhuǎn)換器組件，該組件的返回值必須為MessageConverter類(lèi)型

package gzu.edu.com.rabbitmq_demo.config; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig {/*** 定制JSON格式的消息轉(zhuǎn)換器** @return*/@Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();} }

執(zhí)行上述測(cè)試方法，執(zhí)行成功，查看RabbitMQ可視化管理頁(yè)面Queues面板信息，兩個(gè)消息隊(duì)列中各自擁有一條待接收的消息。

單擊某個(gè)隊(duì)列的詳情頁(yè)面，查看具體的信息

c.消息消費(fèi)者接收消息

消息消費(fèi)者接受消息,創(chuàng)建一個(gè)針對(duì)RabbitMQ消息中間件進(jìn)行消息接收和處理的業(yè)務(wù)類(lèi)RabbitMQService，在該類(lèi)中編寫(xiě)如下方法

package gzu.edu.com.rabbitmq_demo.service; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; @Service public class RabbitMQService {/*** Publish/Subscribe工作模式接收，處理郵件業(yè)務(wù)* @param message*/@RabbitListener(queues = "fanout_queue_email")public void psubConsumerEmail(Message message) {byte[] body = message.getBody();String s = new String(body);System.out.println("郵件業(yè)務(wù)接收到消息： "+s);}/*** Publish/Subscribe工作模式接收，處理短信業(yè)務(wù)* @param message*/@RabbitListener(queues = "fanout_queue_sms")public void psubConsumerSms(Message message) {byte[] body = message.getBody();String s = new String(body);System.out.println("短信業(yè)務(wù)接收到消息： "+s);} }

啟動(dòng)項(xiàng)目來(lái)監(jiān)聽(tīng)并接收消息隊(duì)列中的消息。程序啟動(dòng)成功后，立即查看控制臺(tái)打印結(jié)果

2.基于配置類(lèi)的方式

① 打開(kāi)RabbitMQ消息配置類(lèi)RabbitMQConfig，定義fanout類(lèi)型的交換器、不同名稱(chēng)的消息隊(duì)列以及將不同名稱(chēng)的消息隊(duì)列與交換器綁定。

/*** 使用基于配置類(lèi)的方式定制消息組件* @return*/// 1、定義fanout類(lèi)型的交換器@Beanpublic Exchange fanout_exchange(){ return ExchangeBuilder.fanoutExchange("fanout_exchange").build(); } // 2、定義兩個(gè)不同名稱(chēng)的消息隊(duì)列@Beanpublic Queue fanout_queue_email(){ return new Queue("fanout_queue_email"); } @Beanpublic Queue fanout_queue_sms(){ return new Queue("fanout_queue_sms"); } // 3、將兩個(gè)不同名稱(chēng)的消息隊(duì)列與交換器進(jìn)行綁定@Beanpublic Binding bindingEmail(){ return BindingBuilder.bind(fanout_queue_email()).to(fanout_exchange()).with("").noargs();}@Beanpublic Binding bindingSms(){return BindingBuilder.bind(fanout_queue_sms()).to(fanout_exchange()).with("").noargs();}

② 重新運(yùn)行測(cè)試類(lèi)的psubPublisher方法，來(lái)發(fā)送信息，之前定義的消息消費(fèi)者會(huì)消費(fèi)這個(gè)消息。

6.Routing(路由模式)

1.使用基于注解的方式定制消息組件和消息消費(fèi)者

打開(kāi)業(yè)務(wù)類(lèi)RabbitMQService，在該類(lèi)中使用@RabbitListener注解及其相關(guān)屬性定制Routing路由模式的消息組件，并模擬編寫(xiě)消息消費(fèi)者接收的方法。

/*** 2.1、路由模式消息接收，處理error級(jí)別日志信息* @param message*/@RabbitListener(bindings =@QueueBinding(value =@Queue("routing_queue_error"),exchange =@Exchange(value = "routing_exchange",type = "direct"),key = "error_routing_key"))public void routingConsumerError(String message) {System.out.println("接收到error級(jí)別日志消息： "+message);}/*** 2.2、路由模式消息接收，處理info、error、warning級(jí)別日志信息* @param message*/@RabbitListener(bindings =@QueueBinding(value =@Queue("routing_queue_all"),exchange =@Exchange(value = "routing_exchange",type = "direct"),key = {"error_routing_key","info_routing_key","warning_routing_key"}))public void routingConsumerAll(String message) {System.out.println("接收到info、error、warning等級(jí)別日志消息： "+message);}

2.消息發(fā)送者發(fā)送消息

在測(cè)試類(lèi)中使用RabbitTemplate模板類(lèi)實(shí)現(xiàn)Routing路由模式下的消息發(fā)送

/*** 2、Routing工作模式消息發(fā)送端*/@Testpublic void routingPublisher() {rabbitTemplate.convertAndSend("routing_exchange","error_routing_key","routing send error message");}

將測(cè)試方法routingPublisher()中進(jìn)行消息發(fā)送的參數(shù)進(jìn)行修改，調(diào)整發(fā)送info級(jí)別的日志信息（注意同時(shí)修改info_routing_key路由鍵）

再次啟動(dòng)該測(cè)試方法，查看控制臺(tái)執(zhí)行效果，如圖

通過(guò)RabbitMQ可視化管理頁(yè)面查看Routing模式的消息組件，如圖所示

7.Topics(通配符模式)

1.使用基于注解的方式定制消息組件和消息消費(fèi)者

打開(kāi)業(yè)務(wù)類(lèi)RabbitMQService，在該類(lèi)中使用@RabbitListener注解及其相關(guān)屬性定制Topics通配符模式的消息組件，并模擬編寫(xiě)消息消費(fèi)者接收的方法。

/*** 3.1、通配符模式消息接收，進(jìn)行郵件業(yè)務(wù)訂閱處理* @param message*/@RabbitListener(bindings =@QueueBinding(value =@Queue("topic_queue_email"),exchange =@Exchange(value = "topic_exchange",type = "topic"),key = "info.#.email.#"))public void topicConsumerEmail(String message) {System.out.println("接收到郵件訂閱需求處理消息： "+message);}/*** 3.2、通配符模式消息接收，進(jìn)行短信業(yè)務(wù)訂閱處理* @param message*/@RabbitListener(bindings =@QueueBinding(value =@Queue("topic_queue_sms"),exchange =@Exchange(value = "topic_exchange",type = "topic"),key = "info.#.sms.#"))public void topicConsumerSms(String message) {System.out.println("接收到短信訂閱需求處理消息： "+message);}

2.消息發(fā)送者發(fā)送消息

在項(xiàng)目測(cè)試類(lèi)Chapter08ApplicationTests中使用RabbitTemplate模板類(lèi)實(shí)現(xiàn)Routing路由模式下的消息發(fā)送。

/*** 3、Topcis工作模式消息發(fā)送端*/@Testpublic void topicPublisher() {// 1、只發(fā)送郵件訂閱用戶消息 // rabbitTemplate.convertAndSend("topic_exchange","info.email","topics send email message");// 2、只發(fā)送短信訂閱用戶消息 // rabbitTemplate.convertAndSend("topic_exchange","info.sms","topics send sms message");// 3、發(fā)送同時(shí)訂閱郵件和短信的用戶消息rabbitTemplate.convertAndSend("topic_exchange","info.email.sms","topics send email and sms message");}

執(zhí)行測(cè)試方法topicPublisher()，先進(jìn)行郵件訂閱用戶的消息發(fā)送

進(jìn)行短信訂閱用戶的消息發(fā)送

同時(shí)進(jìn)行郵件和短信訂閱用戶的消息發(fā)送方法

通過(guò)RabbitMQ可視化管理頁(yè)面查看自動(dòng)定制的Topics通配符模式的消息組件，使用基于注解的方式自動(dòng)生成了Topics通配符模式下的消息組件，并進(jìn)行了自動(dòng)綁定。

作業(yè) ：Publish/Subscribe（發(fā)布訂閱模式）下，使用注解方式實(shí)現(xiàn)消息服務(wù)

附錄

https://blog.csdn.net/weixin_37641832/article/details/83270778 深入理解AMQP協(xié)議

總結(jié)

以上是生活随笔為你收集整理的8.springBoot消息服务的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。