RabbitMQ-ttl-dlx的讲解文档.md · cd封风疯/rabbit-dlx

基于RabbitMQ的TTL以及死信队列，使用SpringBoot实现延迟付款，手动补偿操作

1.代码

application.properties

~~~properties
spring.rabbitmq.host=192.168.174.128
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=root
spring.rabbitmq.password=123456
spring.rabbitmq.port=5672

# 配置数据库连接池及数据库驱动
spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
# 配置mysql的链接信息
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.url=jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8

mybatis.mapper-locations="classpath*:mapper/*.xml"

logging.level.com.lagou.demo.dao=debug
~~~

实体类order

~~~java
@Data
@Accessors(chain = true)
@Table(name = "c_order")
public class Order {

/**
     * 订单id
     */
    @Id
    private String id;

/**
     * 订单状态 0：待支付 1：已支付  2：已超时取消
     */
    private Integer status;
}
~~~

RabbitmqConfig，配置正常的订单消费队列和死信队列

~~~java
@Configuration
public class RabbitmqConfig {

@Bean("dlx_queue")
    public Queue dlxQueue() {
        return QueueBuilder.durable("dlx_queue").build();
    }

@Bean("dlx_exchange")
    public Exchange dlxExchange() {
        return ExchangeBuilder.directExchange("dlx_exchange").durable(true).build();
    }

@Bean
    public Binding dlxBinding(@Qualifier("dlx_queue") Queue queue, @Qualifier("dlx_exchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("dlx").noargs();
    }

/**
     * 正常下单消费队列
     */
    @Bean("order_queue")
    public Queue orderQueue() {
        HashMap<String, Object> argumentMap = new HashMap<>(2);
        //设置消息20s过期
        argumentMap.put("x-message-ttl", 20 * 1000);
        argumentMap.put("x-dead-letter-exchange", "dlx_exchange");
        argumentMap.put("x-dead-letter-routing-key", "dlx");
        return QueueBuilder.durable("order_queue").withArguments(argumentMap).build();
    }

/**
     * 正常下单消费交换机
     */
    @Bean("order_exchange")
    public Exchange orderExchange() {

return ExchangeBuilder.directExchange("order_exchange").durable(true).build();
    }

@Bean
    public Binding orderBinding(@Qualifier("order_queue") Queue queue,
                                @Qualifier("order_exchange") Exchange exchange) {

return BindingBuilder.bind(queue).to(exchange).with("order.biz").noargs();

}

}
~~~

controller: 一个下单接口，一个付款接口

~~~java
@RestController
@RequestMapping("/order")
public class OrderController {

@Autowired
    OrderService orderService;

/**
     * 用户下单接口（==>发送一条等待付款的消息）
     */
    @RequestMapping("/placeOrder")
    public String placeOrder(){
        return orderService.placeOrder();
    }

/**
     * 用户付款
     */
    @GetMapping("/paidOrder/{orderId}")
    public String paidOrder(@PathVariable("orderId") String orderId){
        return orderService.paidOrder(orderId);
    }

}
~~~

serviceImpl：主要接口实现逻辑

~~~java
@Service
@Slf4j
public class OrderServiceImpl implements OrderService {

@Autowired
    OrderDao orderDao;
    @Autowired
    RabbitTemplate rabbitTemplate;

@Override
    public String placeOrder() {

//1.模拟用户下单
        String orderId = UUID.randomUUID().toString();
        Integer orderStatus = OrderStatus.TO_BE_PAID;
        Order order = new Order();
        order.setId(orderId).setStatus(orderStatus);
        orderDao.insertOneOrder(order);
        log.info("用户下单成功，订单号为:{}，订单的状态为:{}", orderId, orderStatus);

//2.发送一条等待付款的消息
        rabbitTemplate.convertAndSend("order_exchange", "order.biz", orderId);
        log.info("用户生成了一个待付款的订单，订单号为:{}", orderId);

return "用户下单成功,跳转到等待付款页面";
    }

@Override
    public String paidOrder(String orderId) {

//1.检查订单状态，只有订单待付款才消费订单信息
        Order order = orderDao.getOrderById(orderId);
        Integer status = order.getStatus();
        if (Objects.equals(status, OrderStatus.TO_BE_PAID)) {
            //使用拉消息的模式
            String receivedOrderId = (String) rabbitTemplate.receiveAndConvert("order_queue");
            log.info("用户在规定时间内付款主动消费了一条待付款的订单消息，该订单号为:{}", receivedOrderId);

//将订单的状态修改为已支付
            orderDao.updateOrderStatusById(receivedOrderId, OrderStatus.HAS_PAID);
            log.info("订单号为:{}的订单状态修改为已支付", receivedOrderId);
            return "跳转到付款成功页面";
        }
        return "付款失败";
    }
}
~~~

处理超时未支付的消费者，即处理死信队列的逻辑，修改订单状态为超时未支付,使用推消息模式

~~~java
![微信截图_20210322084614](C:\Users\19242\Desktop\微信截图_20210322084614.png)@Component
@Slf4j
public class OrderConsumer {

@Autowired
    OrderDao orderDao;

/**
     * 用户超时未支付
     */
    @RabbitListener(queues = "dlx_queue")
    public void timeoutOrderMessage(@Payload String message) {

log.info("接收到用户的超时未支付订单,订单号为:{}", message);
        //将订单的状态修改为超时未支付
        orderDao.updateOrderStatusById(message, OrderStatus.CANCELLED_OVER_TIME);
        log.info("订单号为:{}的订单状态修改为超时未支付", message);

//查询用户所有付款超时而取消的订单
        List<Order> overTimeOrder = orderDao.getOrderByStatus(OrderStatus.CANCELLED_OVER_TIME);
        log.info("用户所有付款超时而取消的订单为:{}", overTimeOrder);
    }
}
~~~

2.效果

1.用户在规定的时间内付款

![](img/微信截图_20210322084652.png)

![](img/微信截图_20210322084700.png)

![](img/微信截图_20210322084614.png)

2.用户超时未支付

![](img/微信截图_20210322084652.png)

订单超时的时间设置为20s,等待20s后

![](img/微信截图_20210322084817.png)

rabbitmq控制台查看交换机和队列的状态

![](img/微信截图_20210322090310.png)

![](img/微信截图_20210322090320.png)

cd封风疯/rabbit-dlx

简介

发行版

贡献者

近期动态

cd封风疯/rabbit-dlx .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

cd封风疯/rabbit-dlx