文章目录

RabbitMQ
- 一、RabbitMQ介绍
- - 1.1 现存问题
- 一、RabbitMQ介绍
- 二、RabbitMQ安装
- 三、RabbitMQ架构
- 四、RabbitMQ通信方式
- - 4.1 RabbitMQ提供的通讯方式
  - 4.2 Helloworld 方式
  - 4.2Work queues
  - 4.3 Publish/Subscribe
  - 4.4 Routing
  - 4.5 Topics
  - 4.6 RPC (了解)
- 五、Springboot 操作RabbitMQ
- 六、RabbitMQ保证消息可靠性
- - 6.1、保证消息一定送达到Exchange
  - 6.2、保证消息可以路由到Queue中
  - 6.3、保证队列持久化消息
  - 6.4、保证消息者可以正常消费消息
  - 6.5 SpringBoot实现上述操作
  - - 6.5.1 Confirm
  - 6.5.2 Return
  - 6.5.3 消息持久化
- 七、RabbitMQ死信队列 & 延迟交换机
- - 7.4、准备Exchange&Queue
  - 7.5、实现效果
- 八、RabbitMQ的集群

RabbitMQ

一、RabbitMQ介绍

1.1 现存问题

- 服务异步调用：服务A如何保证异步请求一定能被服务B接收到并处理
- 削峰: 海量请求，如何实现削峰的效果，将请求全部放到一个队列中，慢慢的消费，这个队列怎么实现？

在这里插入图片描述

- 服务解耦：如何尽量的降低服务之间的耦合问题，如果在订单与积分和商家服务解构，需要一个队列，而这个队列依然需要实现上述两个情况功能。

在这里插入图片描述

一、RabbitMQ介绍

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

AMQP协议：
在这里插入图片描述

Erlang：

Erlang在1991年由爱立信公司向用户推出了第一个版本，经过不断的改进完善和发展，在1996年爱立信又为所有的Erlang用户提供了一个非常实用且稳定的OTP软件库并在1998年发布了第一个开源版本。Erlang同时支持的操作系统有linux,windows,unix等，可以说适用于主流的操作系统上，尤其是它支持多核的特性非常适合多核CPU，而分布式特性也可以很好融合各种分布式集群。

二、RabbitMQ安装

docker-compose.yml

version: “3.1”
services:
rabbitmq:
image: daocloud.io/library/rabbitmq:3.8.5
container_name: rabbitmq
restart: always
volumes:
- ./data/:/var/lib/rabbitmq/
ports:
- 5672:5672
- 15672:15672

在这里插入图片描述

docker-compose.yml文件内容：
在这里插入图片描述

镜像拉取完成后，直接在linux 内部执行： curl localhost:5672

在这里插入图片描述
执行后能够显示AMQP 字样的内容就说明执行成功了

执行 docker exec -it rabbitmq bash 命令进入容器内部
cd opt/rabbitmq/ 目录下

执行cd plugins/sbin 命令进入目录下，执行 ./rabbitmq-plugins enable rabbitmq_managent命令启动rabbitmq 图形化界面

访问15672 端口，默认的账号密码是guest/guest

三、RabbitMQ架构

在这里插入图片描述

四、RabbitMQ通信方式

4.1 RabbitMQ提供的通讯方式

Hello World :为了入门操作
Work queues ：一个队列被多个消费者消费
Publish/Subscribe：手动创建Exchange（FANOUT）
Routing：手动创建Exchange（DIRECT）
Topics ：手动创建Exchange（TOPIC）
RPC： RPC方式
Publisher Confirms

4.2 Helloworld 方式

在这里插入图片描述

//工具类
public class RabbitMQConnectionUtil {
    public static final String RABBITMQ_HOST ="172.16.177.133";
    public static final int RABBITMQ_POST =5672;
    public static final String RABBITMQ_USERNAME ="guest";
    public static final String RABBITMQ_PASSWORD ="guest";
    public static final String RABBITMQ_VIRTUAL_HOST ="/";

    /**
     * 构建RabbitMQ的连接对象
     * @return
     */
    public static Connection getConnection() throws Exception{
        //1.创建connection    工厂对象
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(RABBITMQ_HOST);
        connectionFactory.setPort(RABBITMQ_POST);
        connectionFactory.setUsername(RABBITMQ_USERNAME);
        connectionFactory.setPassword(RABBITMQ_PASSWORD);
        connectionFactory.setVirtualHost(RABBITMQ_VIRTUAL_HOST);
        //2.设置RabbitMQ的连接信息
        Connection connection = connectionFactory.newConnection();
        //3. 返回连接对象

        return connection;
    }
//生产者
 @Test
    public void consume() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(Publisher.QUEUE_NAME,false,false,false,null);
        //4. 监听消息
        DefaultConsumer callback =new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消息者获取消息"+new String(body,"UTF-8"));
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,true,callback);

        System.out.println("开始监听");
        System.in.read();

    }

//消费者
    @Test
    public void publish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4. 发布消息
        String message="hello word!";
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送成功");
        System.in.read();
    }

4.2Work queues

在这里插入图片描述
一个队列中的消息，只会被一个消费者成功的消费，默认情况下，RabbitMQ的队列会将消息以轮询的方式交给不同的消费者消费，消费者拿到消息后，需要给RabbitMQ一个ack，RabbitMQ认为消费者已经拿到消息了

//消费者
 @Test
    public void consume() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(Publisher.QUEUE_NAME,false,false,false,null);
        //3.1 设置消息的控制,一次拿几个消息
//#2        channel.basicQos(1);
        //4. 监听消息
        DefaultConsumer callback =new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                	//模拟业务执行时间
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消息者获取消息"+new String(body,"UTF-8"));
//#1				channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,true,callback);
//#1		channel.basicConsume(Publisher.QUEUE_NAME,false,callback);
        System.out.println("开始监听");
        System.in.read();
 //消费者2
 @Test
    public void consume2() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(Publisher.QUEUE_NAME,false,false,false,null);
        //3.1 设置消息的控制,一次拿几个消息
//#2        channel.basicQos(1);
        //4. 监听消息 
        DefaultConsumer callback =new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                	//模拟业务执行时间
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消息者获取消息"+new String(body,"UTF-8"));
                //basicAck(标识,是否批量操作)
//#1				channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,true,callback);
//#1		channel.basicConsume(Publisher.QUEUE_NAME,false,callback);

        System.out.println("开始监听");
        System.in.read();

    }
//生产者
 public static final String QUEUE_NAME="work";
    @Test
    public void publish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //4. 发布消息
        for (int i=0;i<10;i++){
            String message="hello word!"+i;
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
         }
        System.out.println("消息发送成功");
        System.in.read();
    }

当两台消费者的消费能力不相同的时候，为了提高效率，就不能以轮询的方式进行分发，而是以消费者消费完成后手动传递ack 的方式进行下一个消息的分发，将== #1 #2 的代码 ==打开即可
操作步骤：

操作#1 让消费者关闭自动ack，并且设置消息的流控，最终实现消费者可以尽可能去多消费消息
操作#2 设置每次拿几个消息

4.3 Publish/Subscribe

自行创建路由器，并绑定队列
在这里插入图片描述
如何构建一个自定义交换机，并指定类型是FANOUT，让交换机和多个Queue绑定到一起

//生产者 
public static final String EXCHANGE_NAME="pubsub";
    public static final String QUEUE_NAME1="pubsub-one";
    public static final String QUEUE_NAME2="pubsub-two";
    @Test
    public void pulish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2. 构建Channal
        Channel channel = connection.createChannel();
        //3.构建虚拟机, exchange 交换机名称，type 交换机类型【枚举】
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //4. 构建队列
        channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(QUEUE_NAME2, false, false, false, null);
        //5. 绑定 交换机 和 队列,使用的是FANOUT类型的交换机，绑定方式是直接绑定 ,routingkey 参数随便写什么都可以，
        channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"");
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"");
        //6.发消息到交换机
        channel.basicPublish(EXCHANGE_NAME,"",null,"publish/subscribe!".getBytes());
        System.out.println("消息成功发送");
        
    }

4.4 Routing

DIRECT 类型的交换机，在绑定Exchange和Queue时，需要指定好routingKey同时在发送消息的时候，也指定routingkey，只有在routingkey 一致时，才会把指定的消息路由到指定的队列
在这里插入图片描述

public static final String EXCHANGE_NAME="routing";
    public static final String QUEUE_NAME1="routing-one";
    public static final String QUEUE_NAME2="routing-two";
    @Test
    public void pulish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2. 构建Channal
        Channel channel = connection.createChannel();
        //3.构建虚拟机, exchange 交换机名称，type 交换机类型【枚举】 * 交换机类型 改成 BuiltinExchangeType.DIRECT
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //4. 构建队列
        channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(QUEUE_NAME2, false, false, false, null);
        //5. 绑定 交换机 和 队列,使用的是FANOUT类型的交换机，绑定方式是直接绑定 ,routingkey 参数随便写什么都可以，
        channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"ORANGE");
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"BLACK");
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"GREEN");
        //6.发消息到交换机
        channel.basicPublish(EXCHANGE_NAME,"ORANGE",null,"大橙子!".getBytes());
        channel.basicPublish(EXCHANGE_NAME,"BLACK",null,"黑布林!".getBytes());
        channel.basicPublish(EXCHANGE_NAME,"WHITE",null,"小白兔!".getBytes());


        System.out.println("消息成功发送");

    }

4.5 Topics

topics 模式支持模糊匹配RoutingKey，就像是sql中的 like子句模糊查询，而路由模式等同于sql中的where子句等值查询

在这里插入图片描述

通过模糊路由到队列。该方式的Routing key必须具有固定格式：以 . 间隔的一串单词，比如：quick.orange.rabbit，Routing key 最多不能超过255byte。

交换机和队列的Binding key用通配符来表示，有两种语法：

* 可以替代一个单词；
# 可以替代 0 或多个单词；

例如 #.com.#
#可以表示0级或多级。xx.com、com.xx、com、xx.com.xx.xx、xx.xx.com.xx都可以

例如 *.com. *
*表示一级，xx.com.xx 可以，com.xx 不可以，前面缺少一级，xx.com.xx.xx不可以，com后面只能有一级xx，最终格式必须是 xx.com.xx

 public static final String EXCHANGE_NAME="topics";
    public static final String QUEUE_NAME1="topics-one";
    public static final String QUEUE_NAME2="topics-two";
    @Test
    public void pulish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2. 构建Channal
        Channel channel = connection.createChannel();
        //3.构建虚拟机, exchange 交换机名称，type 交换机类型【枚举】 * 交换机类型 改成 BuiltinExchangeType.TOPIC
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        //4. 构建队列
        channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
        channel.queueDeclare(QUEUE_NAME2, false, false, false, null);
        //5. 绑定 交换机 和 队列,
        // Topic类型的交换机，在和队列绑定时，需要以aaa.bbb.ccc 方式编写routingKey
        // 其中有两个特殊字符： *(相当于占位符)，# (相当通配符)
        channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,"*.orange.*");
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"*.*.rabbit");
        channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,"lazy.#");
        //6.发消息到交换机
        channel.basicPublish(EXCHANGE_NAME,"big.orange.rabbit",null,"大橙子兔子!".getBytes());//匹配1、2
        channel.basicPublish(EXCHANGE_NAME,"small.while.rabbit",null,"小兔子!".getBytes());//匹配1、2
        channel.basicPublish(EXCHANGE_NAME,"lazy.dog.dog.dog.dog",null,"懒狗狗狗狗!".getBytes());//匹配 3


        System.out.println("消息成功发送");

    }

4.6 RPC (了解)

因为两个服务在交互时，可以尽量做到Client和server的结偶，通过RabbitMQ进行结藕操作
需要让client 发送消息时，携带两个属性，

replyto告知server将相应信息放到哪个队列
correlationId告知server 发送相应消息时，需要携带位置标识来告知client响应的消息

在这里插入图片描述

public class Publisher {
    public static final String QUEUE_PUBLISHER = "rpc_publisher";
    public static final String QUEUE_CONSUMER = "rpc_consumer";

    @Test
    public void publish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(QUEUE_PUBLISHER,false,false,false,null);
        channel.queueDeclare(QUEUE_CONSUMER,false,false,false,null);
        //4. 发布消息
        String message="hello rpc!";
        String uuid = UUID.randomUUID().toString();
        AMQP.BasicProperties prop =new AMQP.BasicProperties()
            .builder().replyTo(QUEUE_CONSUMER).correlationId(uuid).build();
        channel.basicPublish("",QUEUE_PUBLISHER,prop,message.getBytes());

        channel.basicConsume(QUEUE_CONSUMER,false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String id = properties.getCorrelationId();
                if(id!=null && id.equals(uuid)){
                    System.out.println("接收到服务端的响应："+new String(body));
                }
            channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });

        System.out.println("消息发送成功");
        System.in.read();
    }
}

public class Consumer {
    public static final String QUEUE_PUBLISHER = "rpc_publisher";
    public static final String QUEUE_CONSUMER = "rpc_consumer";
    @Test
    public void consume() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(QUEUE_PUBLISHER,false,false,false,null);
        channel.queueDeclare(QUEUE_CONSUMER,false,false,false,null);
        //4. 监听消息
        DefaultConsumer callback =new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消息者获取消息"+new String(body,"UTF-8"));
                String resp = "获取到client发出的请求，这里是响应的信息";
                String respQueueName = properties.getReplyTo();
                String uuid = properties.getCorrelationId();
                AMQP.BasicProperties prop =new AMQP.BasicProperties()
                .builder().correlationId(uuid).build();
                channel.basicPublish("",respQueueName,prop,resp.getBytes());
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(QUEUE_PUBLISHER,false,callback);

        System.out.println("开始监听");
        System.in.read();

    }
}

五、Springboot 操作RabbitMQ

创建项目
导入依赖

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

配置rabbitmq信息

spring:
  rabbitmq:
    host: 172.16.177.133
    password: guest
    username: guest
    port: 5672
    virtual-host: /

配置类声明队列

@Configuration
public class RabbitMQConfig {
    public static final String EXCHANGE="boot-exchange";
    public static final String QUEUE="boot-queue";
    public static final String ROUTING_KEY="*.black.*";

    @Bean
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE).build();
    }

    @Bean
    public Queue bootQueue(){
        return QueueBuilder.durable(QUEUE).build();
    }

    @Bean
    public Binding bootBinding(Exchange bootExchange,Queue bootQueue){
        return BindingBuilder.bind(bootQueue).to(bootExchange).with(ROUTING_KEY).noargs();
    }
}

生产者配置

@SpringBootTest
class SpringbootRabbitmqApplicationTests {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
    }

    @Test
    public void publisher(){
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,"big.black.dog","message");
        System.out.println("消息发送成功");
    }

    @Test
    public void publiWithProps(){
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, "big.black.dog", "message", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setCorrelationId("123");
                return message;
            }
        });
        System.out.println("消息发送成功2");
    }
}

消费者配置

@Component
public class ConsumeListener {

    /**
     *
     * @param msg
     * @param channel 前提是配置好spring.rabbitmq.listener.simple.acknowledge-mode: manual #开启手动ack
     * @param message
     * @throws IOException
     */
    @RabbitListener(queues = RabbitMQConfig.QUEUE)
    public void consume(String msg, Channel channel, Message message) throws IOException {
        System.out.println("队列消息为："+msg);
        String correlationId = message.getMessageProperties().getCorrelationId();
        System.out.println("标识为："+correlationId);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

声明交换机&队列

六、RabbitMQ保证消息可靠性

confirm机制
可以通过confirm效果保证消息一定送达到Exchange，官方提供了三种，选择了对于效率影响最低的异步回调的效果

6.1、保证消息一定送达到Exchange

使用confirm机制

 public static final String QUEUE_NAME="confirms ";
    @Test
    public void publish() throws Exception {
        //1.获取连接对象
        Connection connection = RabbitMQConnectionUtil.getConnection();
        //2.创建Channel 通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //3.1 开启confirms 的异步回调
        channel.confirmSelect();
        String message="hello word!";
        //3.2 设置confirms的异步回调
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息成功发送到Exchange");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息没有发送到Exchange，尝试重试，或者保存到数据库做其他补偿操作");
            }
        });
        //3.3 设置return回调，确认消息是否路由到了队列
       channel.addReturnListener(new ReturnListener() {
           @Override
           public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
               System.out.println("消息没有到指定队列时，做其他的补偿措施！！");
           }
       });
        //3.4、设置消息持久化
        AMQP.BasicProperties prop =new AMQP.BasicProperties().builder()
                .deliveryMode(2)//消息持久化
               .build();

        //4. 发布消息
        channel.basicPublish("",QUEUE_NAME,true,prop,message.getBytes());
        System.out.println("消息发送成功");
        System.in.read();
    }

6.2、保证消息可以路由到Queue中

使用return 机制
为了保证Exchange上的消息一定可以送达到Queue

//6.2设置return 回调，确认消息是否路由到了Queue
channel.addReturnListener(new ReturnListener() {
           @Override
           public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
               System.out.println("消息没有到指定队列时，做其他的补偿措施！！");
           }
       });
//7.在发送消息时，将basicPublish方法参数中的mandatory设置为true，即可开启Return机制，当消息没有路由到队列中时，就会执行return	 回调

6.3、保证队列持久化消息

DeliveryMode设置消息持久化

//6.3、设置消息持久化
        AMQP.BasicProperties prop =new AMQP.BasicProperties().builder()
                .deliveryMode(2)//消息持久化
               .build();
 //4. 发布消息
        channel.basicPublish("",QUEUE_NAME,true,prop,message.getBytes());

6.4、保证消息者可以正常消费消息

详情看WorkQueue模式

6.5 SpringBoot实现上述操作

6.5.1 Confirm

编写配置文件开启Confirm机制

spring:
rabbitmq:
publisher-confirm-type: correlated # 新版本开启confirm机制
publisher-confirms: true # 老版本

在发送消息时，配置RabbitTemplate

@Test
    public void publishWithConfirms() throws IOException {
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b){
                    System.out.println("消息已经送达到交换机");
                }else{
                    System.out.println("消息没有送到到Exchange，需要做一些补偿操作！");
                }
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,"big.black.dog","message");
        System.out.println("消息发送成功");

        System.in.read();
    }

6.5.2 Return

编写配置文件开启Return机制

spring:
rabbitmq:
publisher-returns: true # 开启return机制

在发送消息时，配置RabbitTemplate

@Test
    public void publishWithReturn() throws IOException {
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                String msg = new String(message.getBody());
                System.out.println("消息失败："+msg+"路由队列失败！！做补救操作");
            }
        });
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,"big.black.dog","message");
        System.out.println("消息发送成功");

        System.in.read();
    }

6.5.3 消息持久化

//3.4、设置消息持久化
AMQP.BasicProperties prop =new AMQP.BasicProperties().builder()
                .deliveryMode(2)//消息持久化 #1
               .build();
        //4. 发布消息，  将参数mandatory设置为true，即可开启Return机制，当消息没有路由到队列中时，就会执行return回调
        channel.basicPublish("",QUEUE_NAME,true,prop,message.getBytes());

七、RabbitMQ死信队列 & 延迟交换机

###7.1、消息被消费者拒绝，requeue设置为false
###7.2.1、发送消息时设置消息的生存时间，如果生存时间到了，还没有被消费。
###7.2.2、也可以指定某个队列中所有消息的生存时间，如果生存时间到了，还没有被消费
###7.3、队列已经达到消息的最长长度后，再路由过来的消息直接变成死信
在这里插入图片描述

7.4、准备Exchange&Queue

@Configuration
public class DeadLetterConfig {
    public static final String NORMAL_EXCHANGE="normal-exchange";
    public static final String NORMAL_QUEUE="normal-queue";
    public static final String NORMAL_ROUTING_KEY="normal.#";


    public static final String DEAD_EXCHANGE="dead-exchange";
    public static final String DEAD_QUEUE="dead-queue";
    public static final String DEAD_ROUTING_KEY="dead.#";
    /**普通交换机*/
    @Bean
    public Exchange normalExchange(){
        return ExchangeBuilder.topicExchange(NORMAL_EXCHANGE).build();
    }
    /**普通队列*/
    @Bean
    public Queue normalQueue(){
        return QueueBuilder.durable(NORMAL_QUEUE)
                .deadLetterExchange(DEAD_EXCHANGE)//绑定死信队列
                .build();
    }
    /**普通队列绑定路由*/
    @Bean
    public Binding normalBingding(Queue normalQueue,Exchange normalExchange){
        return BindingBuilder.bind(normalQueue).to(normalExchange).with(DEAD_ROUTING_KEY).noargs();
    }

    /**死信交换机*/
    @Bean
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).build();
    }

    /**死信队列*/
    @Bean
    public Queue deadQueue(){
        return QueueBuilder.durable(DEAD_QUEUE ).build();
    }
    /**绑定死信队列和交换机*/
    @Bean
    public Binding deadBinding(Queue deadQueue,Exchange deadExchange){
        return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs();
    }
}

7.5、实现效果

基于消费者进行reject 或者 nack 实现死信效果

@Component
public class DeadListener {
    /**
     *
     * @param msg
     * @param channel 需要手动启动ACK 才能有效 spring.rabbitmq.listener.simple.acknowledge-mode: manual
     * @param message 需要手动启动ACK 才能有效 spring.rabbitmq.listener.simple.acknowledge-mode: manual
     */
    @RabbitListener(queues = DeadLetterConfig.NORMAL_QUEUE)
    public void comsume(String msg, Channel channel, Message message) throws IOException {
        System.out.println("接收到normal队列的消息："+msg);
//        channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
    }
}