RabbitMQ消息队列实现延迟任务示例

for (long i = 1; i <= 10; i++) {
    /* 1.模拟生成订单 */
    BuOrder order = createOrder(i);
    /* 2.订单入库 */
    orderService.removeById(order);
    orderService.saveOrUpdate(order);
    /* 3.将订单存入信息Redis */
    RedisUtils.setObject(RabbitTemplateConfig.ORDER_PREFIX + i, order);
    /* 4.向RabbitMQ异步投递消息 */
    rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));
}

public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    if (correlationData == null) {
        return;
    }
    String key = ORDER_PREFIX + correlationData.getId();
    if (ack) {
        /* 如果消息投递成功，则删除Redis中订单数据，回收内存 */
        RedisUtils.deleteObject(key);
    } else {
        /* 从Redis中读取订单数据，重新投递 */
        BuOrder order = RedisUtils.getObject(key, BuOrder.class);
        /* 重新投递消息 */
        rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));
    }
}

@RabbitListener(queues = RabbitmqConfig.DELAY_QUEUE_NAME)
public void consumeNode01(Channel channel, Message message, BuOrder order) throws IOException {
    if (Objects.equals(0, order.getOrderStatus())) {
        /* 修改订单状态，设置为关闭状态 */
        orderService.updateById(new BuOrder(order.getOrderId(), -1));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info(String.format("消费者节点01消费编号为【%s】的消息", order.getOrderId()));
    }
}

<dependency>
  <groupId>xin.altitude.cms</groupId>
  <artifactId>ucode-cms-common</artifactId>
  <version>1.4.3.1</version>
</dependency>