延时队列未到时间 手动消费demo

好的，下面是一个简单的延迟队列手动消费的示例。

首先，需要在 pom.xml 文件中添加 RabbitMQ 的依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

然后，在 application.properties 文件中设置 RabbitMQ 的连接信息：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

接着，创建一个 RabbitMQ 的配置类，用于创建延迟队列和交换机：

@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue delayQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "myexchange");
        args.put("x-dead-letter-routing-key", "mykey");
        return new Queue("mydelayqueue", true, false, false, args);
    }

    @Bean
    public Queue queue() {
        return new Queue("myqueue", true);
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange("myexchange");
    }

    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("mykey");
    }

    @Bean
    public Binding delayBinding(Queue delayQueue, TopicExchange exchange) {
        return BindingBuilder.bind(delayQueue).to(exchange).with("mydelaykey");
    }
}

上面的代码创建了一个名为 "mydelayqueue" 的延迟队列，一个名为 "myqueue" 的普通队列，一个名为 "myexchange" 的交换机，并使用 "mykey" 绑定它们。

其中，"mydelayqueue" 队列设置了 x-dead-letter-exchange 和 x-dead-letter-routing-key 属性，用于指定消息过期后的转发地址。

最后，创建一个手动消费者类，用于手动接收延迟队列中的消息，并将过期的消息转发到普通队列：

@Component
public class ManualConsumer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = "mydelayqueue")
    public void receiveMessage(Message message, Channel channel) throws IOException {
        String body = new String(message.getBody(), "UTF-8");
        System.out.println("Received message: " + body);

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 模拟处理消息的耗时
            Thread.sleep(5000);
            // 手动确认消息已经被消费
            channel.basicAck(deliveryTag, false);
            // 将过期的消息转发到普通队列
            rabbitTemplate.convertAndSend("myexchange", "mykey", body);
            System.out.println("Message forwarded to myqueue: " + body);
        } catch (Exception e) {
            // 消费消息出现异常，拒绝消息并重新加入队列
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

上面的代码使用 @RabbitListener 注解监听名为 "mydelayqueue" 的延迟队列，并在接收到消息时手动确认消息已经被消费。

接着，模拟处理消息的耗时，并将过期的消息转发到普通队列 "myqueue"。

如果消费消息出现异常，拒绝消息并重新加入队列。

希望这个示例能对你有所帮助！