RabbitMQ--实战

一、原生代码小Demo

　　pom：

        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.9.0</version>
        </dependency>

　　producer：

        // 1. 创建一个 ConnectionFactory 工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.124.8");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("lcl");
        factory.setPassword("123456");
        // 2. 创建一个 Connection
        Connection conn = factory.newConnection();
        // 3. 获取一个信道 Channel
        Channel channel = conn.createChannel();
        // 声明一个交换机
        channel.exchangeDeclare("cc", BuiltinExchangeType.DIRECT,true);
        // 4. 通过 Channel 发送消息
        String msg = "cuihuaaaa";
        for (int i = 0; i < 100; i++) {
            channel.basicPublish("cc", "hello", null, msg.getBytes());
            System.out.println(i);
        }
        // 5. 关闭资源
        channel.close();
        conn.close();

　　consumer：

        // 1. 创建 ConnetionFactory 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //factory.setUsername("guest");
        //factory.setPassword("guest");
        factory.setHost("192.168.124.8");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("lcl");
        factory.setPassword("123456");
        // 2. 获取 Connection 连接对象
        Connection connection = factory.newConnection();
        // 3. 创建 Channel 信道
        Channel channel = connection.createChannel();
        // 声明交换机
        String exchangeName = "aa";
        channel.exchangeDeclare(exchangeName, "direct", true);
        // 4. 声明队列
        String queueName = channel.queueDeclare().getQueue();
        String routingKey = "msg02";
        channel.queueBind(queueName, exchangeName, routingKey);
        // 5. 消费消息
        while (true){
            boolean autoAck = false;
            String consmerTag = "";
            channel.basicConsume(queueName, autoAck, consmerTag, new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // super.handleDelivery(consumerTag, envelope, properties, body);
                    String routingKey = envelope.getRoutingKey();
                    String contentType = properties.getContentType();
                    System.out.println("消费的路由键：" + routingKey + " 消费的内容类型：" + contentType);
                    long deliveryTag = envelope.getDeliveryTag();
                    // 确认消息
                    channel.basicAck(deliveryTag, false);
                    System.out.println("消费的消息体：");
                    String bodyMsg = new String(body, "UTF-8");
                    System.out.println(bodyMsg);

                }
            });
        }
    }

二、不同类型交换机的使用及Springboot集成RabbitMQ

　　pom：

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

　　application.yml：

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    host: 192.168.124.8
    port: 5672
    username: lcl
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

　　然后就是具体的实现了。

（一）直连型交换机Direct Exchange

　　直连交换机就是一对一，一个交换机对应一个队列，然后消费者消费指定队列的消息。

　　1、创建配置文件

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 DirectRabbitConfig {
    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("TestDirectQueue",true);
    }
    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        //  return new DirectExchange("TestDirectExchange",true,true);
        return new DirectExchange("TestDirectExchange",true,false);
    }
    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }
}

　　　　（1）对于创建队列，源码如下

　　　　　　　　durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效

　　　　　　　　exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable

　　　　　　　　autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。

　　　　　　一般设置一下队列的持久化就好,其余两个就是默认false

　　　　　　

　　　　（2） 对于创建交换机，源码如下：

　　　　　　　　durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效

　　　　　　　　autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。

　　　　　　

 　　　　（3）交换机和队列绑定

　　2、Producer

@RestController
public class SendMessageApi {
    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
    @GetMapping("/direct")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

　　3、consumer

　　这里也需要配置相关的ip和用户等信息，我这里是将生产者和消费者放到了同一个项目中，上面已经配置过，这里就不需要在配置了。

@Component
@RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
public class DirectConsumer {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver消费者收到消息  : " + testMessage.toString());
    }
}

（二）主题交换机Topic Exchange

　　主题交换机与直连交换机的区别就是可以一对多，也就是一个交换机对应多个队列，然后消费者消费指定队列的消息。

　　这里模拟分别有三个队列，分别是topic.man、topic.woman、topic.man.lcl，统一都绑定到一个交换机上：

　　　　一个是用全匹配绑定（topic.man），只有发送的routingkey为topic.man时，才会被路由到该队列

　　　　一个是#模糊匹配（topic.#），也就只要routingkey为topic.（后面只有一个单词），则会被路由到队列

　　　　一个是*模糊匹配（topic.*），也就是只要routingkey为topic.开头（后面可以有任意个单词），则会被路由到该队列

　　1、创建配置文件

@Configuration
public class TopicRabbitConfig {
    //绑定键
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";
    public final static String lcl = "topic.man.lcl";
    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }
    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }
    @Bean
    public Queue lclQueue() {
        return new Queue(TopicRabbitConfig.lcl);
    }
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }
    //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
    //这样只要是消息携带的路由键是topic.man,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列(topic后只能有任意个词)
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.*
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列(topic后只能有一个词)
    @Bean
    Binding bindingExchangeMessage3() {
        return BindingBuilder.bind(lclQueue()).to(exchange()).with("topic.*");
    }
}

　　2、Producer

　　发送三个消息，routingkey分别为topic.man、topic.woman、topic.man.lcl

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

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

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

　　3、Consumer

　　创建三个消费者，消费的队列分别为topic.man、topic.woman、topic.man.lcl

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

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

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

　　4、调用结果

　　　　调用方法一，发送的是topic.man，由于三个队列的routingkey分别为topic.man、topic.#、topic.*，因此三个队列都会有信息路由到。

　　　　调用方法二，发送的是topic.woman，topic.man肯定路由不到，topic.#、topic.*符合路由规则，因此这两个队列都会有信息路由到。

　　　　调用方法三，发送的是topic.man.lcl，只有topic.*被路由到，因此只有这一个队列会有消息。

　　　　　　

（三）广播交换机Fanout

　　广播交换机的特点就是不需要设置routingkey，直接就可以将交换机的消息路由到绑定的队列上，这里演示同一个交换机绑定多个队列。

　　1、创建配置文件

@Configuration
public class FanoutRabbitConfig {
    @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());
    }
}

　　2、Producer

    @GetMapping("/fanout")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: testFanoutMessage ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }

　　3、Consumer

　　分别创建三个消费者，消费绑定的三个队列

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

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

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

　　4、演示结果

　　　　发送一条消息后，三个消费端均可以接收

　　　　　　

 三、消息的可靠性

　　消息的可靠性分为投递的可靠性和接收可靠性，也就是生产者如何保证消息发送成功而不丢失，消费者如何可以正确的消费消息。

（一）消息投递可靠性分析

 　　1、方案一：使用消息确认监听

　　　　　　

 　　　　发送消息时，将业务数据和消息数据一起入库，并添加消息监听，如果消息被确认接收成功，则更新数据库中消息的状态。同时使用定时任务轮询那些没有被确认的消息，可以重新发送。

　　　　但是这种场景有个问题，就是在大并发的场景下，会对数据库造成很大的压力，从而造成系统瓶颈，这里可以使用数据库的分库分表从而降低数据库的压力。

　　2、方案二：消息延迟投递、做二次确认，回调检查

　　　　　　

 　　　　先将消息落库，然后再发送消息，这里发送了两条消息，一条是真正的业务消息，一条是原消息做延迟检查的消息，一般情况下会延迟2-5分钟发送。

　　　　消费者接收到消息后，进行消费，然后消费完成后给予确认

　　　　CallBack服务通过监听器，监听到消费者的确认之后，对消息做最终的存储

　　　　当接收到延迟投递检查时，callback服务监听到检查的消息，开始进行处理，如果已经接收到消费者消费完成的确认信息，则完成本次检查的消费；如果没有收到消费者消费确认信息，这时callback需要做补偿，其会主动发起RPC通信，让生产者再发送一次。

　　　　这么做的目的，其实就是少一次DB操作，从而减少数据库的压力。

（二）生产者Confirm消息确认与Return机制

　　消息确认是指在生产者投递消息后，如果broker收到消息，则会给生产者一个应答，生产者根据应答来确定消息是否已经被发送到broker

　　消息的确认是通过异步的方式来确认的。

　　消息确认和普通的发送消息有两点区别：需要开启确认模式和增加监听

　　1、原生代码模式

        //4 指定我们的消息投递模式: 消息的确认模式
        channel.confirmSelect();
        //5 发送一条消息
        channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
        System.err.println("------- 发送完成 ----------");
        //6 添加一个确认监听
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("------- 没有 ACK ----------");
            }
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.err.println("------- 收到 ACK -----------");
            }
        });

　　2、SpringBoot代码

　　　（1）application.yml配置文件

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    host: 192.168.124.8
    port: 5672
    username: lcl
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

    #消息确认配置项
    #确认消息已发送到交换机(Exchange) 老版本使用配置项 publisher-confirms: true
    publisher-confirm-type: correlated
    #确认消息已发送到队列(Queue)
    publisher-returns: true

　　（2）配置代码

@Configuration
public class ConfirmRabbitConfig {
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
            }
        });

        return rabbitTemplate;
    }
}

　　　　这里强调一下，一定要加上rabbitTemplate.setMandatory(true);或者在配置文件中设置spring.rabbitmq.template.mandatory=true，return消息机制才会生效。

　　　　可以看到上面写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback； 那么以上这两种回调函数都是在什么情况会触发呢？

　　　　推送消息存在四种情况：

　　　　　　（1）消息推送到server，但是在server里找不到交换机

　　　　　　（2）消息推送到server，找到交换机了，但是没找到队列

　　　　　　（3）消息推送到sever，交换机和队列啥都没找到

　　　　　　（4）消息推送成功

　　　　那么我先写几个接口来分别测试和认证下以上4种情况，消息确认触发回调函数的情况

　　　　（1）消息推送到server，但是在server里找不到交换机

　　　　　　写个测试接口，把消息推送到名为‘non-existent-exchange’的交换机上（这个交换机是没有创建没有配置的）：

    @GetMapping("/ack")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

　　　　输出结果：这种情况触发的是 ConfirmCallback 回调函数。并且由于没有对应的交换机，因此返回false

　　　　　　

 　　　　（2）消息推送到server，找到交换机了，但是没找到队列

　　　　　　这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，由于之前已经在DirectRabitConfig里面新增一个直连交换机，名叫‘lonelyDirectExchange’，但没给它做任何绑定配置操作

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

　　　　　　测试方法:

    @GetMapping("/ack2")
    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";
    }

　　　　　　调用结果：

　　　　　　

 　　　　可以看到这种情况，两个函数都被调用了； 这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true； 而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。 

　　　　（3）消息推送到sever，交换机和队列啥都没找到 这种情况其实和（1）一样。

　　　　　　结论： （3）这种情况触发的是 ConfirmCallback 回调函数。

　　　　（4）消息推送成功

　　　　那么测试下，按照正常调用之前消息推送的接口就行，就调用下 /direct接口

　　　　　　

 　　　　结论： 这种情况触发的是 ConfirmCallback 回调函数。

（三）消费端ack与重回队列

　　和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。

　　消息接收的确认机制主要存在三种模式：

　　　　1、自动确认， 这也是默认的消息确认情况。 AcknowledgeMode.NONE RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。 所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。 一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。

　　　　2、根据情况确认， 这个不做介绍

　　　　3、手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。

　　　　　　消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。

　　　　　　（1）basic.ack用于肯定确认

　　　　　　（2）basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）

　　　　　　（3）basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息

　　消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。 而basic.nack,basic.reject表示没有被正确处理。

　　着重讲下reject，因为有时候一些场景是需要重新入列的。

　　　　channel.basicReject(deliveryTag, true); 拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。 下次不想再消费这条消息了。

　　　　使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。 但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。

　　顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。

　　　　channel.basicNack(deliveryTag, false, true); 第一个参数依然是当前消息到的数据的唯一id; 第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。 第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。 同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

　　1、原生代码

            channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 获取传送标签
                    // long deliveryTag = envelope.getDeliveryTag();
                    // 确认消息
                    // channel.basicAck(deliveryTag, false);
                    System.out.println("消费的 Body：" + new String(body, "UTF-8"));
                }
            });

　　2、Springboot代码

　　（1）添加一个手动确认类

　　　　如果成功，返回肯定确认，如果catch到异常，返回否定确认。

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            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); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//            channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
            //根据不同的队列走不同的消费逻辑
            /*if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("执行TestDirectQueue中的消息的业务处理流程......");

            }
            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("执行fanout.A中的消息的业务处理流程......");

            }*/
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
    //{key=value,key=value,key=value} 格式转换成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;
    }

　　（2）添加配置代码

@Configuration
public class MessageAckListenerConfig {
    @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默认是自动确认，这里改为手动确认消息
        //设置一个队列
        container.setQueueNames("TestDirectQueue");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);
        return container;
    }
}

　　　　调用接口验证

　　　　　　

 　　（3）说明

　　　　上面的代码演示的是只有一个队列，但是有的业务场景需要设置多个队列手动确认，那么就可以参照注释掉的代码，在确认时，根据不同的队列名进行手动确认，同时在配置时，设置多个队列

四、消费端消息限流

　　消费端限流实际上就是消息在消费端的平稳投递消费，而不会导致一时间大量的消息打入到消费端，给消费端造成压力。

　　RabbitMQ 提供了一种 qos（服务质量保证）功能，即在非自动确认消息的前提下，如果一定数目的消息（通过基于 consumer 或者 channel 设置 Qos 的值）未被确认前，不进行消费新的消息。

　　在原生代码中，对channel进行设置

        /**
         * 限流设置:  prefetchSize：每条消息大小的设置，0是无限制
         * prefetchCount:标识每次推送多少条消息 一般是一条
         * global:false标识channel级别的  true:标识消费者级别的
         */
        channel.basicQos(0,1,false);

　　在Springboot中就更简单了，只需要对配置文件做调整即可

spring:
  #配置rabbitMq 服务器
  rabbitmq:
    listener:
      simple:
        #消费者最小数量
        concurrency: 1
        #消费之最大数量
        max-concurrency: 10
        #在单个请求中处理的消息个数，他应该大于等于事务数量(unack的最大数量)
        prefetch: 1
        # 是否手动确认
        #acknowledge-mode: manual

五、特殊队列（TTL队列与死信队列）

（一）TTL队列

　　TTL 是 Time To Live 的所写，也就是生存时间，RabbitMQ 支持消息的过期时间，在消息发送时可以进行指定。

　　RabbitMQ 支持队列的过期时间，从消息入队列开始计算，只要超过了队列的超时时间配置，那么消息会自动地清除。

　　生产者在发送时设置消息的有效期

Builder bd = new AMQP.BasicProperties().builder();
bd.deliveryMode(2);//持久化
bd.expiration("100000");//设置消息有效期100秒钟
BasicProperties pros = bd.build();
String message = "测试ttl消息";
channel.basicPublish(EXCHANGE_NAME, "error", true,false, pros, message.getBytes());

　　消费者设置队列中消息的有效期

//设置队列上所有的消息的有效期,单位为毫秒
Map<String, Object> argss = new HashMap<String , Object>();
arguments.put("x-message-ttl " , 5000);//5秒钟
channel.queueDeclare(queueName , durable , exclusive , autoDelete , arguments) ;

（二）死信队列

　　死信队列 DLX, Dead-Letter-Exchange，利用 DLX，当消息在一个队列中变成死信（dead message）之后，它能被重新 pulish 到另一个 Exchange，这个 Exchange 就是 DLX。

　　当某个队列中有死信时，RabbitMQ 就会自动的将这个消息重新发布到设置的 Exchange 上去，进而被路由到另一个队列。

　　死信队列也叫延时队列。基于TTL模式的延时队列会涉及到2个交换机、2个路由键、2个队列，如下图所示：

　　　　（1）生产者将消息(msg)和路由键(routekey)发送指定的死信交换机(DelayExchange)上

　　　　（2）死信交换机(delayexchange)根据路由键(routekey1)找到绑定自己的死信队列(DelayQueue)并把消息给它

　　　　（3）消息(msg)到期死亡变成死信转发给死信接收交换机(ReceiveExchange)

　　　　（4）死信接收交换机(ReceiveExchange)根据路由键(routekey2)找到绑定自己的死信接收队列(ReceiveQueue)并把消息给它

　　　　（5）死信接收队列(ReceiveQueue)再把消息发送给监听它的消费者(customer）

　　　　　　

 　　消息变成死信队列的几种情况：

　　　　1、消息被拒绝（basic.reject / basic.nack）并且 requeue = false 

　　　　2、消息 TTL 过期

　　　　3、队列达到最大长度

　　DLX 也是一个正常的 Exchange，和一般的 Exchange 没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。

 　　代码样例：

　　　　1、配置队列绑定

　　　　　　分别创建死信队列交换机、接收对接交换机、死信队列、接收队列以及绑定关系

@Configuration
public class DelayRabbitConfig {

    /**
     * 死信交换机
     * @return
     */
    @Bean
    public DirectExchange delayExchange(){
        return new DirectExchange("delay_exchange");
    }

    /**
     * 死信队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        Map<String,Object> map = new HashMap<>(16);
        map.put("x-dead-letter-exchange","receive_exchange");
        map.put("x-dead-letter-routing-key", "receive_key");
        return new Queue("delay_queue",true,false,false,map);
    }

    /**
     * 给死信队列绑定交换机
     * @return
     */
    @Bean
    public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange){
        return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay_key");
    }

    /**
     * 死信接收交换机
     * @return
     */
    @Bean
    public DirectExchange receiveExchange(){
        return new DirectExchange("receive_exchange");
    }

    /**
     * 死信接收队列
     * @return
     */
    @Bean
    public Queue receiveQueue(){
        return new Queue("receive_queue");
    }

    /**
     * 死信交换机绑定消费队列
     * @return
     */
    @Bean
    public Binding receiveBinding(Queue receiveQueue,DirectExchange receiveExchange){
        return BindingBuilder.bind(receiveQueue).to(receiveExchange).with("receive_key");
    }

　　　　上面的设置死信队列时的key是固定的，如果不清楚，可以参看以下控制台。

　　　　　　

 　　　　2、Producer

    @GetMapping("/delay")
    public String sendDelayMessage() {
        //这里的消息可以是任意对象，无需额外配置，直接传即可
        String messageData = "sendDelayMessage~~~~~~~~~ ";
        log.info("===============延时队列生产消息====================");
        log.info("发送时间:{},发送内容:{}", LocalDateTime.now(), messageData);
        this.rabbitTemplate.convertAndSend(
                "delay_exchange",
                "delay_key",
                messageData,
                message -> {
                    //注意这里时间要是字符串形式
                    message.getMessageProperties().setExpiration("60000");
                    return message;
                }
        );
        log.info("{}ms后执行", 60000);
        return "OK~";
    }

　　　　3、consumer

@Component
@Slf4j
public class DelayConsumer {

    @RabbitListener(queues = "receive_queue")
    public void cfgUserReceiveDealy(List<Integer> list, Message message, Channel channel) throws IOException {
        log.info("===============接收队列接收消息====================");
        log.info("接收时间:{},接受内容:{}", LocalDateTime.now(), list.toString());
        //通知 MQ 消息已被接收,可以ACK(从队列中删除)了
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        try {
            log.error("============dosomething.....!==============");
        } catch (Exception e) {
            log.error("============消费失败,尝试消息补发再次消费!==============");
            log.error(e.getMessage());
            /**
             * basicRecover方法是进行补发操作，
             * 其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer(集群)接收到，
             * 设置为false是只补发给当前的consumer
             */
            channel.basicRecover(false);
        }
    }
}

　　　　4、验证