欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

RabbitMQ交換機(jī)與Springboot整合的簡單實現(xiàn)

 更新時間:2021年07月24日 09:13:44   作者:Pluto372  
這篇文章主要介紹了RabbitMQ交換機(jī)與Springboot整合的簡單實現(xiàn),本文通過實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下

RabbitMQ-交換機(jī)

1、交換機(jī)是干什么的?
消息(Message)由Client發(fā)送,RabbitMQ接收到消息之后通過交換機(jī)轉(zhuǎn)發(fā)到對應(yīng)的隊列上面。Worker會從隊列中獲取未被讀取的數(shù)據(jù)處理。

在這里插入圖片描述
在這里插入圖片描述

1、交換機(jī)的種類
RabbitMQ包含四種不同的交換機(jī)類型:

  • Direct exchange:直連交換機(jī),轉(zhuǎn)發(fā)消息到routigKey指定的隊列
  • Fanout exchange:扇形交換機(jī),轉(zhuǎn)發(fā)消息到所有綁定隊列(速度最快)
  • Topic exchange:主題交換機(jī),按規(guī)則轉(zhuǎn)發(fā)消息(最靈活)
  • Headers exchange:首部交換機(jī) (不常用)

1、Direct exchange(直連交換機(jī))

直連交換機(jī)是一種帶路由功能的交換機(jī),根據(jù)消息攜帶的路由鍵將消息投遞給對應(yīng)隊列。
一個隊列通過routing_key(路由鍵)與一個交換機(jī)綁定,當(dāng)消息被發(fā)送的時候,需要指定一個routing_key,這個消息被送達(dá)交換機(jī)的時候,就會被交換機(jī)送到指定的隊列里面去。
類似一對一的關(guān)系!

在這里插入圖片描述

1.1 Springboot的簡單實現(xiàn)

創(chuàng)建2個springboot項目,一個 rabbitmq-provider (生產(chǎn)者),一個rabbitmq-consumer(消費者)。

1、首先創(chuàng)建 rabbitmq-provider

pom.xml添加依賴:

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

配置application.yml:

server:
  port: 8090
spring:
  application:
    name: rabbitmq-provider
  rabbitmq:
    host: 192.168.152.173
    port: 5672     #注意15672是web頁面端口,消息發(fā)送端口為5672
    username: guest
    password: guest

編寫配置文件

@Configuration
public class DirectRabbitConfig {

    /**
     * 創(chuàng)建消息隊列 起名:TestDirectQueue
     * durable:是否持久化,默認(rèn)是false,持久化隊列:會被存儲在磁盤上,當(dāng)消息代理重啟時仍然存在,暫存隊列:當(dāng)前連接有效
     * exclusive:默認(rèn)也是false,只能被當(dāng)前創(chuàng)建的連接使用,而且當(dāng)連接關(guān)閉后隊列即被刪除。此參考優(yōu)先級高于durable
     * autoDelete:是否自動刪除,當(dāng)沒有生產(chǎn)者或者消費者使用此隊列,該隊列會自動刪除。
     * 一般設(shè)置一下隊列的持久化就好,其余兩個就是默認(rèn)false
     * @return new Queue(name,durable,exclusive,autoDelete)
     */
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }

    /**
     * Direct交換機(jī) 起名:TestDirectExchange
     */
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }

    /**
     * 綁定 
     * 將隊列和交換機(jī)綁定, 并設(shè)置用于匹配鍵(路由鍵):TestDirectRouting
     */
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }   
}

編寫controller層

@RestController
@RequestMapping("/rabbit")
public class SendMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法

    @RequestMapping("/send")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "League of Legends never dies!";
        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項目運行,調(diào)用下接口:

在這里插入圖片描述

因為我們目前還沒弄消費者 rabbitmq-consumer,消息沒有被消費的,我們?nèi)abbitMq管理頁面看看,是否推送成功:

在這里插入圖片描述

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

2、創(chuàng)建rabbitmq-consumer項目:
pom.xml和application.yml跟上邊一樣(把端口號和應(yīng)用名換了即可)

創(chuàng)建消息接收監(jiān)聽類,DirectReceiver01.java:

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

啟動消費者項目:
運行結(jié)果如下

在這里插入圖片描述

兩個消息直接被消費了。然后可以再繼續(xù)調(diào)用rabbitmq-provider項目的推送消息接口,可以看到消費者即時消費消息!

擴(kuò)展
直連交換機(jī)既然是一對一,那如果咱們配置多臺監(jiān)聽綁定到同一個直連交互的同一個隊列,會怎么樣?
直接復(fù)制兩份DirectReceiver.java,改下打印內(nèi)容!

在這里插入圖片描述

重新啟動項目!在生產(chǎn)者多發(fā)送幾次消息。運行結(jié)果如下:

在這里插入圖片描述

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

2、Topic Exchange(主題交換機(jī))

直連交換機(jī)的routing_key方案非常簡單,但是它是一對一的關(guān)系,那么我們需要一對多呢?希望一條消息發(fā)送給多個隊列。所以RabbitMQ提供了一種主題交換機(jī),發(fā)送到主題交換機(jī)上的消息需要攜帶指定規(guī)則的routing_key,主題交換機(jī)會根據(jù)這個規(guī)則將數(shù)據(jù)發(fā)送到對應(yīng)的(多個)隊列上。

主題交換機(jī)的routing_key需要有一定的規(guī)則,交換機(jī)和隊列的binding_key需要采用 *.# 的格式,每個部分用.分開,其中:
*表示一個單詞
#表示任意數(shù)量(零個或多個)單詞。(例如:topic.man)

在這里插入圖片描述

1、在rabbitmq-provider項目里面創(chuàng)建TopicRabbitConfig.java:

@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ā)到該隊列
      @Bean
      Binding bindingExchangeMessage() {
          return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
      }

      //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
      // 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊列
      @Bean
      Binding bindingExchangeMessage2() {
          return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
      }
}

編寫controller層

@RequestMapping("rabbit")
@RestController
public class TopicSendController {
    @Autowired
    private RabbitTemplate template;

    @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);
        template.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);
        template.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

不要急著去運行,先去編寫消費者,創(chuàng)建兩個接口

TopicManReceiver :

@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消費者收到消息  : " + testMessage.toString());
    }
}

TopicWoManReceiver :

@Component
@RabbitListener(queues = "topic.woman")
public class TopicWoManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicWoManReceiver消費者收到消息  : " + testMessage.toString());
    }
}

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

在這里插入圖片描述

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

在這里插入圖片描述

所以可以看到兩個監(jiān)聽消費者receiver都成功消費到了消息,因為這兩個recevier監(jiān)聽的隊列的綁定鍵都能與這條消息攜帶的路由鍵匹配上。

在這里插入圖片描述

用/sendTopicMessage2接口:

在這里插入圖片描述

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

在這里插入圖片描述

所以可以看到兩個監(jiān)聽消費者,只有TopicWoManReceiver 成功消費到了消息,

在這里插入圖片描述

3、Fanout exchange(扇形交換機(jī))

扇形交換機(jī)是最基本的交換機(jī)類型,它做的事情很簡單–廣播信息。Fanout交換機(jī)會把接收到的消息全部轉(zhuǎn)發(fā)到綁定的隊列上。因為廣播不需要“思考”,所以Fanout交換機(jī)是四種交換機(jī)中速度最快的。

在這里插入圖片描述

同樣地,先在rabbitmq-provider項目上創(chuàng)建FanoutRabbitConfig.java:

@Configuration
public class FanoutRabbitConfig {

    /**
     * 創(chuàng)建三個隊列 :fanout.A   fanout.B  fanout.C
     * 將三個隊列都綁定在交換機(jī) fanoutExchange 上
     * 因為是扇型交換機(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());
    }
}

編寫controller層:

@RestController
@RequestMapping("rabbit")
public class FanoutController {

    @Autowired
    private RabbitTemplate template;

    @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);
        template.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }
}

接著在rabbitmq-consumer項目里加上消息消費類,

FanoutReceiverA:

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverB

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB{
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverC :

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費者收到消息  : " +testMessage.toString());
    }
}

最后將rabbitmq-provider和rabbitmq-consumer項目都跑起來,調(diào)用下接口/sendFanoutMessage,運行結(jié)果如下:

在這里插入圖片描述

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

4、消息確認(rèn)(發(fā)送確認(rèn)與接收確認(rèn)) 4.1發(fā)送確認(rèn)

在rabbitmq-provider項目中
編寫配置文件:

server:
  port: 8090
spring:
  application:
    name: rabbitmq-consumer
  rabbitmq:
    host: 192.168.152.173
    port: 5672
    username: guest
    password: guest
    #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)
    publisher-confirm-type: correlated
    #確認(rèn)消息已發(fā)送到隊列(Queue)
    publisher-returns: true

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

/**
 * 配置相關(guān)的消息確認(rèn)回調(diào)函數(shù)
 */
@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);

        //確認(rèn)回調(diào)
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean condition, String cause) {
                System.out.println("ConfirmCallback:     " + "相關(guān)數(shù)據(jù):" + correlationData);
                System.out.println("ConfirmCallback:     " + "確認(rèn)情況:" + condition);
                System.out.println("ConfirmCallback:     " + "原因:" + cause);
            }
        });

        //設(shè)置返回消息的回調(diào)
        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;
    }
}

ConfirmCallback
ConfirmCallback是一個回調(diào)接口,就是只確認(rèn)是否正確到達(dá) Exchange 中。
ReturnCallback
通過實現(xiàn) ReturnCallback 接口,啟動消息失敗返回,此接口是在交換器路由不到隊列時觸發(fā)回調(diào),該方法可以不使用,因為交換器和隊列是在代碼里綁定的,如果存在綁定隊列失敗,那就是你代碼寫錯了。

從總體的情況分析,推送消息存在四種情況:

1、消息推送到server,但是在server里找不到交換機(jī)
2、消息推送到server,找到交換機(jī)了,但是沒找到隊列
3、消息推送到sever,交換機(jī)和隊列啥都沒找到
4、消息推送成功

1、消息推送到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);
        //non-existent-exchange這個交換機(jī)沒有創(chuàng)建和配置
        template.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

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

在這里插入圖片描述

因為這個交換機(jī)沒有創(chuàng)建,消息不能到達(dá)到達(dá) Exchange 中。所以會觸發(fā)ConfirmCallback這個回調(diào)函數(shù)

結(jié)論: 只會觸發(fā)ConfirmCallback。

2、消息推送到server,找到交換機(jī)了,但是沒找到隊列
新增一個交換機(jī),但是不給這個交換機(jī)綁定隊列,在DirectRabitConfig里面新增一個直連交換機(jī),不做任何綁定配置操作:

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

同樣編寫測試接口

 @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)用方法,查看運行結(jié)果:

在這里插入圖片描述

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

3、消息推送到sever,交換機(jī)和隊列啥都沒找到
結(jié)論: 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。

4、消息推送成功
結(jié)論: 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。

總結(jié): ConfirmCallback回調(diào)函數(shù)總是會被觸發(fā),確認(rèn)情況true和false是有沒有找到交換機(jī),false的話打印原因。ReturnCallback是交換機(jī)路由不到隊列是回調(diào),也就是你交換機(jī)沒有綁定隊列,這個回調(diào)函數(shù)可以不寫,綁定失敗的話就好好看看代碼哪里寫錯了吧。

以上是生產(chǎn)者推送消息的消息確認(rèn) 回調(diào)函數(shù)的使用介紹!

4.2 接受確認(rèn)

消費者確認(rèn)發(fā)生在監(jiān)聽隊列的消費者處理業(yè)務(wù)失敗,如,發(fā)生了異常,不符合要求的數(shù)據(jù)……,這些場景我們就需要手動處理,比如重新發(fā)送或者丟棄。

消息確認(rèn)模式有:

  • AcknowledgeMode.NONE:自動確認(rèn)
  • AcknowledgeMode.AUTO:根據(jù)情況確認(rèn)
  • AcknowledgeMode.MANUAL:手動確認(rèn)

參考:https://blog.csdn.net/qq_35387940/article/details/100514134

4.1.1 自動確認(rèn)

修改消費者配置文件

rabbitmq:
  host: 192.168.152.173
  port: 5672
  username: guest
  password: guest
  listener:
    direct:
      acknowledge-mode: auto   # manual:手動

在消費者項目里,
新建MessageListenerConfig.java上添加代碼相關(guān)的配置代碼:

package com.zsn.demo.config;

import com.zsn.demo.resceiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
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: zhouzhou
 * @date:2021/7/22 9:49
 */
@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è)置一個隊列
        container.setQueueNames("TestDirectQueue");
        //如果同時設(shè)置多個如下: 前提是隊列都是必須已經(jīng)創(chuàng)建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");


        //另一種設(shè)置隊列的方法,如果使用這種情況,那么要設(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)模式需要實現(xiàn) ChannelAwareMessageListener):

package com.zsn.demo.resceiver;

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;

/**
 * @author: zhouzhou
 * @date:2021/7/22 9:50
 */
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
    /**
     * 用于處理接收到的 Rabbit 消息的回調(diào)。
     * 實現(xiàn)者應(yīng)該處理指定的信息
     * 通常通過給定的會話發(fā)送回復(fù)消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進(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("消費的主題消息來自:"+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個參數(shù),手動確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時,則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個參數(shù),true會重新放回隊列,所以需要自己根據(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;
    }
}

需要注意的 basicAck 方法需要傳遞兩個參數(shù)

  • deliveryTag(唯一標(biāo)識 ID):當(dāng)一個消費者向 RabbitMQ 注冊后,會建立起一個 Channel ,RabbitMQ 會用 basic.deliver 方法向消費者推送消息,這個方法攜帶了一個 delivery tag, 它代表了 RabbitMQ 向該 Channel 投遞的這條消息的唯一標(biāo)識 ID,是一個單調(diào)遞增的正整數(shù),delivery tag 的范圍僅限于 Channel
  • multiple:為了減少網(wǎng)絡(luò)流量,手動確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時,則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息

basicNack方法需要傳遞三個參數(shù)

  • deliveryTag(唯一標(biāo)識 ID):上面已經(jīng)解釋了。
  • multiple:上面已經(jīng)解釋了。
  • requeue: true :重回隊列,false :丟棄,我們在nack方法中必須設(shè)置 false,否則重發(fā)沒有意義。

basicReject方法需要傳遞兩個參數(shù)

  • deliveryTag(唯一標(biāo)識 ID):上面已經(jīng)解釋了。
  • requeue:上面已經(jīng)解釋了,在reject方法里必須設(shè)置true。

總結(jié)

  1. basic.ack用于肯定確認(rèn)
  2. basic.nack用于否定確認(rèn)(注意:這是AMQP 0-9-1的RabbitMQ擴(kuò)展)
  3. basic.reject用于否定確認(rèn),但與basic.nack相比有一個限制:一次只能拒絕單條消息

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

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

在這里插入圖片描述

擴(kuò)展1
除了直連交換機(jī)的隊列TestDirectQueue需要變成手動確認(rèn)以外,我們還需要將一個其他的隊列或者多個隊列也變成手動確認(rèn),而且不同隊列實現(xiàn)不同的業(yè)務(wù)處理。
那么我們需要做的第一步,往SimpleMessageListenerContainer里添加多個隊列

在這里插入圖片描述
MyAckReceiver.java 就可以同時將上面設(shè)置到的隊列的消息都消費下來。

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
   /**
     * 用于處理接收到的 Rabbit 消息的回調(diào)。
     * 實現(xiàn)者應(yīng)該處理指定的信息
     * 通常通過給定的會話發(fā)送回復(fù)消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進(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("消費的消息來自的隊列名為:"+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務(wù)處理流程......");
                
            }
 
            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費的消息來自的隊列名為:"+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行fanout.A中的消息的業(yè)務(wù)處理流程......");
 
            }
            
            channel.basicAck(deliveryTag, true);
//			channel.basicReject(deliveryTag, true);//為true會重新放回隊列
        } 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;
    }
}

分別調(diào)用這兩個隊列相關(guān)接口,運行結(jié)果如下:

在這里插入圖片描述

擴(kuò)展2: 手動確認(rèn)的另一種寫法:
我們在實際工作中可能不會這么寫,一般會寫在server層,自己定義一個接口:
RabbitConsumerService

public interface RabbitConsumerService {
    void process(Channel channel, Message message) throws IOException;
}
@Component
public class RabbitConsumerServiceImpl implements RabbitConsumerService{
    @RabbitHandler
    @RabbitListener(queues = "TestDirectQueue")
    @Override
    public void process(Channel channel, Message message) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點進(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("消費的主題消息來自:"+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個參數(shù),手動確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時,則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個參數(shù),true會重新放回隊列,所以需要自己根據(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;
    }
}

@Component這個理論上要換成@service,但是由于我沒有添加日志,所以換成@service會報錯,這里就是提供一個思路,運行結(jié)果如下:

在這里插入圖片描述

附:
rabbitmq在application.yml文件中的其他配置

rabbitmq:
    addresses: 192.168.152.193:5672
    username: guest
    password: guest  
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    virtual-host: /
    listener:
      simple:
        prefetch: 1               #設(shè)置一次處理一個消息
        acknowledge-mode: manual  #設(shè)置消費端手動 ack
        concurrency: 3            #設(shè)置同時有3個消費者消費
    connection-timeout: 500
    # 確認(rèn)消息是否正確到達(dá)queue,如果沒有則觸發(fā),如果有則不觸發(fā)
    publisher-returns: true
    #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)
    publisher-confirm-type: correlated
    template:
      #設(shè)置為 true后 消費者在消息沒有被路由到合適隊列情況下會被return監(jiān)聽,而不會自動刪除
      mandatory: true

解釋:Mandatory 設(shè)置為 true 則會監(jiān)聽器會接受到路由不可達(dá)的消息,然后處理。如果設(shè)置為 false,broker 將會自動刪除該消息。

參考https://blog.csdn.net/qq_35387940/article/details/100514134

到此這篇關(guān)于RabbitMQ交換機(jī)與Springboot整合的簡單實現(xiàn)的文章就介紹到這了,更多相關(guān)RabbitMQ Springboot整合內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • spring boot實現(xiàn)profiles動態(tài)切換的示例

    spring boot實現(xiàn)profiles動態(tài)切換的示例

    Spring Boot支持在不同的環(huán)境下使用不同的配置文件,該技術(shù)非常有利于持續(xù)集成,在構(gòu)建項目的時候只需要使用不同的構(gòu)建命令就可以生成不同運行環(huán)境下war包,而不需要手動切換配置文件。
    2020-10-10
  • java用split分割字符串的一個有趣現(xiàn)象

    java用split分割字符串的一個有趣現(xiàn)象

    最近在項目中使用了java中的split分割字符串,發(fā)現(xiàn)了一個bug,充分了展示了自己對java底層的認(rèn)知有很多的不足和欠缺。下面將這次的經(jīng)過總結(jié)出來分享給大家,有需要的朋友們可以參考借鑒,下面來一起看看吧。
    2016-12-12
  • 關(guān)于@ApiModel和@ApiModelProperty的使用

    關(guān)于@ApiModel和@ApiModelProperty的使用

    這篇文章主要介紹了關(guān)于@ApiModel和@ApiModelProperty的使用方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
    2023-11-11
  • apllo開源分布式配置中心詳解

    apllo開源分布式配置中心詳解

    這篇文章主要為大家介紹了apllo開源分布式配置中心部署詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪
    2023-02-02
  • Java弱引用集合WeakHashMap總結(jié)

    Java弱引用集合WeakHashMap總結(jié)

    這篇文章主要介紹了Java弱引用集合WeakHashMap總結(jié),WeakHashMap利用WeakReference的弱引用特性讓用戶在使用的過程中不會因為沒有釋放Map中的資源而導(dǎo)致內(nèi)存泄露,WeakHashMap實現(xiàn)了Map接口,使用方式和其他的Map相同,需要的朋友可以參考下
    2023-09-09
  • Spring Cloud Gateway去掉url前綴

    Spring Cloud Gateway去掉url前綴

    這篇文章主要介紹了Spring Cloud Gateway去掉url前綴的操作,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2021-07-07
  • SpringCloud項目集成Feign、Hystrix過程解析

    SpringCloud項目集成Feign、Hystrix過程解析

    這篇文章主要介紹了SpringCloud項目集成Feign、Hystrix過程解析,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下
    2019-11-11
  • mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在

    mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在

    這篇文章主要介紹了mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在,具有很好的參考價值,希望對的有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-08-08
  • Java設(shè)計模式探究之觀察者模式詳解

    Java設(shè)計模式探究之觀察者模式詳解

    這篇文章主要為大家詳細(xì)介紹了JAVA的觀察者模式,文中示例代碼介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們可以參考一下,希望能夠給你帶來幫助
    2022-08-08
  • JavaWeb中過濾器Filter的用法詳解

    JavaWeb中過濾器Filter的用法詳解

    過濾器通常對一些web資源進(jìn)行攔截,做完一些處理器再交給下一個過濾器處理,直到所有的過濾器處理器,再調(diào)用servlet實例的service方法進(jìn)行處理。本文將通過示例為大家講解JavaWeb中過濾器Filter的用法與實現(xiàn),需要的可以參考一下
    2022-08-08

最新評論