惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

F
Fortinet All Blogs
WordPress大学
WordPress大学
The Cloudflare Blog
云风的 BLOG
云风的 BLOG
博客园 - Franky
D
Docker
小众软件
小众软件
阮一峰的网络日志
阮一峰的网络日志
H
Hackread – Cybersecurity News, Data Breaches, AI and More
Engineering at Meta
Engineering at Meta
MongoDB | Blog
MongoDB | Blog
U
Unit 42
M
MIT News - Artificial intelligence
B
Blog
GbyAI
GbyAI
C
Check Point Blog
P
Proofpoint News Feed
博客园 - 司徒正美
Hugging Face - Blog
Hugging Face - Blog
雷峰网
雷峰网
IT之家
IT之家
Google DeepMind News
Google DeepMind News
V
V2EX
Stack Overflow Blog
Stack Overflow Blog

祈雨的笔记

安全多方计算MPC spark原理解析 kueue执行源码分析 spark on k8s执行源码分析 spark-operator源码解析 系统压测遇到的缓存击穿问题 我的世界PC与安卓联机 蚂蚁金服流量投放平台的AIG改造 G1大对象致Old区占用率高 日志打印导致接口响应率下跌分析 Groovy加载类导致OOM分析 ERROR日志打印导致CPU满载 记OceanBase死锁超时 应用发版期间服务响应超时 Ark Serverless初探 系统优化复盘一二三 The user specified as a definer does not exist Kong网关初探 API网关选型调研 CPU火焰图常用工具 配置中心选型调研 root操作Nginx导致用户组错误 基于Proxifier使用代理 FastJSON字段智能匹配踩坑 Nacos初探 记一次Nginx服务器CPU满荷载故障 基于券系统分库分表的思考 limit不参与SQL成本计算致索引失效 Linux常用性能监控命令 golang低版本http2偶现400
rabbitmq(9)延时消息
祈雨的笔记 · 2018-03-25 · via 祈雨的笔记

1、概述

rabbitmq没有直接支持延时消息的功能,但可以通过死信队列实现延时消息的功能。

2、代码

2.1、生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
private final static String QUEUE_NAME = "MAIN_QUEUE";
private final static String _DIRECT_NAME = "_delay_delay";

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.7");
factory.setUsername("root");
factory.setPassword("123456");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, "amq.direct", QUEUE_NAME);

Map<String, Object> arguments = new HashMap<>();
arguments.put("x-dead-letter-exchange", "amq.direct");
arguments.put("x-dead-letter-routing-key", QUEUE_NAME);
channel.queueDeclare(_DIRECT_NAME, true, false, false, arguments);


long delay = 5*1000;

String message = "Hello World!";
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
AMQP.BasicProperties properties = builder.expiration(delay+"").build();
channel.basicPublish("", _DIRECT_NAME, properties, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

channel.close();
connection.close();
}

2.2、消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private final static String QUEUE_NAME = "MAIN_QUEUE";

public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.7");
factory.setUsername("root");
factory.setPassword("123456");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println(" [*] Waiting for messages.");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
}