SpringBoot整合rabbitMq实现消息延时发送
java1234
共 3411字，需浏览 7分钟
·
2020-11-17 09:21
点击上方蓝色字体，选择“标星公众号”
优质文章，第一时间送达
作者 | 山阴路的秋天
来源 | urlify.cn/VRfei2
实现思路：利用mq的ttl设置消息失效时间 当达到设置时间后通过交换机到达死信队列中，消费者端绑定读取死信队列中信息来达到延时发送消息的功能。
demo 如下：
（1）在pom.xml 中引入rabbitMq相关包
org.springframework.boot
spring-boot-starter-amqp
（2）创建rabbitMq连接的工具类
public class MqConnectionUtil {
public static Connection getConnection() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("virtualHost_wl");
factory.setUsername("admin");
factory.setPassword("admin");
Connection connection = factory.newConnection();
return connection;
}
}
（3）创建生产者发送消息
public class Send {
private final static String QUEUE_NAME = "msg_ttl_queue";
public static void main(String[] argv) throws Exception {
//获取连接
Connection connection = MqConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
//设置延时队列
HashMap args = new HashMap<>();
args.put("x-dead-letter-exchange", "delay-exchange");
args.put("x-dead-letter-routing-key", "msg_ttl_routingKey");
channel.queueDeclare("delay_queue", true, false, false, args);
//声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//绑定路由(delay-queue队列消息失效之后转发到msg_ttl_queue中)
channel.queueBind(QUEUE_NAME, "delay-exchange", "msg_ttl_routingKey");
//设置延时属性
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//设置延时时间 1分钟
AMQP.BasicProperties properties = builder.expiration("60000").deliveryMode(2).build();
//消息内容
SimpleDateFormat sft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String message = "生产者创建时间： " + sft.format(new Date());
channel.basicPublish("", "delay_queue", properties, message.getBytes());
System.out.println("Sent： '" + message + "'");
//关闭通道和连接
channel.close();
connection.close();
}
}
（4）创建对应的消费者接受消息
public class Receive {
private final static String QUEUE_NAME = "msg_ttl_queue";
public static void main(String[] argv) throws Exception {
//获取连接
Connection connection = MqConnectionUtil.getConnection();
//创建通道
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 定义队列的消费者
MyConsumer myConsumer = new MyConsumer(channel);
// 监听队列
channel.basicConsume(QUEUE_NAME, true, myConsumer);
}
}
自定义消费者类为：
public class MyConsumer extends DefaultConsumer {
public MyConsumer(Channel channel) {
super(channel);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SimpleDateFormat sft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println("消费者接受时间：" + sft.format(new Date()));
System.out.println("consumerTag: " + consumerTag);
System.out.println("envelope: " + envelope);
System.out.println("properties: " + properties);
System.out.println("body: " + new String(body));
}
}
生产者发送消息到延时队列：
到达设定的失效时间1分钟后到达指定的队列中：
设置的延时交换机为：
控制台打印信息为：
粉丝福利：实战springboot+CAS单点登录系统视频教程免费领取
???
?长按上方微信二维码 2 秒即可获取资料感谢点赞支持下哈 
评论