實作思路:利用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分鐘後到達指定的隊列中:
設定的延時交換機為: