我的编程空间，编程开发者的网络收藏夹
学习永远不晚
位置：首页-资讯-后端开发

rabbitmq五种模式详解(含实现代码)

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒
省份

北京

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团
手机号
获取短信验证
立即预约

请填写图片验证码后获取短信验证码

看不清楚，换张图片

免费获取短信验证码

rabbitmq五种模式详解(含实现代码)

一、五种模式详解

1.简单模式(Queue模式)

当生产端发送消息到交换机,交换机根据消息属性发送到队列,消费者监听绑定队列实现消息的接收和消费逻辑编写.简单模式下,强调的一个队列queue只被一个消费者监听消费.

1.1 结构

生产者:生成消息,发送到交换机交换机:根据消息属性,将消息发送给队列消费者:监听这个队列,发现消息后,获取消息执行消费逻辑

1.2应用场景

常见的应用场景就是一发,一接的结构
例如:

手机短信邮件单发

2.争抢模式(Work模式)

强调的也是后端队列与消费者绑定的结构

2.1结构

生产者:发送消息到交换机交换机:根据消息属性将消息发送给队列消费者:多个消费者,同时绑定监听一个队列,之间形成了争抢消息的效果

2.2应用场景

  1. 抢红包
  2. 资源分配系统

3.路由模式(Route模式 Direct定向)

从路由模式开始，关心的就是消息如何到达的队列，路由模式需要使用的交换机类型就是路由交换机（direct）

3.1 结构

  • 生产端：发送消息，在消息中处理消息内容，携带一个routingkey
  • 交换机：接收消息，根据消息的routingkey去计算匹配后端队列的routingkey
  • 队列：存储交换机发送的消息
  • 消费端：简单模式 工作争抢

3.2应用场景

  • 短信
  • 聊天工具
  • 邮箱。。

手机号/邮箱地址，都可以是路由key

4.发布订阅模式(Pulish/Subscribe模式 Fanout广播)

不计算路由的一种特殊交换机

4.1结构

4.2应用场景

  • 消息推送
  • 广告

5.主题模式(Topics模式 Tpoic通配符)

路由key值是一种多级路径。中国.四川.成都.武侯区

5.1结构

生产端：携带路由key，发送消息到交换机

队列：绑定交换机和路由不一样，不是一个具体的路由key，而可以使用*和#代替一个范围
| * | 字符串，只能表示一级 |
| --- | --- |
| # | 多级字符串 |

交换机：根据匹配规则，将路由key对应发送到队列

消息路由key:

  • 北京市.朝阳区.酒仙桥
  • 北京市.#: 匹配true
  • 上海市.浦东区.*: 没匹配false
  • 新疆.乌鲁木齐.#

5.2 应用场景

做物流分拣的多级传递.

6.完整结构

二、代码实现

1.创建SpringBoot工程

1.1 工程基本信息

1.2 依赖信息

1.3 配置文件applicasion.properties


# 应用名称
spring.application.name=springboot-demo
# Actuator Web 访问端口
management.server.port=8801
management.endpoints.jmx.exposure.include=*
management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always
# 应用服务 WEB 访问端口
server.port=8801

######################### RabbitMQ配置 ########################
# RabbitMQ主机
spring.rabbitmq.host=127.0.0.1
# RabbitMQ虚拟主机
spring.rabbitmq.virtual-host=demo
# RabbitMQ服务端口
spring.rabbitmq.port=5672
# RabbitMQ服务用户名
spring.rabbitmq.username=admin
# RabbitMQ服务密码
spring.rabbitmq.password=admin
# RabbitMQ服务发布确认属性配置
## NONE值是禁用发布确认模式，是默认值
## CORRELATED值是发布消息成功到交换器后会触发回调方法
## SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;
spring.rabbitmq.publisher-confirm-type=simple
# RabbitMQ服务开启消息发送确认
spring.rabbitmq.publisher-returns=true
######################### simple模式配置 ########################
# RabbitMQ服务 消息接收确认模式
## NONE：不确认
## AUTO：自动确认
## MANUAL：手动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 指定最小的消费者数量
spring.rabbitmq.listener.simple.concurrency=1
# 指定最大的消费者数量
spring.rabbitmq.listener.simple.max-concurrency=1
# 开启支持重试
spring.rabbitmq.listener.simple.retry.enabled=true

2.简单模式

2.1 创建SimpleQueueConfig 简单队列配置类


package com.gmtgo.demo.simple;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class SimpleQueueConfig {
    
    private final String simpleQueue = "queue_simple";
    @Bean
    public Queue simpleQueue() {
        return new Queue(simpleQueue);
    }
}

2.2 编写生产者


package com.gmtgo.demo.simple;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class SimpleProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage() {
        for (int i = 0; i < 5; i++) {
            String message = "简单消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend( "queue_simple", message);
        }
    }
}

2.3 编写消费者


package com.gmtgo.demo.simple;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class SimpleConsumers {

    @RabbitListener(queues = "queue_simple")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息:{}", new String(message.getBody()));
    }
}

2.4 编写访问类


package com.gmtgo.demo.simple;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping(value = "/rabbitMq")
public class SimpleRabbitMqController {

    @Autowired
    private SimpleProducer simpleProducer;

    @RequestMapping(value = "/simpleQueueTest")
    public String simpleQueueTest() {
        simpleProducer.sendMessage();
        return "success";
    }
}

2.5 测试启动项目访问 simpleQueueTest

访问地址:http://127.0.0.1:8801/rabbitMq/simpleQueueTest

结果：

3.Work队列

3.1 编写工作配置


package com.gmtgo.demo.work;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class WorkQueueConfig {

    
    private final String work = "work_queue";

    @Bean
    public Queue workQueue() {
        return new Queue(work);
    }
}

3.2 编写生产者


package com.gmtgo.demo.work;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class WorkProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage() {
        for (int i = 0; i < 10; i++) {
            String message = "工作消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("work_queue", message);
        }
    }
}

3.3 编写消费者1


package com.gmtgo.demo.work;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class WorkConsumers1 {

    @RabbitListener(queues = "work_queue")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息1:{}", new String(message.getBody()));
    }
}

3.4 编写消费者2


package com.gmtgo.demo.work;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class WorkConsumers2 {

    @RabbitListener(queues = "work_queue")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息2:{}", new String(message.getBody()));
    }
}

3.5 编写测试方法


package com.gmtgo.demo.work;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping(value = "rabbitMq")
public class WorkRabbitMqController {
    @Autowired
    private WorkProducer workProducer;

    @RequestMapping(value = "workQueueTest")
    public String workQueueTest() {
        workProducer.sendMessage();
        return "success";
    }
}

3.6 测试启动项目访问 workQueueTest

访问地址http://127.0.0.1:8801/rabbitMq/workQueueTest

结果：

控制台打印，发现10条消息 偶数条消费者1获取，奇数条消费者2获取，并且平均分配。
当然通过代码实现按需分配，即谁的性能强，谁优先原则，实现负载均衡。
配置可控分配数

4. 发布订阅模式(Publish/Subscibe模式)

订阅模式–多个消费者监听不同的队列，但队列都绑定同一个交换机

4.1 编写订阅配置类


package com.gmtgo.demo.fanout;

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 FanoutQueueConfig {

    
    private final String fanout1 = "fanout_queue_1";

    private final String fanout2 = "fanout_queue_2";

    
    private final String fanoutExchange = "fanoutExchange";

    
    @Bean
    public Queue fanoutQueue1() {
        return new Queue(fanout1);
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue(fanout2);
    }

    
    @Bean
    public FanoutExchange exchange() {
        return new FanoutExchange(fanoutExchange);
    }

    
    @Bean
    public Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange exchange) {
        return BindingBuilder.bind(fanoutQueue1).to(exchange);
    }

    @Bean
    public Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange exchange) {
        return BindingBuilder.bind(fanoutQueue2).to(exchange);
    }
}

4.2 编写订阅生产者


package com.gmtgo.demo.fanout;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class FanoutProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage() {
        for (int i = 0; i < 5; i++) {
            String message = "订阅模式消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("fanoutExchange", "", message);
        }
    }
}

4.3 编写订阅消费者1


package com.gmtgo.demo.fanout;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;



@Slf4j
@Component
public class FanoutConsumers1 {

    @RabbitListener(queues = "fanout_queue_1")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息1:{}", new String(message.getBody()));
    }
}

4.4 编写订阅消费者2


package com.gmtgo.demo.fanout;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class FanoutConsumers2 {

    @RabbitListener(queues = "fanout_queue_2")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息2:{}", new String(message.getBody()));
    }
}

4.5 编写测试方法


package com.gmtgo.demo.fanout;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping(value = "rabbitMq")
public class FanoutRabbitMqController {
    @Autowired
    private FanoutProducer fanoutProducer;

    @RequestMapping(value = "fanoutQueueTest")
    public String fanoutQueueTest() {
        fanoutProducer.sendMessage();
        return "success";
    }
}

3.6 测试启动项目访问 fanoutQueueTest

  • 访问地址http://127.0.0.1:8801/rabbitMq/fanoutQueueTest
  • 结果：

控制台打印 ，发现两个绑定了不同队列的消费者都接受到了同一条消息查看RabbitMq 服务器：


5. 路由模式(Route模式 Direct定向)

5.1 编写路由配置类


package com.gmtgo.demo.direct;

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;


@Configuration
public class DirectQueueConfig {

    
    private final String direct1 = "direct_queue_1";

    private final String direct2 = "direct_queue_2";

    
    private final String directExchange = "directExchange";

    
    @Bean
    public Queue directQueue1() {
        return new Queue(direct1);
    }

    @Bean
    public Queue directQueue2() {
        return new Queue(direct2);
    }

    
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(directExchange);
    }

    
    @Bean
    Binding bindingDirectExchange1(Queue directQueue1, DirectExchange exchange) {
        return BindingBuilder.bind(directQueue1).to(exchange).with("update");
    }

    
    @Bean
    Binding bindingDirectExchange2(Queue directQueue2, DirectExchange exchange) {
        return BindingBuilder.bind(directQueue2).to(exchange).with("add");
    }

}

5.2 编写生产者


package com.gmtgo.demo.direct;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class DirectProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessageA() {
        for (int i = 0; i < 5; i++) {
            String message = "路由模式--routingKey=update消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("directExchange", "update", message);
        }
    }

    public void sendMessageB() {
        for (int i = 0; i < 5; i++) {
            String message = "路由模式--routingKey=add消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("directExchange", "add", message);
        }
    }
}

5.3 编写消费者1


package com.gmtgo.demo.direct;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class DirectConsumers1 {

    @RabbitListener(queues = "direct_queue_1")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息1:{}", new String(message.getBody()));
    }
}

5.4 编写消费者2


package com.gmtgo.demo.direct;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


@Slf4j
@Component
public class DirectConsumers2 {

    @RabbitListener(queues = "direct_queue_2")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息2:{}", new String(message.getBody()));
    }
}

5.5 编写访问类


package com.gmtgo.demo.direct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping(value = "rabbitMq")
public class DirectRabbitMqController {
    @Autowired
    private DirectProducer directProducer;

    @RequestMapping(value = "directQueueTest1")
    public String directQueueTest1() {
        directProducer.sendMessageA();
        return "success";
    }

    @RequestMapping(value = "directQueueTest2")
    public String directQueueTest2() {
        directProducer.sendMessageB();
        return "success";
    }
}

5.6 测试启动项目访问directQueueTest1 , directQueueTest2

访问地址http://127.0.0.1:8801/rabbitMq/directQueueTest1

访问地址http://127.0.0.1:8801/rabbitMq/directQueueTest2

结果：directQueueTest1：

directQueueTest2：

6. 主题模式(Topics模式 Tpoic通配符)

6.1 编写路由配置类


package com.gmtgo.demo.topic;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class TopicQueueConfig {
    
    private final String topic1 = "topic_queue_1";

    private final String topic2 = "topic_queue_2";

    
    private final String topicExchange = "topicExchange";

    
    @Bean
    public Queue topicQueue1() {
        return new Queue(topic1);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(topic2);
    }

    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(topicExchange);
    }

    
    @Bean
    Binding bindingTopicExchange1(Queue topicQueue1, TopicExchange exchange) {
        return BindingBuilder.bind(topicQueue1).to(exchange).with("topic.keyA");
    }

    
    @Bean
    Binding bindingTopicExchange2(Queue topicQueue2, TopicExchange exchange) {
        return BindingBuilder.bind(topicQueue2).to(exchange).with("topic.#");
    }
}

6.2 编写生产者


package com.gmtgo.demo.topic;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class TopicProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessageA() {
        for (int i = 0; i < 5; i++) {
            String message = "通配符模式--routingKey=topic.keyA消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("topicExchange", "topic.keyA", message);
        }
    }

    public void sendMessageB() {
        for (int i = 0; i < 5; i++) {
            String message = "通配符模式--routingKey=topic.#消息" + i;
            log.info("我是生产信息:{}", message);
            rabbitTemplate.convertAndSend("topicExchange", "topic.keyD.keyE", message);
        }
    }
}

6.3 编写消费者1


package com.gmtgo.demo.topic;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Slf4j
@Component
public class TopicConsumers1 {

    @RabbitListener(queues = "topic_queue_1")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息1:{}",new String(message.getBody()));
    }
}

6.4 编写消费者2


package com.gmtgo.demo.topic;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Slf4j
@Component
public class TopicConsumers2 {

    @RabbitListener(queues = "topic_queue_2")
    public void readMessage(Message message, Channel channel) throws IOException {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info("我是消费信息2:{}",new String(message.getBody()));
    }
}

6.5 编写访问类


package com.gmtgo.demo.topic;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
@RequestMapping(value = "rabbitMq")
public class TopicRabbitMqController {
    @Autowired
    private TopicProducer topicProducer;

    @RequestMapping(value = "topicQueueTest1")
    public String topicQueueTest1() {
        topicProducer.sendMessageA();
        return "success";
    }

    @RequestMapping(value = "topicQueueTest2")
    public String topicQueueTest2() {
        topicProducer.sendMessageB();
        return "success";
    }
}

6.6 测试启动项目访问topicQueueTest1 , topicQueueTest2

  • 访问地址http://127.0.0.1:8801/rabbitMq/topicQueueTest1
  • 访问地址http://127.0.0.1:8801/rabbitMq/topicQueueTest2
  • 结果：

topicQueueTest1，两个消费者都能消费

topicQueueTest2，只有消费者2 可以消费

至此，五种队列的实现已结束！

7. 实现生产者消息确认

7.1 配置文件


######################### RabbitMQ配置 ########################
# RabbitMQ主机
spring.rabbitmq.host=127.0.0.1
# RabbitMQ虚拟主机
spring.rabbitmq.virtual-host=demo
# RabbitMQ服务端口
spring.rabbitmq.port=5672
# RabbitMQ服务用户名
spring.rabbitmq.username=admin
# RabbitMQ服务密码
spring.rabbitmq.password=admin
# RabbitMQ服务发布确认属性配置
## NONE值是禁用发布确认模式，是默认值
## CORRELATED值是发布消息成功到交换器后会触发回调方法
## SIMPLE值经测试有两种效果，其一效果和CORRELATED值一样会触发回调方法，其二在发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;
spring.rabbitmq.publisher-confirm-type=simple
# 连接超时时间
spring.rabbitmq.connection-timeout=20000
# RabbitMQ服务开启消息发送确认
spring.rabbitmq.publisher-returns=true
######################### simple模式配置 ########################
# RabbitMQ服务 消息接收确认模式
## NONE：不确认
## AUTO：自动确认
## MANUAL：手动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 指定最小的消费者数量
spring.rabbitmq.listener.simple.concurrency=1
# 指定最大的消费者数量
spring.rabbitmq.listener.simple.max-concurrency=1
# 每次只消费一个消息
spring.rabbitmq.listener.simple.prefetch=1
# 开启支持重试
spring.rabbitmq.listener.simple.retry.enabled=true
# 启用强制信息，默认为false
spring.rabbitmq.template.mandatory=true

7.2 编写消息发送确认类 RabbitConfirmCallback


package com.gmtgo.demo.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

@Slf4j
public class RabbitConfirmCallback implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("=======ConfirmCallback=========");
        log.info("correlationData {} " , correlationData);
        log.info("ack = {}" , ack);
        log.info("cause = {}" , cause);
        log.info("=======ConfirmCallback=========");
    }
}

7.3 编写消息发送交换机返回机制RabbitConfirmReturnCallBack


package com.gmtgo.demo.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;


@Slf4j
public class RabbitConfirmReturnCallBack implements RabbitTemplate.ReturnCallback {
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("--------------ReturnCallback----------------");
        log.info("message = " + message);
        log.info("replyCode = {}", replyCode);
        log.info("replyText = {}", replyText);
        log.info("exchange = {}", exchange);
        log.info("routingKey = {}", routingKey);
        log.info("--------------ReturnCallback----------------");
    }
}

7.4 RabbitMQ配置

在我们的rabbit队列配置类里设置RabbitTemplate
举例：


package com.gmtgo.demo.topic;

import com.gmtgo.demo.config.RabbitConfirmCallback;
import com.gmtgo.demo.config.RabbitConfirmReturnCallBack;
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.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;


@Configuration
public class TopicQueueConfig {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initRabbitTemplate() {
        // 设置生产者消息确认
        rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
        rabbitTemplate.setReturnCallback(new RabbitConfirmReturnCallBack());
    }

    
    private final String topic1 = "topic_queue_1";

    private final String topic2 = "topic_queue_2";

    
    private final String topicExchange = "topicExchange";

    
    @Bean
    public Queue topicQueue1() {
        return new Queue(topic1);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(topic2);
    }

    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(topicExchange);
    }

    
    @Bean
    Binding bindingTopicExchange1(Queue topicQueue1, TopicExchange exchange) {
        return BindingBuilder.bind(topicQueue1).to(exchange).with("topic.keyA");
    }

    
    @Bean
    Binding bindingTopicExchange2(Queue topicQueue2, TopicExchange exchange) {
        return BindingBuilder.bind(topicQueue2).to(exchange).with("topic.#");
    }
}

启动项目发送消息，消息被正常消费，confim回调返回ack=true如果我们将exchange修改，发送到一个不存在的exchange中，会怎么样呢？

会发现confirm回调为false，打印出结果为不存在topicExchange1111的交换机

如果我们在消费端处理逻辑时出错会怎么样呢？修改消费端代码我们在消费时让它报错

confirm回调为true，但是在rabbitmq的web界面会发现存在5条没有消费的消息

如果我们把


channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);

中最后一个参数改为false呢，会发现在web管理界面没有未被消费的消息，说明这条消息已经被摒弃。

实际开发中，到底是打回到队列呢还是摒弃，要看自己的需求，但是打回队列应该有次数限制，不然会陷入死循环。
继续测试，将routingKey修改为一个没有的key,

7.5 结论

  • 如果消息没有到exchange,则confirm回调,ack=false
  • 如果消息到达exchange,则confirm回调,ack=true
  • exchange到queue成功,则不回调return
  • exchange到queue失败,则回调return

8. 项目示例代码：

下载地址：springboot-rabbitmq-demo_1619322789961

到此这篇关于rabbitmq五种模式详解(含实现代码)的文章就介绍到这了,更多相关rabbitmq五种模式内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

阅读原文内容投诉

rabbitmq五种模式详解(含实现代码)

下载Word文档到电脑，方便收藏和打印～

下载Word文档

相关文章

猜你喜欢

rabbitmq五种模式详解(含实现代码)

2024-04-02

以PHP代码为实例详解RabbitMQ消息队列中间件的6种模式

这篇文章主要介绍了以PHP代码为实例详解RabbitMQ消息队列中间件的6种模式，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-05-19

Java实现线程安全单例模式的五种方式的示例代码

2024-04-02

RabbitMQ实现延迟队列的两种方式详解

2024-04-02

RabbitMQ延时队列详解与Java代码实现

这篇文章主要介绍了RabbitMQ延时队列详解与Java代码实现,RabbitMQ延时队列是指消息在发送到队列后，并不立即被消费者消费，而是等待一段时间后再被消费者消费。这种队列通常用于实现定时任务,需要的朋友可以参考下
2023-05-18

Spring Boot详解五种实现跨域的方式

2024-04-02

servlet下载文件实现代码详解（五）

本文实例为大家分享了servlet下载文件的具体代码，供大家参考，具体内容如下1.servlet下载文件　　servlet下载文件就是将服务器端的文件传输到客户端。2案例下载文件servlet类package com.learn;impor
2023-05-31

Redis实现分布式锁的五种方法详解

目录1. 单机数据一致性2. 分布式数据一致性3. Redis实现分布式锁3.1 方式一3.2 方式二（改进方式一）3.3 方式三（改进方式二）3.4 方式四（改进方式三）3.5 方式五（改进方式四）3.6 小结在单体应用中，如果我们对共享
2022-06-14

使用Python实现桥接模式的代码详解

Python桥接模式代码详解摘要桥接模式将抽象和实现分离，允许独立修改。在Python中，抽象类定义接口，由具体类实现。接口定义了方法，具体类必须实现。创建抽象类和具体类的实例，并动态改变实现。桥接模式提供了解耦、可扩展性、可读性和可维护性优势。它适用于需要分离抽象和实现，或需要动态更改对象实现的场景。
使用Python实现桥接模式的代码详解
2024-04-02

python 三种方式实现截屏（详解+完整代码）

2023-09-08

Python实现单例模式的四种方式详解

2024-04-02

Java单例模式的6种实现方式详解

2024-04-02

SpringAop实现原理及代理模式详解

2024-04-02

Java实现AOP代理的三种方式详解

2024-04-02

Java动态代理四种实现方式详解

代理模式也是一种非常常见的设计模式。了解Spring框架的都知道，Spring AOP 使用的就是动态代理模式。今天就来系统的重温一遍代
Java动态代理四种实现方式详解
2022-06-07

Transformer模型入门详解及代码实现

2023-10-18

C++单例模式的几种实现方法详解

2024-04-02

C++常用的11种设计模式解释及示例代码详解

c++常用的设计模式包括单例模式、工厂模式、抽象工厂模式、适配器模式、装饰者模式、代理模式、外观模式、桥接模式、组合模式、享元模式、观察者模式和命令模式等，这篇文章主要介绍了C++常用的11种设计模式解释及示例,需要的朋友可以参考下
2023-02-07

PHP中间件模式的两种实现方法详解

PHP中间件模式实现了请求处理的拦截和修改，可分为两种实现方法：基于服务的中间件：使用服务提供者管理，适用于服务容器框架，灵活且可重用。基于路由的中间件：直接在路由中指定，适用于路由框架，简单且性能较高。根据框架和需求，选择适合的实现方法：服务容器框架优先基于服务的中间件。路由框架优先基于路由的中间件。
PHP中间件模式的两种实现方法详解
2024-04-02

Android Spinner与适配器模式详解及实例代码

最近做项目对Android Spinner 使用，这里简单写个小例子，来测试如何使用。 Spinner是一个下拉列表，往安卓界面中拖拽一个Spinner控件，在属性中设置Android:entries=“@array/spinner_dat
2022-06-06

热门标签

Linux(148)PHP(127)Java(102)正则表达式(101)JavaScript(69)最佳实践(67)jQuery(44)MySQL(39)Docker(37)C语言(36)性能优化(34)Python(34)XML(28)string(27)第三方库(23)回调函数(23)ZIP(22)数组(22)可扩展性(22)字符串比较(21)find(20)RPM(20)Go(20)grep(19)ASP.NETCore(19)XML解析器(19)事件(19)事件处理程序(19)StringBuilder(18)Nginx(18)

编程热搜

  • Python 学习之路 - Python
    一、安装Python34Windows在Python官网（https://www.python.org/downloads/）下载安装包并安装。Python的默认安装路径是：C:\Python34配置环境变量：【右键计算机】--》【属性】-
    Python 学习之路 - Python
  • chatgpt的中文全称是什么
    chatgpt的中文全称是生成型预训练变换模型。ChatGPT是什么ChatGPT是美国人工智能研究实验室OpenAI开发的一种全新聊天机器人模型，它能够通过学习和理解人类的语言来进行对话，还能根据聊天的上下文进行互动，并协助人类完成一系列
    chatgpt的中文全称是什么
  • C/C++中extern函数使用详解
    C/C++中extern函数使用详解
  • C/C++可变参数的使用
    可变参数的使用方法远远不止以下几种，不过在C,C++中使用可变参数时要小心，在使用printf()等函数时传入的参数个数一定不能比前面的格式化字符串中的’%’符号个数少，否则会产生访问越界，运气不好的话还会导致程序崩溃
    C/C++可变参数的使用
  • css样式文件该放在哪里
    css样式文件该放在哪里
  • php中数组下标必须是连续的吗
    php中数组下标必须是连续的吗
  • Python 3 教程
    Python 3 教程 Python 的 3.0 版本，常被称为 Python 3000，或简称 Py3k。相对于 Python 的早期版本，这是一个较大的升级。为了不带入过多的累赘，Python 3.0 在设计的时候没有考虑向下兼容。 Python
    Python 3 教程
  • Python pip包管理
    一、前言    在Python中， 安装第三方模块是通过 setuptools 这个工具完成的。 Python有两个封装了 setuptools的包管理工具： easy_install  和  pip ， 目前官方推荐使用 pip。    
    Python pip包管理
  • ubuntu如何重新编译内核
    ubuntu如何重新编译内核
  • 改善Java代码之慎用java动态编译
    改善Java代码之慎用java动态编译
查看更多

编程资源站

  • 资料下载
  • 历年试题
2021年下半年软考高级信息系统项目管理师高频考点精选资料
2021下半年软考高级信息系统技术知识点记忆口诀精选资料
2021下半年软考《信息系统项目管理师》考试真题及答案精选资料
2021下半年软考高级考试备考攻略精选资料
2021年软考高级《信息系统项目管理师》巩固练习题汇总精选资料
2021下半年软考高级信息系统项目管理师30个易考知识点汇总精选资料
2021下半年软考高级知识点这样记，还担心记不住吗精选资料
2021年下半年软考高级考试重点汇总精选资料
2021下半年软考高级信息系统项目管理师计算公式汇总精选资料
2021年下半年软考高级《信息系统项目管理师》模拟试题精选资料
查看更多
信息系统项目管理师选择题每日一练（2024）历年试题
2023年下半年信息系统项目管理师综合知识真题演练历年试题
查看更多

目录

本网页已闲置超过3分钟，按键盘任意键或点击空白处，即可回到网页
最新资讯更多
    友情链接编程网

    反馈

    我要
    反馈