第一步， 首先需要创建2个队列。Queue1和Queue2。Queue1是一个消息缓冲队列，在这个队列里面实现消息的过期转发。如下图，设置Dead letter exchange和Dead letter routing key。设置这两个属性就是当消息在这个队列中expire后，采用哪个路由发送。这个dlx的exchange需要事先创建好，就是一个普通的exchange。由于我们还需要向Queue1发送消息，那么还需要创建一个exchange，并且和Queue1绑定。例子中，exchange同样取名：queue1。
我们还需要建一个Queue2，这个队列用于消息在Queue1中过期后转发的目标队列。所以这个Queue2队列建好以后，需要绑定Queue1设置的死信路由：dlx。完成Queue2的绑定以后，环境就搭建完成了。
第二步，实现消息的Producer。由于我们的目的是让进入Queue1的消息过期，然后自动转送到Queue2中，所以发送的时候，需要设置过期时间。
ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("bsp"); factory.setPassword("123456"); factory.setVirtualHost("https://www.jb51.net/"); factory.setHost("10.23.22.42"); factory.setPort(5672); conn = factory.newConnection(); channel = conn.createChannel(); byte[] messageBodyBytes = "Hello, world!".getBytes(); byte i = 10; while (i-- > 0) { channel.basicPublish("queue1", "queue1", new AMQP.BasicProperties.Builder().expiration(String.valueOf(i * 1000)).build(), new byte[] { i }); }
上面的代码我模拟了1-10号消息，消息的内容里面是1-10。过期的时间是10-1秒。这里要注意，虽然10是第一个发送，但是它过期的时间最长。
第三步，实现消息的Consumer。Consumer就是延迟任务的具体实施者。由于具体的任务往往是一个比较耗时的任务，所以一般来说，任务一般在异步线程中执行。
ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("bsp"); factory.setPassword("123456"); factory.setVirtualHost("https://www.jb51.net/"); factory.setHost("10.23.22.42"); factory.setPort(5672); conn = factory.newConnection(); channel = conn.createChannel(); channel.basicConsume("queue2", true, "consumer", new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); //do some work async System.out.println(body[0]); } });
运行后如上面的程序，过了10s以后，消费者开始收到数据，但是它是一次性收到如下结果：
10、9 、8 、7 、6、5 、4 、3 、2 、1
Consumer第一个收到的还是10。虽然10是第一个放进队列，但是它的过期时间最长。所以由此可见，即使一个消息比在同一队列中的其他消息提前过期，提前过期的也不会优先进入死信队列，它们还是按照入库的顺序让消费者消费。如果第一进去的消息过期时间是1小时，那么死信队列的消费者也许等1小时才能收到第一个消息。参考官方文档发现“Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered).”只有当过期的消息到了队列的顶端（队首），才会被真正的丢弃或者进入死信队列。
所以在考虑使用RabbitMQ来实现延迟任务队列的时候，需要确保业务上每个任务的延迟时间是一致的。如果遇到不同的任务类型需要不同的延时的话，需要为每一种不同延迟时间的消息建立单独的消息队列。
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对脚本之家的支持。
您可能感兴趣的文章: