Rabbitmq---訊息佇列

RabbitMQ · 發表 2018-11-14 20:57:00

摘要：一 . MQ:message queue 訊息佇列的作用: 1 通訊解耦 2 高峰限流原理分析: 一開始,認證系統是強耦合的,A系統傳遞認證系統訊息接收計算結果的過程中 1 傳給認證系統 2 認證系統計算 3 返回計算結果 4 讀取A系統邏輯只...

一 . MQ:message queue

訊息佇列的作用: 1 通訊解耦 2 高峰限流

原理分析:

一開始,認證系統是強耦合的,A系統傳遞認證系統訊息接收計算結果的過程中

1 傳給認證系統

2 認證系統計算

3 返回計算結果

4 讀取A系統邏輯

只要當前計算沒有完成,對於認證系統來講消耗執行緒資源.並存在強耦合現象

有了訊息佇列,每一次連線不管是生成訊息還是消費訊息,都有各自的邏輯與其他邏輯無關--通訊解耦

通訊強耦合的情況下高峰訪問拒絕,達到了高峰限流的效果

二 . Rabbitmq

1 rabbitmq的結構(元件)

外部: 生產者和消費者

生產者:對於訊息來講,生成訊息客戶端是生產者

消費者:消費訊息執行消費後的邏輯的客戶端是消費者

客戶端可以使用各種其他技術或者語言都不是rabbitmq自身的技術

2 rabbitmq內部元件

connection:基於底層通訊邏輯的長連線

channel:基於長連線建立的;可以在一次長連線的基礎上多次頻繁的建立和銷燬,佔用資源非常少

交換機Exchange 優點:併發能力高併發穩定

客戶端連線傳送訊息,多種情況都需要併發的傳送,如果從客戶端執行併發的傳送邏輯,rabbitmq的併發能力就限制在了客戶端

exchange基於erlang語言開發的(併發,可控制執行緒)

queue:rabbitmq中的佇列,根據不同的情況佇列可以完成不同的工作

使用場景:在專案中，將一些無需即時返回且耗時的操作提取出來，進行了非同步處理，而這種非同步處理的方式大大的節省了伺服器的請求響應時間，從而提高了系統的吞吐量。

三 . rabbitmq的五種工作模式

1 簡單模式

1 )一個生產者將訊息交給預設的交換機(AMQP default)

2 )交換機獲取訊息後交給繫結的這個生產者的佇列(其中關係是通過佇列名稱完成的)

3 )監聽當前佇列的消費者獲取訊息,執行消費邏輯

應用場景:簡訊聊天

2 工作模式(資源爭搶)

1 )生產者將訊息交給交換機

2 )交換機交給繫結的佇列

3 )佇列由多個消費者同時監聽,只有其中一個能獲取者一條訊息,形成了資源的爭搶,誰的資源空閒大,爭搶到的可能越大

3 釋出訂閱(publish/fanout)

1 )生產者扔給交換機訊息

2 )交換機根據自身的型別(fanout)將會把所有訊息複製同步到所有與其繫結的佇列

3 )每個佇列可以有一個消費者,接收訊息進行消費邏輯

4 路由模式(routing/dircet)

1 )生產者還是將訊息傳送給交換機,訊息攜帶具體的路由key(routingKey)

2 )交換機型別direct,將接受到訊息中的routingKey與之繫結佇列的routingKey比對

3 )消費者監聽一個佇列,獲取訊息,執行消費邏輯.

5 topic主題模式

1 )生產端傳送訊息,訊息攜帶具體的路由key

2 )交換機的型別topic

3 )佇列繫結交換機不在使用具體的路由key而是一個範圍值

與路由模式的區別:路由模式中的queue繫結攜帶的具體的key值,路由細化劃分,topic主題模式queue攜帶的key是一個範圍的匹配,某一類訊息的獲取

四 . rabbitmq的安裝及虛擬機器和使用者名稱的建立

https://www.cnblogs.com/nanlinghan/p/9960361.html

五 . springboot整合rabbitmq

整合思路 :1 )配置資訊 2 )配置類 (明確初始化物件,初始化一個工廠,就從工廠獲取連線;生產端邏輯實現簡單;) 3 )二次封裝

springboot

自動配置:掃描同名,下級包

根據依賴的檔案自動配置需要的內容(一旦依賴了jdbc,不配置datasource就報錯)

springboot就會根據依賴的amqp自動建立連線訊息佇列的內部connection,封裝一個對外呼叫的物件rabbitTemplate模板物件;做生產邏輯

利用底層連線,實現非同步非阻塞監聽,只需要在方法上使用註解,就可以將監聽內容傳遞給方法的引數;

1 新增依賴

<starter-amqp>簡化依賴

groupId在maven庫是一個具有結構的資料夾

artifactId是一整個資料夾

version是一個資料夾

org\springframework\boot\spring-boot-starter-amqp

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

2 配置檔案(application.properties)

spring.rabbitmq.host=10.9.100.26
spring.rabbitmq.port=5672
spring.rabbitmq.username=easymall
spring.rabbitmq.password=123456
spring.rabbitmq.virtualHost=/easymall

3 配置檔案中宣告物件

內部完成各種連線channel等的封裝
宣告需要的佇列和交換機
配置類中完成這種宣告,並且交換機,佇列的物件由框架維護,何時呼叫由業務邏輯決定;

 1 package com.jt.config;
 2 import org.springframework.amqp.core.Binding;
 3 import org.springframework.amqp.core.BindingBuilder;
 4 import org.springframework.amqp.core.DirectExchange;
 5 import org.springframework.amqp.core.Queue;
 6 import org.springframework.context.annotation.Bean;
 7 import org.springframework.context.annotation.Configuration;
 8 @Configuration
 9 public class RabbitmqConfig {
10//路由模式做案例
11//準備2個queue,1個路由交換機
12//宣告2個佇列
13@Bean 
14public Queue queue01(){
15//org.springframework.amqp.core.
16return new Queue("springboot-q1", false);
17}
18@Bean 
19public Queue queue02(){
20//org.springframework.amqp.core.
21return new Queue("springboot-q2", false);
22} 
23@Bean//宣告交換機
24public DirectExchange ex(){
25return new DirectExchange("springboot-ex");
26}
27//宣告繫結關係
28@Bean
29public Binding bind01(){
30return BindingBuilder.
31bind(queue01()).to(ex()).with("item.add");
32}
33@Bean
34public Binding bind02(){
35return BindingBuilder.
36bind(queue02()).to(ex()).with("item.update");}}

4 編寫生成者程式碼

編寫到一個controller中,接收訪問的引數,利用引數作為訊息傳送到交換機攜帶路由key

 1 package com.jt.controller;
 2 
 3 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 4 import org.springframework.beans.factory.annotation.Autowired;
 5 import org.springframework.stereotype.Controller;
 6 import org.springframework.web.bind.annotation.RequestMapping;
 7 import org.springframework.web.bind.annotation.ResponseBody;
 8 
 9 @Controller
10 public class MsgController {
11@Autowired
12private RabbitTemplate rabbit;
13/*
14* 接收訊息msg
15*/
16@RequestMapping("msg")
17@ResponseBody
18public String sendMsg(String msg,String routingKey){
19//注入自動配置的rabbitTemplate物件傳送訊息
20rabbit.convertAndSend("springboot-ex", routingKey, msg);
21return "success";
22}
23 }

5 消費端

2個消費者,監聽2個佇列,傳送的不同路由key會在不同的消費端完成消費邏輯

 1 package com.jt.component;
 2 
 3 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 4 import org.springframework.stereotype.Component;
 5 @Component
 6 public class ConsumerCon {
 7//實現消費邏輯,需要完成非同步監聽
 8//監聽註解會繫結佇列的消費者和當前的方法,一旦消費者獲取訊息,會把訊息內容傳遞給
 9//自定義的方法引數msg
10@RabbitListener(queues="springboot-q1")
11public void process01(String msg){
12System.out.println("消費者1接收商品新增:"+msg);
13}
14
15@RabbitListener(queues="springboot-q2")
16public void process02(String msg){
17System.out.println("消費者2接收到商品更新:"+msg);
18}
19 }