本文轉(zhuǎn)抄自CSDN平臺
作者：@小目標(biāo)青年
點(diǎn)擊跳轉(zhuǎn)到文章地址

開始

該篇文章內(nèi)容較多，包括有rabbitMq相關(guān)的一些簡單理論介紹，provider消息推送實(shí)例，consumer消息消費(fèi)實(shí)例，Direct、Topic、Fanout的使用，消息回調(diào)、手動確認(rèn)等。（但是關(guān)于rabbitMq的安裝，就不介紹了）

在安裝完rabbitMq后，輸入http://ip:15672/ ，是可以看到一個簡單后臺管理界面的。

image.png

在這個界面里面我們可以做些什么？
可以手動創(chuàng)建虛擬host，創(chuàng)建用戶，分配權(quán)限，創(chuàng)建交換機(jī)，創(chuàng)建隊(duì)列等等，還有查看隊(duì)列消息，消費(fèi)效率，推送效率等等

以上這些管理界面的操作在這篇暫時不做擴(kuò)展描述，我想著重介紹后面實(shí)例里會使用到的。

首先先介紹一個簡單的一個消息推送到接收的流程，提供一個簡單的圖：

image.png

黃色的圈圈就是我們的消息推送服務(wù)，將消息推送到中間方框里面也就是 rabbitMq的服務(wù)器，然后經(jīng)過服務(wù)器里面的交換機(jī)、隊(duì)列等各種關(guān)系（后面會詳細(xì)講）將數(shù)據(jù)處理入列后，最終右邊的藍(lán)色圈圈消費(fèi)者獲取對應(yīng)監(jiān)聽的消息。

常用的交換機(jī)有以下三種，因?yàn)橄M(fèi)者是從隊(duì)列獲取信息的，隊(duì)列是綁定交換機(jī)的（一般），所以對應(yīng)的消息推送/接收模式也會有以下幾種：

Direct Exchange

直連型交換機(jī)，根據(jù)消息攜帶的路由鍵將消息投遞給對應(yīng)隊(duì)列。
大致流程，有一個隊(duì)列綁定到一個直連交換機(jī)上，同時賦予一個路由鍵 routing key 。
然后當(dāng)一個消息攜帶著路由值為X，這個消息通過生產(chǎn)者發(fā)送給交換機(jī)時，交換機(jī)就會根據(jù)這個路由值X去尋找綁定值也是X的隊(duì)列。

Fanout Exchange

扇型交換機(jī)，這個交換機(jī)沒有路由鍵概念，就算你綁了路由鍵也是無視的。這個交換機(jī)在接收到消息后，會直接轉(zhuǎn)發(fā)到綁定到它上面的所有隊(duì)列。

Topic Exchange

主題交換機(jī)，這個交換機(jī)其實(shí)跟直連交換機(jī)流程差不多，但是它的特點(diǎn)就是在它的路由鍵和綁定鍵之間是有規(guī)則的。

簡單地介紹下規(guī)則：
'*' (星號) 用來表示一個單詞 (必須出現(xiàn)的)
'#' (井號) 用來表示任意數(shù)量（零個或多個）單詞
通配的綁定鍵是跟隊(duì)列進(jìn)行綁定的，舉個小例子
隊(duì)列Q1 綁定鍵為 .TT. 隊(duì)列Q2綁定鍵為 TT.#
如果一條消息攜帶的路由鍵為 A.TT.B，那么隊(duì)列Q1將會收到；
如果一條消息攜帶的路由鍵為TT.AA.BB，那么隊(duì)列Q2將會收到；

主題交換機(jī)是非常強(qiáng)大的，為啥這么膨脹？

當(dāng)一個隊(duì)列的綁定鍵為 "#"（井號）的時候，這個隊(duì)列將會無視消息的路由鍵，接收所有的消息。
當(dāng) * (星號) 和 # (井號) 這兩個特殊字符都未在綁定鍵中出現(xiàn)的時候，此時主題交換機(jī)就擁有的直連交換機(jī)的行為。
所以主題交換機(jī)也就實(shí)現(xiàn)了扇形交換機(jī)的功能，和直連交換機(jī)的功能
另外還有 Header Exchange 頭交換機(jī) ，Default Exchange 默認(rèn)交換機(jī)，Dead Letter Exchange 死信交換機(jī)，這幾個該篇暫不做講述。

好了，一些簡單的介紹到這里為止，接下來我們來一起編碼。

本次實(shí)例教程需要創(chuàng)建2個springboot項(xiàng)目，一個 rabbitmq-provider （生產(chǎn)者），一個rabbitmq-consumer（消費(fèi)者）。
首先創(chuàng)建 rabbitmq-provider
pom.xml里用到的jar依賴：

        <!--rabbitmq-->
        <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>

然后application.yml：

ps：里面的虛擬host配置項(xiàng)不是必須的，我自己在rabbitmq服務(wù)上創(chuàng)建了自己的虛擬host，所以我配置了；你們不創(chuàng)建，就不用加這個配置項(xiàng)。

server:
  port: 8021
spring:
  #給項(xiàng)目來個名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    virtual-host: JCcccHost

接著我們先使用下direct exchange(直連型交換機(jī)),創(chuàng)建DirectRabbitConfig.java（對于隊(duì)列和交換機(jī)持久化以及連接使用設(shè)置，在注釋里有說明，后面的不同交換機(jī)的配置就不做同樣說明了）：

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

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

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;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@RestController
public class SendMessageController {
 
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
 
    @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ī)TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

把rabbitmq-provider項(xiàng)目運(yùn)行，調(diào)用下接口:

image.png

因?yàn)槲覀兡壳斑€沒弄消費(fèi)者 rabbitmq-consumer，消息沒有被消費(fèi)的，我們?nèi)abbitMq管理頁面看看，是否推送成功：

image.png

再看看隊(duì)列（界面上的各個英文項(xiàng)代表什么意思，可以自己查查哈，對理解還是有幫助的）：

image.png

很好，消息已經(jīng)推送到rabbitMq服務(wù)器上面了。

接下來，創(chuàng)建rabbitmq-consumer項(xiàng)目：
pom.xml里的jar依賴：

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

然后是 application.yml：

 
server:
  port: 8022
spring:
  #給項(xiàng)目來個名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    virtual-host: JCcccHost

然后一樣，創(chuàng)建DirectRabbitConfig.java（消費(fèi)者單純的使用，其實(shí)可以不用添加這個配置，直接建后面的監(jiān)聽就好，使用注解來讓監(jiān)聽器監(jiān)聽對應(yīng)的隊(duì)列即可。配置上了的話，其實(shí)消費(fèi)者也是生成者的身份，也能推送該消息。）：

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;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class DirectRabbitConfig {
 
    //隊(duì)列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }
 
    //Direct交換機(jī) 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange");
    }
 
    //綁定  將隊(duì)列和交換機(jī)綁定, 并設(shè)置用于匹配鍵：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

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

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

然后將rabbitmq-consumer項(xiàng)目運(yùn)行起來，可以看到把之前推送的那條消息消費(fèi)下來了：

image.png

然后可以再繼續(xù)調(diào)用rabbitmq-provider項(xiàng)目的推送消息接口，可以看到消費(fèi)者即時消費(fèi)消息：

image.png

那么直連交換機(jī)既然是一對一，那如果咱們配置多臺監(jiān)聽綁定到同一個直連交互的同一個隊(duì)列，會怎么樣？

image.png

可以看到是實(shí)現(xiàn)了輪詢的方式對消息進(jìn)行消費(fèi)，而且不存在重復(fù)消費(fèi)。

接著，我們使用Topic Exchange 主題交換機(jī)。
在rabbitmq-provider項(xiàng)目里面創(chuàng)建TopicRabbitConfig.java：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
 
@Configuration
public class TopicRabbitConfig {
    //綁定鍵
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";
 
    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }
 
    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }
 
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }
 
 
    //將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man
    //這樣只要是消息攜帶的路由鍵是topic.man,才會分發(fā)到該隊(duì)列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
 
    //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
    // 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊(duì)列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
}

然后添加多2個接口，用于推送消息到主題交換機(jī)：

 @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: M A N ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> manMap = new HashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.man", manMap);
        return "ok";
    }
 
    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: woman is all ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> womanMap = new HashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

生產(chǎn)者這邊已經(jīng)完事，先不急著運(yùn)行，在rabbitmq-consumer項(xiàng)目上，創(chuàng)建TopicManReceiver.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消費(fèi)者收到消息  : " + testMessage.toString());
    }
}

再創(chuàng)建一個TopicTotalReceiver.java：

package com.elegant.rabbitmqconsumer.receiver;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
 
@Component
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicTotalReceiver消費(fèi)者收到消息  : " + testMessage.toString());
    }
}

同樣，加主題交換機(jī)的相關(guān)配置，TopicRabbitConfig.java（消費(fèi)者一定要加這個配置嗎？不需要的其實(shí)，理由在前面已經(jīng)說過了。）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
 
@Configuration
public class TopicRabbitConfig {
    //綁定鍵
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";
 
    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }
 
    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }
 
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }
 
 
    //將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man
    //這樣只要是消息攜帶的路由鍵是topic.man,才會分發(fā)到該隊(duì)列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
 
    //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
    // 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊(duì)列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
 
}

然后把rabbitmq-provider，rabbitmq-consumer兩個項(xiàng)目都跑起來，先調(diào)用/sendTopicMessage1 接口：

image.png

然后看消費(fèi)者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監(jiān)聽隊(duì)列1，綁定鍵為：topic.man
TopicTotalReceiver監(jiān)聽隊(duì)列2，綁定鍵為：topic.#
而當(dāng)前推送的消息，攜帶的路由鍵為：topic.man

所以可以看到兩個監(jiān)聽消費(fèi)者receiver都成功消費(fèi)到了消息，因?yàn)檫@兩個recevier監(jiān)聽的隊(duì)列的綁定鍵都能與這條消息攜帶的路由鍵匹配上。

image.png

接下來調(diào)用接口/sendTopicMessage2:

image.png

然后看消費(fèi)者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監(jiān)聽隊(duì)列1，綁定鍵為：topic.man
TopicTotalReceiver監(jiān)聽隊(duì)列2，綁定鍵為：topic.#
而當(dāng)前推送的消息，攜帶的路由鍵為：topic.woman

所以可以看到兩個監(jiān)聽消費(fèi)者只有TopicTotalReceiver成功消費(fèi)到了消息。

image.png

接下來是使用Fanout Exchang 扇型交換機(jī)。
同樣地，先在rabbitmq-provider項(xiàng)目上創(chuàng)建FanoutRabbitConfig.java：

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;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
 
@Configuration
public class FanoutRabbitConfig {
 
    /**
     *  創(chuàng)建三個隊(duì)列 ：fanout.A   fanout.B  fanout.C
     *  將三個隊(duì)列都綁定在交換機(jī) fanoutExchange 上
     *  因?yàn)槭巧刃徒粨Q機(jī), 路由鍵無需配置,配置也不起作用
     */
 
 
    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }
 
    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
 
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
 
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
 
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

然后是寫一個接口用于推送消息，

    @GetMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: testFanoutMessage ";
        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);
        rabbitTemplate.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }

接著在rabbitmq-consumer項(xiàng)目里加上消息消費(fèi)類，

FanoutReceiverA.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverA消費(fèi)者收到消息  : " +testMessage.toString());
    }
 
}

FanoutReceiverB.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverB消費(fèi)者收到消息  : " +testMessage.toString());
    }
 
}

FanoutReceiverC.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費(fèi)者收到消息  : " +testMessage.toString());
    }
 
}

然后加上扇型交換機(jī)的配置類，F(xiàn)anoutRabbitConfig.java（消費(fèi)者真的要加這個配置嗎？不需要的其實(shí)，理由在前面已經(jīng)說過了）：

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;
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class FanoutRabbitConfig {
 
    /**
     *  創(chuàng)建三個隊(duì)列 ：fanout.A   fanout.B  fanout.C
     *  將三個隊(duì)列都綁定在交換機(jī) fanoutExchange 上
     *  因?yàn)槭巧刃徒粨Q機(jī), 路由鍵無需配置,配置也不起作用
     */
 
 
    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }
 
    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
 
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
 
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
 
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

最后將rabbitmq-provider和rabbitmq-consumer項(xiàng)目都跑起來，調(diào)用下接口/sendFanoutMessage ：

image.png

然后看看rabbitmq-consumer項(xiàng)目的控制臺情況：

image.png

可以看到只要發(fā)送到 fanoutExchange 這個扇型交換機(jī)的消息，三個隊(duì)列都綁定這個交換機(jī)，所以三個消息接收類都監(jiān)聽到了這條消息。

到了這里其實(shí)三個常用的交換機(jī)的使用我們已經(jīng)完畢了，那么接下來我們繼續(xù)講講消息的回調(diào)，其實(shí)就是消息確認(rèn)（生產(chǎn)者推送消息成功，消費(fèi)者接收消息成功）。

在rabbitmq-provider項(xiàng)目的application.yml文件上，加上消息確認(rèn)的配置項(xiàng)后：

ps：本篇文章使用springboot版本為 2.1.7.RELEASE ;
如果你們在配置確認(rèn)回調(diào)，測試發(fā)現(xiàn)無法觸發(fā)回調(diào)函數(shù)，那么存在原因也許是因?yàn)榘姹緦?dǎo)致的配置項(xiàng)不起效，
可以把publisher-confirms: true 替換為 publisher-confirm-type: correlated

server:
  port: 8021
spring:
  #給項(xiàng)目來個名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服務(wù)器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    virtual-host: JCcccHost
    #消息確認(rèn)配置項(xiàng)
 
    #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)
    publisher-confirms: true
    #確認(rèn)消息已發(fā)送到隊(duì)列(Queue)
    publisher-returns: true

然后是配置相關(guān)的消息確認(rèn)回調(diào)函數(shù)，RabbitConfig.java：

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;
 
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/3
 * @Description :
 **/
@Configuration
public class RabbitConfig {
 
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //設(shè)置開啟Mandatory,才能觸發(fā)回調(diào)函數(shù),無論消息推送結(jié)果怎么樣都強(qiáng)制調(diào)用回調(diào)函數(shù)
        rabbitTemplate.setMandatory(true);
 
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相關(guān)數(shù)據(jù)："+correlationData);
                System.out.println("ConfirmCallback:     "+"確認(rèn)情況："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });
 
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回應(yīng)碼："+replyCode);
                System.out.println("ReturnCallback:     "+"回應(yīng)信息："+replyText);
                System.out.println("ReturnCallback:     "+"交換機(jī)："+exchange);
                System.out.println("ReturnCallback:     "+"路由鍵："+routingKey);
            }
        });
 
        return rabbitTemplate;
    }
 
}

到這里，生產(chǎn)者推送消息的消息確認(rèn)調(diào)用回調(diào)函數(shù)已經(jīng)完畢。
可以看到上面寫了兩個回調(diào)函數(shù)，一個叫 ConfirmCallback ，一個叫 RetrunCallback；
那么以上這兩種回調(diào)函數(shù)都是在什么情況會觸發(fā)呢？

先從總體的情況分析，推送消息存在四種情況：
①消息推送到server，但是在server里找不到交換機(jī)
②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列
③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到
④消息推送成功

那么我先寫幾個接口來分別測試和認(rèn)證下以上4種情況，消息確認(rèn)觸發(fā)回調(diào)函數(shù)的情況：

①消息推送到server，但是在server里找不到交換機(jī)

寫個測試接口，把消息推送到名為‘non-existent-exchange’的交換機(jī)上（這個交換機(jī)是沒有創(chuàng)建沒有配置的）：

    @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        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);
        rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

調(diào)用接口，查看rabbitmq-provuder項(xiàng)目的控制臺輸出情況（原因里面有說，沒有找到交換機(jī)'non-existent-exchange'）：

2019-09-04 09:37:45.197 ERROR 8172 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'JCcccHost', class-id=60, method-id=40)
ConfirmCallback:     相關(guān)數(shù)據(jù)：null
ConfirmCallback:     確認(rèn)情況：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'JCcccHost', class-id=60, method-id=40)

結(jié)論： ①這種情況觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。

②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列

這種情況就是需要新增一個交換機(jī)，但是不給這個交換機(jī)綁定隊(duì)列，我來簡單地在DirectRabitConfig里面新增一個直連交換機(jī)，名叫‘lonelyDirectExchange’，但沒給它做任何綁定配置操作：

    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

然后寫個測試接口，把消息推送到名為‘lonelyDirectExchange’的交換機(jī)上（這個交換機(jī)是沒有任何隊(duì)列配置的）：

    @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        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);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

調(diào)用接口，查看rabbitmq-provuder項(xiàng)目的控制臺輸出情況：

ReturnCallback:     消息：(Body:'{createTime=2019-09-04 09:48:01, messageId=563077d9-0a77-4c27-8794-ecfb183eac80, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回應(yīng)碼：312
ReturnCallback:     回應(yīng)信息：NO_ROUTE
ReturnCallback:     交換機(jī)：lonelyDirectExchange
ReturnCallback:     路由鍵：TestDirectRouting

ConfirmCallback:     相關(guān)數(shù)據(jù)：null
ConfirmCallback:     確認(rèn)情況：true
ConfirmCallback:     原因：null

可以看到這種情況，兩個函數(shù)都被調(diào)用了；
這種情況下，消息是推送成功到服務(wù)器了的，所以ConfirmCallback對消息確認(rèn)情況是true；
而在RetrunCallback回調(diào)函數(shù)的打印參數(shù)里面可以看到，消息是推送到了交換機(jī)成功了，但是在路由分發(fā)給隊(duì)列的時候，找不到隊(duì)列，所以報了錯誤 NO_ROUTE 。

結(jié)論：②這種情況觸發(fā)的是 ConfirmCallback和RetrunCallback兩個回調(diào)函數(shù)。

③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到

這種情況其實(shí)一看就覺得跟①很像，沒錯，③和①情況回調(diào)是一致的，所以不做結(jié)果說明了。
結(jié)論： ③這種情況觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。

④消息推送成功

那么測試下，按照正常調(diào)用之前消息推送的接口就行，就調(diào)用下 /sendFanoutMessage接口，可以看到控制臺輸出：

ConfirmCallback:     相關(guān)數(shù)據(jù)：null
ConfirmCallback:     確認(rèn)情況：true
ConfirmCallback:     原因：null

結(jié)論： ④這種情況觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。
以上是生產(chǎn)者推送消息的消息確認(rèn) 回調(diào)函數(shù)的使用介紹（可以在回調(diào)函數(shù)根據(jù)需求做對應(yīng)的擴(kuò)展或者業(yè)務(wù)數(shù)據(jù)處理）。

接下來我們繼續(xù)，消費(fèi)者接收到消息的消息確認(rèn)機(jī)制。

和生產(chǎn)者的消息確認(rèn)機(jī)制不同，因?yàn)橄⒔邮毡緛砭褪窃诒O(jiān)聽消息，符合條件的消息就會消費(fèi)下來。
所以，消息接收的確認(rèn)機(jī)制主要存在三種模式：

①自動確認(rèn)，這也是默認(rèn)的消息確認(rèn)情況。 AcknowledgeMode.NONE
RabbitMQ成功將消息發(fā)出（即將消息成功寫入TCP Socket）中立即認(rèn)為本次投遞已經(jīng)被正確處理，不管消費(fèi)者端是否成功處理本次投遞。
所以這種情況如果消費(fèi)端消費(fèi)邏輯拋出異常，也就是消費(fèi)端沒有處理成功這條消息，那么就相當(dāng)于丟失了消息。
一般這種情況我們都是使用try catch捕捉異常后，打印日志用于追蹤數(shù)據(jù)，這樣找出對應(yīng)數(shù)據(jù)再做后續(xù)處理。
② 根據(jù)情況確認(rèn)，這個不做介紹
③ 手動確認(rèn) ，這個比較關(guān)鍵，也是我們配置接收消息確認(rèn)機(jī)制時，多數(shù)選擇的模式。
消費(fèi)者收到消息后，手動調(diào)用basic.ack/basic.nack/basic.reject后，RabbitMQ收到這些消息后，才認(rèn)為本次投遞成功。
basic.ack用于肯定確認(rèn)
basic.nack用于否定確認(rèn)（注意：這是AMQP 0-9-1的RabbitMQ擴(kuò)展）
basic.reject用于否定確認(rèn)，但與basic.nack相比有一個限制:一次只能拒絕單條消息

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

著重講下reject，因?yàn)橛袝r候一些場景是需要重新入列的。

channel.basicReject(deliveryTag, true); 拒絕消費(fèi)當(dāng)前消息，如果第二參數(shù)傳入true，就是將數(shù)據(jù)重新丟回隊(duì)列里，那么下次還會消費(fèi)這消息。設(shè)置false，就是告訴服務(wù)器，我已經(jīng)知道這條消息數(shù)據(jù)了，因?yàn)橐恍┰蚓芙^它，而且服務(wù)器也把這個消息丟掉就行。下次不想再消費(fèi)這條消息了。

使用拒絕后重新入列這個確認(rèn)模式要謹(jǐn)慎，因?yàn)橐话愣际浅霈F(xiàn)異常的時候，catch異常再拒絕入列，選擇是否重入列。

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

順便也簡單講講 nack，這個也是相當(dāng)于設(shè)置不消費(fèi)某條消息。

channel.basicNack(deliveryTag, false, true);
第一個參數(shù)依然是當(dāng)前消息到的數(shù)據(jù)的唯一id;
第二個參數(shù)是指是否針對多條消息；如果是true，也就是說一次性針對當(dāng)前通道的消息的tagID小于當(dāng)前這條消息的，都拒絕確認(rèn)。
第三個參數(shù)是指是否重新入列，也就是指不確認(rèn)的消息是否重新丟回到隊(duì)列里面去。

同樣使用不確認(rèn)后重新入列這個確認(rèn)模式要謹(jǐn)慎，因?yàn)檫@里也可能因?yàn)榭紤]不周出現(xiàn)消息一直被重新丟回去的情況，導(dǎo)致積壓。

看了上面這么多介紹，接下來我們一起配置下，看看一般的消息接收手動確認(rèn)是怎么樣的。

在消費(fèi)者項(xiàng)目里，
新建MessageListenerConfig.java上添加代碼相關(guān)的配置代碼：

 
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;
 
/**
 * @Author : JCccc
 * @CreateTime : 2019/9/4
 * @Description :
 **/
@Configuration
public class MessageListenerConfig {
 
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收處理類
 
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認(rèn)是自動確認(rèn)，這里改為手動確認(rèn)消息
        //設(shè)置一個隊(duì)列
        container.setQueueNames("TestDirectQueue");
        //如果同時設(shè)置多個如下： 前提是隊(duì)列都是必須已經(jīng)創(chuàng)建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
 
 
        //另一種設(shè)置隊(duì)列的方法,如果使用這種情況,那么要設(shè)置多個,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);
 
        return container;
    }
 
 
}

對應(yīng)的手動確認(rèn)消息監(jiān)聽類，MyAckReceiver.java（手動確認(rèn)模式需要實(shí)現(xiàn) ChannelAwareMessageListener）：
之前的相關(guān)監(jiān)聽器可以先注釋掉，以免造成多個同類型監(jiān)聽器都監(jiān)聽同一個隊(duì)列。
這里的獲取消息轉(zhuǎn)換，只作參考，如果報數(shù)組越界可以自己根據(jù)格式去調(diào)整。

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;
 
@Component
 
public class MyAckReceiver implements ChannelAwareMessageListener {
 
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
            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("消費(fèi)的主題消息來自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個參數(shù)，手動確認(rèn)可以被批處理，當(dāng)該參數(shù)為 true 時，則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
//          channel.basicReject(deliveryTag, true);//第二個參數(shù)，true會重新放回隊(duì)列，所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時候使用拒絕
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
     //{key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private 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;
    }
}

這時，先調(diào)用接口/sendDirectMessage，給直連交換機(jī)TestDirectExchange 的隊(duì)列TestDirectQueue 推送一條消息，可以看到監(jiān)聽器正常消費(fèi)了下來：

image.png

到這里，我們其實(shí)已經(jīng)掌握了怎么去使用消息消費(fèi)的手動確認(rèn)了。

但是這個場景往往不夠！因?yàn)楹芏嗷锇橹敖o我評論反應(yīng)，他們需要這個消費(fèi)者項(xiàng)目里面，監(jiān)聽的好幾個隊(duì)列都想變成手動確認(rèn)模式，而且處理的消息業(yè)務(wù)邏輯不一樣。

沒有問題，接下來看代碼

場景：除了直連交換機(jī)的隊(duì)列TestDirectQueue需要變成手動確認(rèn)以外，我們還需要將一個其他的隊(duì)列

或者多個隊(duì)列也變成手動確認(rèn)，而且不同隊(duì)列實(shí)現(xiàn)不同的業(yè)務(wù)處理。
那么我們需要做的第一步，往SimpleMessageListenerContainer里添加多個隊(duì)列：

image.png

然后我們的手動確認(rèn)消息監(jiān)聽類，MyAckReceiver.java 就可以同時將上面設(shè)置到的隊(duì)列的消息都消費(fèi)下來。

但是我們需要做不用的業(yè)務(wù)邏輯處理，那么只需要根據(jù)消息來自的隊(duì)列名進(jìn)行區(qū)分處理即可，如：

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;
 
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
 
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費(fèi)的消息來自的隊(duì)列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費(fèi)到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務(wù)處理流程......");
                
            }
 
            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費(fèi)的消息來自的隊(duì)列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費(fèi)到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行fanout.A中的消息的業(yè)務(wù)處理流程......");
 
            }
            
            channel.basicAck(deliveryTag, true);
//          channel.basicReject(deliveryTag, true);//為true會重新放回隊(duì)列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
    //{key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private Map<String, String> mapStringToMap(String str,int enNum) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",enNum);
        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;
    }
}

ok，這時候我們來分別往不同隊(duì)列推送消息，看看效果：

調(diào)用接口/sendDirectMessage 和 /sendFanoutMessage

image.png

如果你還想新增其他的監(jiān)聽隊(duì)列，也就是按照這種方式新增配置即可（或者完全可以分開多個消費(fèi)者項(xiàng)目去監(jiān)聽處理）。

好，這篇Springboot整合rabbitMq教程就暫且到此。

結(jié)束

此篇博文講述了rabbitmq+springboot的使用，本人看完后覺得受益匪淺，而且博主也列出了一些可能在實(shí)際工作中使用的特殊場景，講解的也非常詳細(xì)，博文很長，代碼也比較通透，可見博主的用心，對于初學(xué)rabbitmq的同學(xué)來說，是一篇值得學(xué)習(xí)的博文。同時在這里也很感謝原CSDN博主''@小目標(biāo)青年''的無私分享。

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Springboot 整合RabbitMq ，用心看完這一篇就夠了

Springboot 整合RabbitMq ，用心看完這一篇就夠了

開始

Direct Exchange

Fanout Exchange

Topic Exchange

主題交換機(jī)是非常強(qiáng)大的，為啥這么膨脹？

好了，一些簡單的介紹到這里為止，接下來我們來一起編碼。

①消息推送到server，但是在server里找不到交換機(jī)

②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列

③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到

④消息推送成功

接下來我們繼續(xù)，消費(fèi)者接收到消息的消息確認(rèn)機(jī)制。

著重講下reject，因?yàn)橛袝r候一些場景是需要重新入列的。

看了上面這么多介紹，接下來我們一起配置下，看看一般的消息接收手動確認(rèn)是怎么樣的。

到這里，我們其實(shí)已經(jīng)掌握了怎么去使用消息消費(fèi)的手動確認(rèn)了。

好，這篇Springboot整合rabbitMq教程就暫且到此。

結(jié)束

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Springboot 整合RabbitMq ，用心看完這一篇就夠了

開始

Direct Exchange

Fanout Exchange

Topic Exchange

主題交換機(jī)是非常強(qiáng)大的，為啥這么膨脹？

好了，一些簡單的介紹到這里為止， 接下來我們來一起編碼。

①消息推送到server，但是在server里找不到交換機(jī)

②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列

③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到

④消息推送成功

接下來我們繼續(xù)， 消費(fèi)者接收到消息的消息確認(rèn)機(jī)制。

著重講下reject，因?yàn)橛袝r候一些場景是需要重新入列的。

看了上面這么多介紹，接下來我們一起配置下，看看一般的消息接收 手動確認(rèn)是怎么樣的。

到這里，我們其實(shí)已經(jīng)掌握了怎么去使用消息消費(fèi)的手動確認(rèn)了。

好，這篇Springboot整合rabbitMq教程就暫且到此。

結(jié)束

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

主題交換機(jī)是非常強(qiáng)大的，為啥這么膨脹？

好了，一些簡單的介紹到這里為止，接下來我們來一起編碼。

①消息推送到server，但是在server里找不到交換機(jī)

②消息推送到server，找到交換機(jī)了，但是沒找到隊(duì)列

③消息推送到sever，交換機(jī)和隊(duì)列啥都沒找到

接下來我們繼續(xù)，消費(fèi)者接收到消息的消息確認(rèn)機(jī)制。

著重講下reject，因?yàn)橛袝r候一些場景是需要重新入列的。

看了上面這么多介紹，接下來我們一起配置下，看看一般的消息接收手動確認(rèn)是怎么樣的。

到這里，我們其實(shí)已經(jīng)掌握了怎么去使用消息消費(fèi)的手動確認(rèn)了。

好，這篇Springboot整合rabbitMq教程就暫且到此。