共计 2114 个字符,预计需要花费 6 分钟才能阅读完成。
rabbitmq 的延迟队列,我们可以通过死信交换器来实现。
生产者发送消息,定义 2 秒后消息过期,消息就会进入死信交换器,最后到死信队列。
// 定义队列的名称 | |
public final static String QUEUE_NAME = "queue.scheduler"; | |
// 定义交换器的名称 | |
public final static String EXCHANGE_NAME = "exchange.scheduler"; | |
// 定义路由的名称 | |
public final static String ROUTE_NAME = "route.scheduler"; | |
// 定义死信队列的名称 | |
public final static String DLX_QUEUE_NAME = "scheduler.queue.name"; | |
// 定义死信交换器的名称 | |
public final static String DLX_EXCHANGE_NAME = "scheduler.exchange.name"; | |
public static void main(String[] args) throws IOException, TimeoutException { | |
// 声明一个连接工厂 | |
ConnectionFactory factory = new ConnectionFactory(); | |
// 创建一个与 rabbitmq 服务器的连接 | |
// 创建一个 Channel | |
try (Connection connection = factory.newConnection(); | |
Channel channel = connection.createChannel()) { | |
// 定义交换器 | |
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, false, null); | |
Map<String, Object> arguments = new HashMap<String, Object>(); | |
arguments.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME); | |
arguments.put("x-message-ttl", 2000); | |
// 定义队列 | |
channel.queueDeclare(QUEUE_NAME, false, false, false, arguments); | |
// 绑定队列 | |
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTE_NAME); | |
// 定义死信交换器 | |
channel.exchangeDeclare(DLX_EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, false, null); | |
// 定义死信队列 | |
channel.queueDeclare(DLX_QUEUE_NAME, false, false, false, null); | |
// 绑定死信队列 | |
channel.queueBind(DLX_QUEUE_NAME, DLX_EXCHANGE_NAME, ROUTE_NAME); | |
// 发送消息 | |
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); | |
channel.basicPublish(EXCHANGE_NAME, ROUTE_NAME, true, null, df.format(new Date()).getBytes()); | |
} | |
} |
消费者,从私信队列获取消息,可以得到延迟后的消息。
public static void main(String[] args) throws IOException, TimeoutException { | |
// 声明一个连接工厂 | |
ConnectionFactory factory = new ConnectionFactory(); | |
// 创建一个与 rabbitmq 服务器的连接 | |
Connection connection = factory.newConnection(); | |
// 创建一个 Channel | |
Channel channel = connection.createChannel(); | |
System.out.println("Waiting for messages."); | |
// 异步回调处理 | |
DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8"); | |
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); | |
System.out.println(df.format(new Date()) + "Received'" + message + "'"); | |
}; | |
// 接收消息 | |
channel.basicConsume(ProducerScheduler.DLX_QUEUE_NAME, true, deliverCallback, consumerTag -> {}); | |
} |
运行结果如下,达到了延迟队列的效果。除此之外,还可以用启用延迟插件。
正文完