1、為什么需要Mq

例如在用戶注冊業(yè)務中，用戶注冊成功后

需要發(fā)注冊郵件和注冊短信，傳統(tǒng)的做法有兩種

• 1.串行的方式；
• 2.并行的方式 ；

假設三個業(yè)務節(jié)點分別使用50ms，串行方式使用時間150ms，并行使用時間100ms。雖然并性已經提高的處理時間，但是，前面說過郵件和短信對我正常的使用網站沒有任何影響，客戶端沒有必要等著其發(fā)送完成才顯示注冊成功，應該是寫入數據庫后就返回。

2、Mq的優(yōu)缺點

• 優(yōu)點：

1、應用解耦 ：每個服務都可以靈活插拔，可替換，服務沒有直接調用，不存在級聯(lián)失敗問題

2、流量削峰：不管發(fā)布事件的流量波動多大，都由Broker接收，訂閱者可以按照自己的速度去處理事件

用戶的請求，服務器收到之后，首先寫入消息隊列，加入消息隊列長度超過最大值，則直接拋棄用戶請求或跳轉到錯誤頁面。

秒殺業(yè)務根據消息隊列中的請求信息，再做后續(xù)處理

3、吞吐量提升：無需等待訂閱者處理完成，響應更快速

• 缺點：

  • 架構復雜了，業(yè)務沒有明顯的流程線，不好管理

  • 需要依賴于Broker的可靠、安全、性能

3、常用的Mq對比

ActiveMQ? RabbitMQ? RocketMQ? Kafka

1、ActiveMQ，性能不是很好，因此在高并發(fā)的場景下，直接被pass掉了。它的Api很完善，在中小型互聯(lián)網公司可以去使用。
2、kafka，主要強調高性能，如果對業(yè)務需要可靠性消息的投遞的時候。那么就不能夠選擇kafka了。但是如果做一些日志收集呢，kafka還是很好的。因為kafka的性能是十分好的。
3、RocketMQ，它的特點非常好。它高性能、滿足可靠性、分布式事物、支持水平擴展、上億級別的消息堆積、主從之間的切換等等。MQ的所有優(yōu)點它基本都滿足。但是它最大的缺點：商業(yè)版收費。因此它有許多功能是不對外提供的。

4、RabbitMQ安裝

mac 可參考Mac RabbitMQ安裝_mac安裝rabbitmq-CSDN博客

其他版本也可搜索后按步驟安裝，或者使用docker鏡像安裝

5、基礎概念

RabbitMQ中的一些角色：

• publisher：生產者
• consumer：消費者
• exchange個：交換機，負責消息路由
• queue：隊列，存儲消息
• virtualHost：虛擬主機，隔離不同租戶的exchange、queue、消息的隔離

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

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

6、代碼實現

官方的API實現消息接收和發(fā)送比較繁瑣，SpringAMQP則是基于RabbitMQ封裝的一套模板，并且利用SpringBoot對其實現了自動裝配，使用起來非常方便。

1、導入依賴，發(fā)送方和接收方都導入

<!--AMQP依賴，包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、配置MQ地址，在publisher和consumer服務的application.yml中添加配置

spring:rabbitmq:host: 192.168.150.101 # 主機名port: 5672 # 端口virtual-host: / # 虛擬主機username: itcast # 用戶名password: 123321 # 密碼

3、發(fā)送方使用RabbitTemplate發(fā)送 消息

package cn.itcast.mq.spring;import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSimpleQueue() {// 隊列名稱String queueName = "simple.queue";// 消息String message = "hello, spring amqp!";// 發(fā)送消息rabbitTemplate.convertAndSend(queueName, message);}
}

4、接收方使用@RabbitListener注解接收消息

package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class SpringRabbitListener {// 表明需要監(jiān)聽的隊列名稱，可自動裝配下述參數msg@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("spring 消費者接收到消息：【" + msg + "】");}
}

說明：消息一旦被消費就會從隊列中刪除(只能讀取一次)，RabbitMQ沒有消息回溯功能。

7、WorkQueue工作隊列

Work queues任務模型。簡單來說就是讓多個消費者綁定到一個隊列，共同消費隊列中的消息，提高處理速度。

我們已經知道消息讀取后即會從隊列中刪除，當消息處理比較耗時的時候，可能生產消息的速度會遠遠大于消息的消費速度。長此以往，消息就會堆積越來越多，無法及時處理。

此時就可以使用work 模型，多個消費者共同處理消息處理，速度就能大大提高了。

/*** workQueue* 向隊列中不停發(fā)送消息，模擬消息堆積。*/
@Test
public void testWorkQueue() throws InterruptedException {// 隊列名稱String queueName = "simple.queue";// 消息String message = "hello, message_";for (int i = 0; i < 50; i++) {// 發(fā)送消息rabbitTemplate.convertAndSend(queueName, message + i);// 控制1s發(fā)送50條消息Thread.sleep(20);}
}// 兩個消費者綁定到同一個隊列
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消費者1接收到消息：【" + msg + "】" + LocalTime.now());// 模擬不同消費者處理能力差異Thread.sleep(20);
}// 兩個消費者綁定到同一個隊列
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {System.err.println("消費者2........接收到消息：【" + msg + "】" + LocalTime.now());// 模擬不同消費者處理能力差異Thread.sleep(200);
}

默認情況下消息會被平均分配給了每個消費者，，并沒有考慮到消費者的處理能力

由于RabbitMQ采取的是消息預取機制，當有消息發(fā)送過來時會將消息都投遞給消費者。

我們可以修改consumer服務的application.yml文件，設置preFetch參數，控制預取消息的上限：

spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能獲取一條消息，處理完成才能獲取下一個消息

8、發(fā)布訂閱模式

發(fā)布訂閱模式與之前案例的區(qū)別就是允許將同一消息發(fā)送給多個消費者。

可以看到，在訂閱模型中，多了一個exchange角色，而且過程略有變化：

??? Publisher：生產者，也就是要發(fā)送消息的程序，但是不再發(fā)送到隊列中，而是發(fā)給交換機
??? Exchange：交換機，。一方面，接收生產者發(fā)送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決于Exchange的類型。Exchange有以下3種類型：
??????? Fanout：廣播，將消息交給所有綁定到交換機的隊列
??????? Direct：定向，把消息交給符合指定routing key 的隊列
??????? Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列
??? Consumer：消費者，與以前一樣，訂閱隊列，沒有變化
??? Queue：消息隊列也與以前一樣，接收消息、緩存消息。

Exchange（交換機）只負責轉發(fā)消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規(guī)則的隊列，那么消息會丟失！

1、Fanout廣播

在廣播模式下，消息發(fā)送流程是這樣的：

• 可以有多個隊列
• 每個隊列都要綁定到Exchange（交換機）
• 生產者發(fā)送的消息，只能發(fā)送到交換機，交換機來決定要發(fā)給哪個隊列，生產者無法決定
• 交換機把消息發(fā)送給綁定過的所有隊列
• 訂閱隊列的消費者都能拿到消息

聲明2個交換機和隊列，然后綁定

package cn.itcast.mq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutConfig {/*** 聲明交換機* @return Fanout類型交換機*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("itcast.fanout");}/*** 第1個隊列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 綁定隊列和交換機* 注意參數名fanoutQueue1，spring會自動裝配*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2個隊列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 綁定隊列和交換機* 注意參數名fanoutQueue2，spring會自動裝配*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

發(fā)送方指明交換機

@Test
public void testFanoutExchange() {// 隊列名稱String exchangeName = "itcast.fanout";// 消息String message = "hello, everyone!";// 中間參數后期有用，暫時不用寫rabbitTemplate.convertAndSend(exchangeName, "", message);
}

接收方

// 綁定隊列
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {System.out.println("消費者1接收到Fanout消息：【" + msg + "】");
}// 綁定隊列
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {System.out.println("消費者2接收到Fanout消息：【" + msg + "】");
}

2、Direct路由

在Fanout模式中，一條消息，會被所有訂閱的隊列都消費。但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。

在Direct模型下：

??? 隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）
??? 消息的發(fā)送方在向 Exchange發(fā)送消息時，也必須指定消息的 RoutingKey。
??? Exchange不再把消息交給每一個綁定的隊列，而是根據消息的Routing Key進行判斷，只有隊列的Routingkey與消息的 Routing key完全一致，才會接收到消息

消息發(fā)送方，指明交換機和routing key

@Test
public void testSendDirectExchange() {// 交換機名稱String exchangeName = "itcast.direct";// 消息String message = "hello red";// 發(fā)送消息，通過中間參數指定key值rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

基于@Bean的方式聲明隊列和交換機比較麻煩，需要聲明多個，Spring還提供了基于注解方式來聲明。

在consumer的SpringRabbitListener中添加兩個消費者，同時基于注解來聲明隊列和交換機：

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消費者接收到direct.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消費者接收到direct.queue2的消息：【" + msg + "】");
}

• bindings = @QueueBinding():聲明并綁定關系
• value = @Queue():隊列名稱
• exchange = @Exchange():交換機名稱，發(fā)布訂閱模式
• key = {}:指定路由key

只有綁定了對應key的才能接收到消息

3、Topic主題

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符！

Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符規(guī)則：

#：匹配一個或多個詞

*：匹配不多不少恰好1個詞

舉例：

china.#：能夠匹配china.spu.nb 或者 item.spu

china.*：只能匹配china.spu

發(fā)送方

/*** topicExchange*/
@Test
public void testSendTopicExchange() {// 交換機名稱String exchangeName = "itcast.topic";// 消息String message = "喜報！孫悟空大戰(zhàn)哥斯拉，勝!";// 發(fā)送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

接收方

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),key = "china.#"
))
public void listenTopicQueue1(String msg){System.out.println("消費者接收到topic.queue1的消息：【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),key = "#.news"
))
public void listenTopicQueue2(String msg){System.out.println("消費者接收到topic.queue2的消息：【" + msg + "】");
}

9、消息轉換器

在SpringAMQP的發(fā)送消息中，發(fā)送接收消息的類型是Object,也就是說我們可以發(fā)送任意對象類型的消息，SpringAMQP會幫我們序列化為字節(jié)后發(fā)送，接收消息的時候，還會把字節(jié)反序列化為Java對象。

默認情況下Spring采用的序列化方式是JDK序列化

JDK序列化存在下列問題：

• 數據體積過大
• 有安全漏洞
• 可讀性差

可以使用JSON方式來做序列化和反序列化

引入jackson依賴

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

在配置類或啟動類中聲明一個序列化轉換類

@Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();
}

10、消息確認

就是發(fā)送方確認消息發(fā)送成功，接收方接收成功后，可以執(zhí)行回調函數

配置文件開啟回調

rabbitmq:host: 127.0.0.1port: 5672username: rootpassword: root#虛擬host 可以不設置,使用server默認hostvirtual-host: JCcccHost#消息確認配置項#確認消息已發(fā)送到交換機(Exchange)publisher-confirms: true#確認消息已發(fā)送到隊列(Queue)publisher-returns: true

設置回調函數

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//設置開啟Mandatory,才能觸發(fā)回調函數,無論消息推送結果怎么樣都強制調用回調函數rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("ConfirmCallback:     "+"相關數據："+correlationData);System.out.println("ConfirmCallback:     "+"確認情況："+ack);System.out.println("ConfirmCallback:     "+"原因："+cause);}});rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("ReturnCallback:     "+"消息："+message);System.out.println("ReturnCallback:     "+"回應碼："+replyCode);System.out.println("ReturnCallback:     "+"回應信息："+replyText);System.out.println("ReturnCallback:     "+"交換機："+exchange);System.out.println("ReturnCallback:     "+"路由鍵："+routingKey);}});return rabbitTemplate;}
}

消費方消息確認

和生產者的消息確認機制不同，因為消息接收本來就是在監(jiān)聽消息，符合條件的消息就會消費下來。

所以，消息接收的確認機制主要存在三種模式：
1、自動確認， 這也是默認的消息確認情況。 AcknowledgeMode.NONE
RabbitMQ成功將消息發(fā)出（即將消息成功寫入TCP Socket）中立即認為本次投遞已經被正確處理，不管消費者端是否成功處理本次投遞。
所以這種情況如果消費端消費邏輯拋出異常，也就是消費端沒有處理成功這條消息，那么就相當于丟失了消息。
一般這種情況我們都是使用try catch捕捉異常后，打印日志用于追蹤數據，這樣找出對應數據再做后續(xù)處理。

2、根據情況確認， 這個不做介紹

3、手動確認 ， 這個比較關鍵，也是我們配置接收消息確認機制時，多數選擇的模式。
消費者收到消息后，手動調用basic.ack/basic.nack/basic.reject后，RabbitMQ收到這些消息后，才認為本次投遞成功。
basic.ack用于肯定確認
basic.nack用于否定確認（注意：這是AMQP 0-9-1的RabbitMQ擴展）
basic.reject用于否定確認，但與basic.nack相比有一個限制：一次只能拒絕單條消息

消費者端以上的3``個方法都表示消息已經被正確投遞，但是basic.ack表示消息已經被正確處理。
而basic.nack,basic.reject表示沒有被正確處理：

著重講下reject，因為有時候一些場景是需要重新入列的。

channel.basicReject(deliveryTag, true); 拒絕消費當前消息，如果第二參數傳入true，就是將數據重新丟回隊列里，那么下次還會消費這消息。設置false，就是告訴服務器，我已經知道這條消息數據了，因為一些原因拒絕它，而且服務器把這個消息丟掉就行，下次不想再消費這條消息了。

使用拒絕后重新入列這個確認模式要謹慎，因為一般都是出現異常的時候，catch異常再拒絕入列，選擇是否重入列。

但是如果使用不當會導致一些每次都被你重入列的消息一直消費-入列-消費-入列這樣循環(huán)，會導致消息積壓。

nack，這個也是相當于設置不消費某條消息。

channel.basicNack(deliveryTag, false, true);

第一個參數依然是當前消息到的數據的唯一id;
第二個參數是指是否針對多條消息；如果是true，也就是說一次性針對當前通道的消息的tagID小于當前這條消息的，都拒絕確認。
第三個參數是指是否重新入列，也就是指不確認的消息是否重新丟回到隊列里面去。

同樣使用不確認后重新入列這個確認模式要謹慎，因為這里也可能因為考慮不周出現消息一直被重新丟回去的情況，導致積壓。

import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MessageListenerConfig {@Autowiredprivate CachingConnectionFactory connectionFactory;@Autowiredprivate MyAckReceiver myAckReceiver;//消息接收處理類@Beanpublic SimpleMessageListenerContainer simpleMessageListenerContainer() {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);container.setConcurrentConsumers(1);container.setMaxConcurrentConsumers(1);// RabbitMQ默認是自動確認，這里改為手動確認消息container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設置一個隊列container.setQueueNames("TestDirectQueue");//如果同時設置多個如下： 前提是隊列都是必須已經創(chuàng)建存在的//  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");//另一種設置隊列的方法,如果使用這種情況,那么要設置多個,就使用addQueues//container.setQueues(new Queue("TestDirectQueue",true));//container.addQueues(new Queue("TestDirectQueue2",true));//container.addQueues(new Queue("TestDirectQueue3",true));container.setMessageListener(myAckReceiver);return container;}
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;@Componentpublic class MyAckReceiver implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {long deliveryTag = message.getMessageProperties().getDeliveryTag();try {//因為傳遞消息的時候用的map傳遞,所以將Map從Message內取出需要做些處理String msg = message.toString();String[] msgArray = msg.split("'");//可以點進Message里面看源碼,單引號直接的數據就是我們的map消息數據Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);String messageId=msgMap.get("messageId");String messageData=msgMap.get("messageData");String createTime=msgMap.get("createTime");System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);System.out.println("消費的主題消息來自："+message.getMessageProperties().getConsumerQueue());channel.basicAck(deliveryTag, true); //第二個參數，手動確認可以被批處理，當該參數為 true 時，則可以一次性確認 delivery_tag 小于等于傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個參數，true會重新放回隊列，所以需要自己根據業(yè)務邏輯判斷什么時候使用拒絕} catch (Exception e) {channel.basicReject(deliveryTag, false);e.printStackTrace();}}//{key=value,key=value,key=value} 格式轉換成mapprivate Map<String, String> mapStringToMap(String str,int entryNum ) {str = str.substring(1, str.length() - 1);String[] strs = str.split(",",entryNum);Map<String, String> map = new HashMap<String, String>();for (String string : strs) {String key = string.split("=")[0].trim();String value = string.split("=")[1];map.put(key, value);}return map;}
}

RabbitMQ詳解，用心看完這一篇就夠了【重點】-CSDN博客

【微服務】RabbitMQ&SpringAMQP消息隊列_springamqp rabbitmq實現動態(tài)發(fā)送指定人-CSDN博客