延时消息队列

使用场景

目的是保证事务最终一致性

消息的TTL

Exchanges 死信

流程

整合使用

配置

package com.jhj.gulimall.ssoclient.config;

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

import java.util.HashMap;
import java.util.Map;


@Configuration
public class MyMQConfig {
    

    /** * 容器中的Binding Queue Exchange 都会自动创建（RabbitMQ中没有的情况下） * RabbitMQ中只要有就不会覆盖 */

    @Bean
    public Queue orderDelayQueue(){
    
        Map<String,Object> arguments = new HashMap<>();

        //死信路由
        arguments.put("x-dead-letter-exchange","order-event-exchange");
        //死信路由键
        arguments.put("x-dead-letter-routing-key","order.release.order");
        //过期时间
        arguments.put("x-message-ttl","60000");
        Queue queue = new Queue("order.delay.queue", true, false, false,arguments);

        return queue;
    }

    @Bean
    public Queue orderReleaseQueue(){
    
        Queue queue = new Queue("order.release.order.queue", true, false, false);
        return queue;
    }

    @Bean
    public Exchange orderEventExchange(){
    
        return new TopicExchange("order-event-exchange", true, false);
    }

    @Bean
    public Binding orderCreateOrderBinding(){
    
       return new Binding("order.delay.queue", Binding.DestinationType.QUEUE,"order-event-exchange","order.create.order",null);
    }

    @Bean
    public Binding orderReleaseOrderBinding(){
    
        return new Binding("order.release.order.queue", Binding.DestinationType.QUEUE,"order-event-exchange","order.release.order",null);
    }
}

发送

/** * 发送消息 */
    @Test
    public void sendMessage(){
    
        rabbitTemplate.convertAndSend("order-event-exchange","order.create.order","test");
    }

接收

@RabbitListener(queues = {
    "order.release.order.queue"})
    public void listener(Message message, String s, Channel channel){
    
        message.getBody();
        System.out.println(s);
        try {
    
            // 第二个参数代表 是否批量回复 确认成功回复 收货
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            // 第二个参数代表 是否批量回复
            // 第三个参数代表 是否重新放到queue中 退货 确认失败回复 false 丢弃 true 重新入队
            //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
        } catch (IOException e) {
    
            throw new RuntimeException(e);
        }
    }

如何保证消息可靠性

消息丢失

消息重复

消息积压

作者声明

如有问题，欢迎指正！