缓存架构之史上讲的最明白的RabbitMQ可靠消息传输实战演练

@BeanDirectExchange advanceExchange() {    return new DirectExchange(exchangeName);
}

@Beanpublic Queue advanceQueue() {    return new Queue(queueName);
}

@Beanpublic Queue helloQueue() {    //将普通队列绑定到私信交换机上
    Map<String, Object> args = new HashMap<>(2);
    args.put(DEAD_LETTER_QUEUE_KEY, deadExchangeName);
    args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKey);
    Queue queue = new Queue(queueName, true, false, false, args);    return queue;
}

# 开启发送确认
spring.rabbitmq.publisher-confirms=true# 开启发送失败退回
spring.rabbitmq.publisher-returns=true

# 开启ACK
spring.rabbitmq.listener.simple.acknowledge-mode=manual

@Configurationpublic class RabbitConfig {    public final static String queueName = "hello_queue";    /**
     * 死信队列：
     */
    public final static String deadQueueName = "dead_queue";    public final static String deadRoutingKey = "dead_routing_key";    public final static String deadExchangeName = "dead_exchange";    /**
     * 死信队列 交换机标识符
     */
    public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";    /**
     * 死信队列交换机绑定键标识符
     */
    public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";    @Bean
    public Queue helloQueue() {        //将普通队列绑定到私信交换机上
        Map<String, Object> args = new HashMap<>(2);
        args.put(DEAD_LETTER_QUEUE_KEY, deadExchangeName);
        args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKey);
        Queue queue = new Queue(queueName, true, false, false, args);        return queue;
    }    /**
     * 死信队列：
     */

    @Bean
    public Queue deadQueue() {
        Queue queue = new Queue(deadQueueName, true);        return queue;
    }    @Bean
    public DirectExchange deadExchange() {        return new DirectExchange(deadExchangeName);
    }    @Bean
    public Binding bindingDeadExchange(Queue deadQueue, DirectExchange deadExchange) {        return BindingBuilder.bind(deadQueue).to(deadExchange).with(deadRoutingKey);
    }

}

@Componentpublic class HelloSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{    @Autowired
    private RabbitTemplate rabbitTemplate;    public void send(String exchange, String routingKey) {
        String context = "你好现在是 " + new Date();
        System.out.println("send content = " + context);        this.rabbitTemplate.setMandatory(true);        this.rabbitTemplate.setConfirmCallback(this);        this.rabbitTemplate.setReturnCallback(this);        this.rabbitTemplate.convertAndSend(exchange, routingKey, context);
    }    /**
     * 确认后回调:
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {        if (!ack) {
            System.out.println("send ack fail, cause = " + cause);
        } else {
            System.out.println("send ack success");
        }
    }    /**
     * 失败后return回调：
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("send fail return-message = " + new String(message.getBody()) + ", replyCode: " + replyCode + ", replyText: " + replyText + ", exchange: " + exchange + ", routingKey: " + routingKey);
    }
}

@Component@RabbitListener(queues = RabbitConfig.queueName)public class HelloReceiver {    @RabbitHandler
    public void process(String hello, Channel channel, Message message) throws IOException {        try {
            Thread.sleep(2000);
            System.out.println("睡眠2s");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }        try {            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉；否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            System.out.println("receiver success = " + hello);
        } catch (Exception e) {
            e.printStackTrace();            //丢弃这条消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            System.out.println("receiver fail");
        }

    }

}

//1、exchange, queue 都正确, confirm被回调, ack=true@RequestMapping("/send1")@ResponseBodypublic String send1() {
    helloSender.send(null, RabbitConfig.queueName);    return "success";
}//2、exchange 错误, queue 正确, confirm被回调, ack=false@RequestMapping("/send2")@ResponseBodypublic String send2() {
    helloSender.send("fail-exchange", RabbitConfig.queueName);    return "success";
}//3、exchange 正确, queue 错误, confirm被回调, ack=true; return被回调 replyText:NO_ROUTE@RequestMapping("/send3")@ResponseBodypublic String send3() {
    helloSender.send(null, "fail-queue");    return "success";
}//4、exchange 错误, queue 错误, confirm被回调, ack=false@RequestMapping("/send4")@ResponseBodypublic String send4() {
    helloSender.send("fail-exchange", "fail-queue");    return "success";
}