SpringBoot 整合 RabbitMQ 的使用方式(代碼示例)

更新時間：2024年10月11日 09:39:36 作者：寰夢

本文詳細介紹了使用RabbitTemplate進行消息傳遞的幾種模式,包括點對點通信、發(fā)布/訂閱模式、工作隊列模式、路由模式和主題模式,每種模式都通過代碼示例展示了生產(chǎn)者和消費者的實現(xiàn),幫助開發(fā)者理解和運用RabbitMQ進行高效的消息處理

一、RabbitTemplate 的使用

1.【導(dǎo)入依賴】

<!-- rabbitMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.6.1</version>
</dependency>

2.【添加配置】

rabbitmq:
    host:   #ip地址
    port: 5672 #端口
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        prefetch: 1 # 默認每次取出一條消息消費， 消費完成取下一條
        acknowledge-mode: manual # 設(shè)置消費端手動ack確認
        retry:
          enabled: true # 是否支持重試
    publisher-confirm-type: correlated  #確認消息已發(fā)送到交換機(Exchange)
    publisher-returns: true  #確認消息已發(fā)送到隊列(Queue)

3.【點對點通信（隊列模式）（Point-to-Point Messaging）】

使用方式：

這種方式也被稱為隊列（Queue）模型。消息發(fā)送者（Producer）發(fā)送消息到隊列，然后消息接收者（Consumer）從隊列中獲取消息進行處理。這種模型下，每個消息只有一個消費者可以接收，確保消息的可靠傳遞和順序處理。

代碼示例：生產(chǎn)者

    /**
     * 第一種模型： 簡單模型
     * 一個消息生產(chǎn)者  一個隊列  一個消費者
     * @return
     */
    @GetMapping("hello/world")
    public void helloWorld() {
        SysUser sysUser = new SysUser();
        // 發(fā)送消息
        // 第一個參數(shù)： String routingKey 路由規(guī)則 【交換機 和隊列的綁定規(guī)則 】  隊列名稱
        // 第二個參數(shù)： object message 消息的內(nèi)容
//        rabbitTemplate.convertAndSend("hello_world_queue", "hello world rabbit!");
        ///  MessagePostProcessor 消息包裝器  如果需要對消息進行包裝
        rabbitTemplate.convertAndSend("hello_world_queue", "hello world rabbit!", message -> {
            // 設(shè)置唯一的標(biāo)識
            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
            return message;
        });

消費者

import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
@Component
@Log4j2
public class HelloWorldConsumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    /**
     * 監(jiān)聽 hello_world_queue 隊列消費消息
     * queues 監(jiān)聽隊列的名稱  要求這個隊列必須是已經(jīng)存在的隊列
     * queuesToDeclare 監(jiān)聽隊列 如果這個隊列不存在 則 rabbitMQ 中 RabbitAdmin 會幫助去構(gòu)建這個隊列
     */
    @RabbitListener(queuesToDeclare = @Queue("hello_world_queue"))
    public void helloWorldConsumer(String msg, Message message, Channel channel) {
        // 獲取消息的唯一標(biāo)識
        String messageId = message.getMessageProperties().getMessageId();
        // 將消息添加到 Redis的set集合中  set 不能重復(fù)的  方法的返回值 添加成功的數(shù)量
        Long count = redisTemplate.opsForSet().add("hello_world_queue", messageId);
        if (count != null && count == 1) {
            // 沒有消費過   正常消費
            log.info("hello_world_queue隊列消費者接收到了消息，消息內(nèi)容：{}", message);
        }
    }
}

4.【發(fā)布/訂閱模式（Publish/Subscribe Messaging）】

使用方式：

在發(fā)布/訂閱模式中，消息發(fā)送者將消息發(fā)布到交換機（Exchange），而不是直接發(fā)送到隊列。交換機負責(zé)將消息路由到一個或多個綁定的隊列中。每個訂閱者（Subscriber）可以選擇訂閱它感興趣的消息隊列，從而接收消息。

代碼示例：生產(chǎn)者

/**
 * 工作隊列
 * 一個生產(chǎn)者  一個隊列  多個消費者
 */
@GetMapping("work/queue")
public void workQueue() {
    for (int i = 1; i <= 10; i++) {
        rabbitTemplate.convertAndSend("work_queue", i + "hello work queue!");
    }
}

消費者

import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@Log4j2
public class WorkQueueConsumer {
    /***
     * 消費者1
     * @param message
     */
    @RabbitListener(queuesToDeclare = @Queue("work_queue"))
    public void workQueueConsumer(String message) throws InterruptedException {
        Thread.sleep(200);
        log.info("work_queue隊列消費者1接收到了消息，消息內(nèi)容：{}", message);
    }
    /***
     * 消費者2
     * @param message
     */
    @RabbitListener(queuesToDeclare = @Queue("work_queue"))
    public void workQueueConsumer2(String message) throws InterruptedException {
        Thread.sleep(400);
        log.info("work_queue隊列消費者2接收到了消息，消息內(nèi)容：{}", message);
    }
}

5.【工作隊列模式（Work Queues）】

使用方式：

工作隊列模式也稱為任務(wù)隊列（Task Queues），它可以用來實現(xiàn)任務(wù)的異步處理。多個工作者（Worker）同時監(jiān)聽同一個隊列，當(dāng)有新的任務(wù)消息被發(fā)送到隊列中時，空閑的工作者會獲取并處理這些任務(wù)，確保任務(wù)能夠并行處理而不會重復(fù)執(zhí)行。

代碼示例：生產(chǎn)者

/**
 * 發(fā)布訂閱
 * 一個生產(chǎn)者  多個隊列   多個消費者   涉及 到交換機  fanout
 */
@GetMapping("publish/subscribe")
public void publishSubscribe() {
    // 第一個參數(shù)： 交換機的名稱  沒有要求
    // 第二個參數(shù)： 交換機和隊列的綁定規(guī)則    如果是發(fā)布訂閱模式 那么這個規(guī)則默認不寫 只需要交換機和隊列綁定即可不需要規(guī)則
    // 第三個參數(shù)： 消息內(nèi)容
    rabbitTemplate.convertAndSend("publish_subscribe_exchange", "",
            "hello publisher subscribe!!");
}

消費者

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class PublisherSubscribeConsumer {
    private static final Logger log = LoggerFactory.getLogger(PublisherSubscribeConsumer.class);
    /**
     * 發(fā)布訂閱模型消費者
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("pb_sb_queue_01"),
            exchange = @Exchange(name = "publish_subscribe_exchange",
                    type = ExchangeTypes.FANOUT)))
    public void publisherSubscribe(String message) {
        log.info("發(fā)布訂閱模型消費者1接收到了消息，消息內(nèi)容：{}", message);
    }
    /**
     * 發(fā)布訂閱模型消費者
     *
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("pb_sb_queue_02"),
            exchange = @Exchange(name = "publish_subscribe_exchange", type = ExchangeTypes.FANOUT)))
    public void publisherSubscribe2(String message) {
        log.info("發(fā)布訂閱模型消費者2接收到了消息，消息內(nèi)容：{}", message);
    }
}

6.【路由模式（Routing）】

使用方式：

路由模式允許發(fā)送者根據(jù)消息的路由鍵（Routing Key）將消息路由到特定的隊列。發(fā)送者將消息發(fā)送到交換機，并且通過設(shè)置不同的路由鍵，使消息能夠被交換機路由到不同的隊列。消費者可以根據(jù)需要選擇監(jiān)聽哪些隊列來接收消息。

代碼示例：生產(chǎn)者

/**
 * 路由模型
 * 一個生產(chǎn)者  多個隊列   多個消費者   涉及 到交換機  direct
 */
@GetMapping("routing")
public void routing() {
    // 第一個參數(shù)： 交換機的名稱  沒有要求
    // 第二個參數(shù)： 交換機和隊列的綁定規(guī)則    字符串 隨意
    // 第三個參數(shù)： 消息內(nèi)容
    rabbitTemplate.convertAndSend("routing_exchange", "aaa",
            "hello routing!!");
}

消費者

import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@Log4j2
public class RoutingConsumer {
    /**
     * 路由模型消費者
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("routing_queue_01"),
            exchange = @Exchange(name = "routing_exchange", type = ExchangeTypes.DIRECT),
            key = { "abc", "error", "info" }))
    public void routingConsumer(String message) {
        log.info("路由模型消費者1接收到了消息，消息內(nèi)容：{}", message);
    }
    /**
     * 路由模型消費者
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("routing_queue_02"),
            exchange = @Exchange(name = "routing_exchange", type = ExchangeTypes.DIRECT),
            key = { "aaa", "ccc", "waadaffas" }))
    public void routingConsumer2(String message) {
        log.info("路由模型消費者2接收到了消息，消息內(nèi)容：{}", message);
    }
    /**
     * 路由模型消費者
     * @param message
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue("routing_queue_03"),
            exchange = @Exchange(name = "routing_exchange", type = ExchangeTypes.DIRECT),
            key = { "bbbb", "asdfasd", "asdfasdf" }))
    public void routingConsumer3(String message) {
        log.info("路由模型消費者3接收到了消息，消息內(nèi)容：{}", message);
    }
}

7.【主題模式（Topics）】

使用方式：

主題模式是路由模式的一種擴展，它允許發(fā)送者根據(jù)消息的多個屬性（如主題）將消息路由到一個或多個隊列。主題交換機（Topic Exchange）使用通配符匹配路由鍵與隊列綁定鍵的模式，從而實現(xiàn)更靈活的消息路由和過濾。

代碼示例：生產(chǎn)者

/**
 * 主題模型
 * 一個生產(chǎn)者  多個隊列   多個消費者   涉及 到交換機  topic
 */
@GetMapping("topic")
public void topic() {
    // 第一個參數(shù)： 交換機的名稱  沒有要求
    // 第二個參數(shù)： 交換機和隊列的綁定規(guī)則    多個單詞  以 “.” 拼起來
    // 第三個參數(shù)： 消息內(nèi)容
    rabbitTemplate.convertAndSend("topic_exchange", "bwie.age.name",
            "hello topic!!");
}

消費者

import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@Log4j2
public class TopicConsumer {
    /**
     * *  表示任意一個單詞
     * #  表示任意一個單詞 或 多個
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic_queue_01"),
            exchange = @Exchange(name = "topic_exchange", type = ExchangeTypes.TOPIC),
            key = { "abc.*", "error.*.info", "#.name" }))
    public void topicConsumer(String message) {
        log.info("xxxxxxxxx1");
    }
    /**
     * *  表示任意一個單詞
     * #  表示任意一個單詞 或 多個
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic_queue_02"),
            exchange = @Exchange(name = "topic_exchange", type = ExchangeTypes.TOPIC),
            key = { "abc.*", "username" }))
    public void topicConsumer2(String message) {
        log.info("xxxxxxxxx2");
    }
    /**
     * *  表示任意一個單詞
     * #  表示任意一個單詞 或 多個
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic_queue_03"),
            exchange = @Exchange(name = "topic_exchange", type = ExchangeTypes.TOPIC),
            key = { "bwie.*", "error.*.info" }))
    public void topicConsumer3(String message) {
        log.info("xxxxxxxxx3");
    }
}

到此這篇關(guān)于SpringBoot 整合 RabbitMQ 的使用的文章就介紹到這了,更多相關(guān)SpringBoot 整合 RabbitMQ內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: