springboot结合rabbitmq的延时队列实现订单超时自动关闭（附源码）

源代码：点我0积分下载

docker pull rabbitmq:management

启动

docker run -d --name rabbitmq --publish 5671:5671 
--publish 5672:5672 --publish 4369:4369 --publish 25672:25672 --publish 15671:15671 --publish 15672:15672 
rabbitmq:management

docker update rabbitmq --restart=always

注：
4369 – erlang发现口
5672 --client端通信口

15672 – 管理界面ui端口
25672 – server间内部通信口

添加rabbitmq的maven依赖

	
           org.springframework.boot
           spring-boot-starter-amqp
    

@Configuration
public class MyMqConfig {

    @Bean
    public Queue orderDelayQueue() {
        //String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map arguments
        //名字、是否持久化、是否排他、是否自动删除、自定义属性....
        Map arguments = new HashMap<>();
        //死信路由
        arguments.put("x-dead-letter-exchange","order-exchange");
        //死信路由键
        arguments.put("x-dead-letter-routing-key","order-release");
        //消息过期时间、单位：毫秒、60000代表1分钟
        arguments.put("x-message-ttl",60000);
        return new Queue("order-delay-queue",true,false,false,arguments);
    }

    @Bean
    public Queue orderReleaseOrderQueue() {
        return new Queue("order-release-queue",true,false,false);
    }

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

    @Bean
    public Binding orderCreate() {
        //String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map arguments
        //目的地、目的地类型、交换机、路由键
        return new Binding("order-delay-queue",Binding.DestinationType.QUEUE,"order-exchange","order-create",null);
    }

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

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Order implements Serializable {
    private String id;//订单编号
    private String content;//订单内容
    private String state;//订单状态
    private String createTime;//创建时间
    private String cancelTime;//取消时间
}

@Component
public class MqListener {

    
    @RabbitListener(queues = "order-release-queue")
    public void listener(Order order, Message message,Channel channel) throws IOException {
        System.out.println("接收到过期未支付的订单信息"+order);
        System.out.println("处理过期订单....");
        try{
            //TODO 判断订单是否已经支付成功，如果支付成功，无需再取消订单
            order.setCancelTime(new Date().toString());
            order.setState("已过期");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);//是否批量签收
        }catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);//是否重新入队
        }
        System.out.println("处理后的订单："+order);
    }

}

@RestController
public class MqController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    
    @GetMapping("/createOrder")
    public String createOrder(){
        //订单下单成功
        Order order = new Order(UUID.randomUUID().toString(), "订单商品信息", "未支付",new Date().toString(),null);
        //给mq发送消息
        rabbitTemplate.convertAndSend("order-exchange","order-create",order);
        return "下单成功"+order;
    }

	
    @GetMapping("/pay/{id}")
    public String pay(@PathVariable("id") String id){
		//TODO 修改订单状态为支付成功
        return id+"支付成功";
    }
}