什么是消息隊(duì)列mq

可以看我之前寫的這篇 消息隊(duì)列MQ

rabbitmq簡介

RabbitMQ是由erlang語言開發(fā)，基于AMQP（Advanced Message Queue 高級(jí)消息隊(duì)列協(xié)議）協(xié)議實(shí)現(xiàn)的消息隊(duì)列，它是一種應(yīng)用程序之間的通信方法，消息隊(duì)列在分布式系統(tǒng)開發(fā)中應(yīng)用非常廣泛。

brew安裝rabbitmq

1、安裝
官方文檔

brew install rabbitmq

這是安裝的 log

Management UI: http://localhost:15672
Homebrew-specific docs: https://rabbitmq.com/install-homebrew.htmlTo start rabbitmq now and restart at login:brew services start rabbitmq
Or, if you don't want/need a background service you can just run:CONF_ENV_FILE="/opt/homebrew/etc/rabbitmq/rabbitmq-env.conf" /opt/homebrew/opt/rabbitmq/sbin/rabbitmq-server

2、打開 http://localhost:15672

默認(rèn)的用戶名密碼都是 guest，登錄后可以在 Admin 那一列菜單內(nèi)添加自己的用戶

權(quán)限

超級(jí)管理員(administrator)
可登陸管理控制臺(tái)，可查看所有的信息，并且可以對(duì)用戶，策略(policy)進(jìn)行操作。

監(jiān)控者(monitoring)
可登陸管理控制臺(tái)，同時(shí)可以查看rabbitmq節(jié)點(diǎn)的相關(guān)信息(進(jìn)程數(shù)，內(nèi)存使用情況，磁盤使用情況等)

策略制定者(policymaker)
可登陸管理控制臺(tái), 同時(shí)可以對(duì)policy進(jìn)行管理。但無法查看節(jié)點(diǎn)的相關(guān)信息(上圖紅框標(biāo)識(shí)的部分)。

普通管理者(management)
僅可登陸管理控制臺(tái)，無法看到節(jié)點(diǎn)信息，也無法對(duì)策略進(jìn)行管理。

其他
無法登陸管理控制臺(tái)，通常就是普通的生產(chǎn)者和消費(fèi)者。

創(chuàng)建虛擬主機(jī)（Virtual Hosts）

為了讓各個(gè)用戶可以互不干擾的工作，RabbitMQ添加了虛擬主機(jī)（Virtual Hosts）的概念。其實(shí)就是一個(gè)獨(dú)立的訪問路徑，不同用戶使用不同路徑，各自有自己的隊(duì)列、交換機(jī)，互相不會(huì)影響。

交換機(jī)類型

1. 直連交換機(jī)（Direct Exchange）

特點(diǎn)：

• 基于路由鍵（Routing Key）的完全匹配來路由消息。

應(yīng)用場(chǎng)景：

• 當(dāng)消息需要明確指定到某個(gè)隊(duì)列時(shí)，使用直連交換機(jī)。

舉例說明：

• 假設(shè)有一個(gè)訂單處理系統(tǒng)，其中包含“訂單處理”和“訂單審核”兩個(gè)隊(duì)列。生產(chǎn)者發(fā)送一個(gè)消息時(shí)，指定路由鍵為“order.process”，那么只有綁定了該路由鍵的“訂單處理”隊(duì)列會(huì)收到消息。如果指定路由鍵為“order.review”，則只有“訂單審核”隊(duì)列會(huì)收到消息。

2. 扇形交換機(jī)（Fanout Exchange）

特點(diǎn)：

• 將消息廣播到所有綁定到它的隊(duì)列，忽略路由鍵。如果N個(gè)隊(duì)列綁定到某個(gè)扇型交換機(jī)上，當(dāng)有消息發(fā)送給此扇型交換機(jī)時(shí)，交換機(jī)會(huì)將消息的發(fā)送給這所有的N個(gè)隊(duì)列

應(yīng)用場(chǎng)景：

• 當(dāng)需要將消息發(fā)送給多個(gè)隊(duì)列，且每個(gè)隊(duì)列都需要接收到相同消息的副本時(shí)，使用扇出交換機(jī)。

舉例說明：

• 假設(shè)有一個(gè)實(shí)時(shí)新聞發(fā)布系統(tǒng)，包含“新聞?dòng)嗛喺逜”和“新聞?dòng)嗛喺連”兩個(gè)隊(duì)列。生產(chǎn)者發(fā)布一條新聞消息到扇出交換機(jī)，那么這條消息將被同時(shí)發(fā)送到“新聞?dòng)嗛喺逜”和“新聞?dòng)嗛喺連”兩個(gè)隊(duì)列，實(shí)現(xiàn)廣播效果。

3. 主題交換機(jī)（Topic Exchange）

特點(diǎn)：

• 基于路由鍵和通配符（*和#）的模糊匹配來路由消息。

應(yīng)用場(chǎng)景：

• 當(dāng)需要根據(jù)消息的不同類型或?qū)傩詫⑵渎酚傻讲煌年?duì)列時(shí)，使用主題交換機(jī)。

舉例說明：

• 假設(shè)有一個(gè)日志系統(tǒng)，包含“error.log”、“info.log”和“debug.log”三個(gè)隊(duì)列。生產(chǎn)者發(fā)送一條錯(cuò)誤日志消息，路由鍵為“system.error”。此時(shí)，綁定了“*.error”或“system.#”的隊(duì)列將接收到這條消息，因此“error.log”隊(duì)列會(huì)收到它，而“info.log”和“debug.log”則不會(huì)（除非它們也綁定了匹配的路由鍵）。

* (星號(hào)) 用來表示一個(gè)單詞 (必須出現(xiàn)的)
# (井號(hào)) 用來表示任意數(shù)量（零個(gè)或多個(gè)）單詞

通配的綁定鍵是跟隊(duì)列進(jìn)行綁定的，舉個(gè)小例子
隊(duì)列Q1 綁定鍵為*.TT.* 隊(duì)列Q2綁定鍵為 TT.#
如果一條消息攜帶的路由鍵為 A.TT.B，那么隊(duì)列Q1將會(huì)收到；
如果一條消息攜帶的路由鍵為TT.AA.BB，那么隊(duì)列Q2將會(huì)收到；

4. 頭部交換機(jī)（Headers Exchange）

特點(diǎn)：

• 基于消息的頭部信息進(jìn)行匹配，而不是路由鍵。

應(yīng)用場(chǎng)景：

• 當(dāng)需要根據(jù)消息的復(fù)雜屬性進(jìn)行路由時(shí)，使用頭部交換機(jī)。這種交換機(jī)類型使用較少，因?yàn)樗枰鼜?fù)雜的匹配邏輯。

舉例說明（假設(shè)場(chǎng)景）：

• 假設(shè)有一個(gè)消息處理系統(tǒng)，其中消息具有多個(gè)頭部屬性，如“priority”（優(yōu)先級(jí)）、“type”（類型）等。生產(chǎn)者發(fā)送一條高優(yōu)先級(jí)的緊急消息，并設(shè)置相應(yīng)的頭部屬性。此時(shí)，只有那些設(shè)置了匹配這些頭部屬性（如“priority=high”）的隊(duì)列才會(huì)接收到這條消息。

RabbitMQ的工作原理

組成部分說明：
Broker：消息隊(duì)列服務(wù)進(jìn)程，此進(jìn)程包括兩個(gè)部分：Exchange和Queue
Exchange：消息隊(duì)列交換機(jī)，按一定的規(guī)則將消息路由轉(zhuǎn)發(fā)到某個(gè)隊(duì)列，對(duì)消息進(jìn)行過慮。
Queue：消息隊(duì)列，存儲(chǔ)消息的隊(duì)列，消息到達(dá)隊(duì)列并轉(zhuǎn)發(fā)給指定的消費(fèi)者
Producer：消息生產(chǎn)者，即生產(chǎn)方客戶端，生產(chǎn)方客戶端將消息發(fā)送
Consumer：消息消費(fèi)者，即消費(fèi)方客戶端，接收MQ轉(zhuǎn)發(fā)的消息。

生產(chǎn)者發(fā)送消息流程：
1、生產(chǎn)者和Broker建立TCP連接。
2、生產(chǎn)者和Broker建立通道。
3、生產(chǎn)者通過通道消息發(fā)送給Broker，由Exchange將消息進(jìn)行轉(zhuǎn)發(fā)。
4、Exchange將消息轉(zhuǎn)發(fā)到指定的Queue（隊(duì)列）

消費(fèi)者接收消息流程：
1、消費(fèi)者和Broker建立TCP連接
2、消費(fèi)者和Broker建立通道
3、消費(fèi)者監(jiān)聽指定的Queue（隊(duì)列）
4、當(dāng)有消息到達(dá)Queue時(shí)Broker默認(rèn)將消息推送給消費(fèi)者。
5、消費(fèi)者接收到消息。
6、ack回復(fù)

使用

基本配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring.application.name=zy-rabbitmqspring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=xxx
spring.rabbitmq.password=xxx
#虛擬host 可以不設(shè)置,使用server默認(rèn)host
spring.rabbitmq.virtual-host: JCcccHost

直連交換機(jī)示例

創(chuàng)建一個(gè)配置文件

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectRabbitConfig {//隊(duì)列 起名：TestDirectQueue@Beanpublic Queue TestDirectQueue() {// durable:是否持久化,默認(rèn)是false,持久化隊(duì)列：會(huì)被存儲(chǔ)在磁盤上，當(dāng)消息代理重啟時(shí)仍然存在，暫存隊(duì)列：當(dāng)前連接有效// exclusive:默認(rèn)也是false，只能被當(dāng)前創(chuàng)建的連接使用，而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級(jí)高于durable// autoDelete:是否自動(dòng)刪除，當(dāng)沒有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列，該隊(duì)列會(huì)自動(dòng)刪除。// return new Queue("TestDirectQueue",true,true,false);//一般設(shè)置一下隊(duì)列的持久化就好,其余兩個(gè)就是默認(rèn)falsereturn new Queue("TestDirectQueue",true);}//Direct交換機(jī) 起名：TestDirectExchange@BeanDirectExchange TestDirectExchange() {//  return new DirectExchange("TestDirectExchange",true,true);return new DirectExchange("TestDirectExchange",true,false);}//綁定  //將隊(duì)列和交換機(jī)綁定, 并設(shè)置用于匹配鍵：TestDirectRouting@BeanBinding bindingDirect() {return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");}@BeanDirectExchange lonelyDirectExchange() {return new DirectExchange("lonelyDirectExchange");}
}

寫個(gè)簡單的接口進(jìn)行消息推送（根據(jù)需求也可以改為定時(shí)任務(wù)等等，具體看需求）

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;@RestController
public class SendMessageController {//使用RabbitTemplate,這提供了接收/發(fā)送等等方法@AutowiredRabbitTemplate rabbitTemplate;  @GetMapping("/sendDirectMessage")public String sendDirectMessage() {String messageId = String.valueOf(UUID.randomUUID());String messageData = "test message, hello!";String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));Map<String,Object> map=new HashMap<>();map.put("messageId",messageId);map.put("messageData",messageData);map.put("createTime",createTime);//將消息攜帶綁定鍵值：TestDirectRouting 發(fā)送到交換機(jī)TestDirectExchangerabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);return "ok";}
}

可以看到消息已經(jīng)推送到rabbitMq服務(wù)器上面了。接下來要消費(fèi)這個(gè)消息了。一般來說要開啟另一個(gè)服務(wù)來消費(fèi)了，但是我們是練習(xí)就在當(dāng)前服務(wù)進(jìn)行消費(fèi)。

然后是創(chuàng)建消息接收監(jiān)聽類

@Component
//監(jiān)聽的隊(duì)列名稱 TestDirectQueue
@RabbitListener(queues = "TestDirectQueue")
public class DirectReceiver {@RabbitHandlerpublic void process(Map testMessage) {System.out.println("DirectReceiver消費(fèi)者收到消息  : " + testMessage.toString());}}

運(yùn)行項(xiàng)目，就把這個(gè)消息消費(fèi)掉了。

扇形交換機(jī)、主題交換機(jī)

和上面的差不多，就先不寫了

參考資料

https://blog.csdn.net/weixin_42039228/article/details/123493937