傻大方


首页 > 潮·科技 > >

SpringBoot整合rabbitMq实现消息延时发送



按关键词阅读:

作者 | 山阴路的秋天
来源 | urlify.cn/VRfei2
实现思路:利用mq的ttl设置消息失效时间 当达到设置时间后通过交换机到达死信队列中 , 消费者端绑定读取死信队列中信息来达到延时发送消息的功能 。
demo 如下:
(1)在pom.xml 中引入rabbitMq相关包
org.springframework.bootspring-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);}@Overridepublic 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));}}生产者发送消息到延时队列:
SpringBoot整合rabbitMq实现消息延时发送文章插图
到达设定的失效时间1分钟后到达指定的队列中:
SpringBoot整合rabbitMq实现消息延时发送文章插图
设置的延时交换机为:
SpringBoot整合rabbitMq实现消息延时发送文章插图
【SpringBoot整合rabbitMq实现消息延时发送】控制台打印信息为:
SpringBoot整合rabbitMq实现消息延时发送文章插图


    稿源:(未知)

    【傻大方】网址:http://www.shadafang.com/c/111J304232020.html

    标题:SpringBoot整合rabbitMq实现消息延时发送


    上一篇:行业标准|自拍清晰就是行业标准?vivo S7认为“清晰”只是“小儿科”

    下一篇:体验|OPPO二代AR眼镜问世,或成为行业新标杆