springboot整合rabbitMq實現訊息延時傳送
阿新 • • 發佈:2020-11-15
實現思路:利用mq的ttl設定訊息失效時間 當達到設定時間後通過交換機到達死信佇列中,消費者端繫結讀取死信佇列中資訊來達到延時傳送訊息的功能。
demo 如下:
(1)在pom.xml 中引入rabbitMq相關包
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
(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<String, Object> 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分鐘後到達指定的佇列中:
設定的延時交換機為:
控制檯列印資訊為:
&n