基于rabbitmq实现分布式延时任务调度(代码片段)

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.7.3</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.58</version>
        </dependency>
    </dependencies>

 1 import com.rabbitmq.client.*;
 2 
 3 import java.io.IOException;
 4 import java.util.concurrent.TimeoutException;
 5 
 6 /**
 7  * 1.连接RabbitMQ
 8  * 2.添加延时任务
 9  * 3.消费延时任务
10  */
11 public class RabbitMQUtil 
12 
13     private static Connection conn;
14     private static Channel channel;
15 
16     /**
17      * 初始化RabbitMQ连接与channel
18      */
19     static 
20         ConnectionFactory factory = new ConnectionFactory();
21         factory.setUsername("guest");
22         factory.setPassword("guest");
23         factory.setVirtualHost("/");
24         factory.setHost("localhost");
25         factory.setPort(5672);
26 
27         try 
28             conn = factory.newConnection();
29             channel = conn.createChannel();
30          catch (IOException e) 
31             System.out.println("获取RabbitMQ连接失败");
32          catch (TimeoutException e) 
33             System.out.println("获取RabbitMQ连接超时");
34         
35     
36 
37 //    public static void close() throws IOException, TimeoutException 
38 //        if (Objects.nonNull(channel))
39 //            channel.close();
40 //        if (Objects.nonNull(conn))
41 //            conn.close();
42 //    
43 
44     /**
45      * 向指定exchange下route key发送延时任务
46      * @param msg 延时任务JSON bytes
47      * @param exchangeName
48      * @param routingKey
49      * @param expiration 延时时间
50      */
51     public static void addTask(byte[] msg, String exchangeName, String routingKey, int expiration) 
52         try 
53             channel.basicPublish(exchangeName, routingKey,
54                     new AMQP.BasicProperties.Builder()
55                             .expiration(String.valueOf(expiration))
56                             .build(), msg);
57          catch (IOException e) 
58             e.printStackTrace();
59         
60     
61 
62     /**
63      * 消费指定queue的消息（延时任务）
64      * @param queueName
65      * @param handler 任务处理器
66      * @param consumerTag 消费者标签（多个消费者同时消息同一queue时可以使用consumerTag作区分）
67      */
68     public static void bindConsumer(String queueName, DemoTaskHandler handler, String consumerTag) 
69         try 
70             channel.basicConsume(queueName, false, consumerTag,
71                     new DefaultConsumer(channel) 
72                         @Override
73                         public void handleDelivery(String consumerTag,
74                                                    Envelope envelope,
75                                                    AMQP.BasicProperties properties,
76                                                    byte[] body)
77                                 throws IOException 
78                             long deliveryTag = envelope.getDeliveryTag();
79                             // (process the message components here ...)
80                             handler.execute(body, consumerTag);
81                             channel.basicAck(deliveryTag, false); // 应答，告知queue成功收到消息
82                         
83                     );
84          catch (IOException e) 
85             e.printStackTrace();
86         
87     
88 
89 

 1 import java.io.Serializable;
 2 
 3 public class DemoTask implements Serializable 
 4 
 5     private int id;
 6 
 7     public int getId() 
 8         return id;
 9     
10 
11     public void setId(int id) 
12         this.id = id;
13     
14 

1 import com.alibaba.fastjson.JSON;
2 
3 public class DemoTaskHandler 
4 
5     public void execute(byte[] body, String consumerTag) 
6         DemoTask task = JSON.parseObject(new String(body), DemoTask.class);
7         System.out.println(consumerTag + "收到延时任务id：" + task.getId() + " 并处理完毕");
8     
9 

 1 import com.alibaba.fastjson.JSON;
 2 
 3 import java.util.Scanner;
 4 
 5 public class Producer 
 6 
 7     public static void main(String[] args) 
 8         // 添加延时任务
 9         System.out.println("按下键盘添加延时任务");
10         Scanner sc = new Scanner(System.in);
11         int i = 1;
12         while (sc.hasNextLine()) 
13             sc.nextLine();
14             DemoTask bo = new DemoTask();
15             bo.setId(i++);
16             RabbitMQUtil.addTask(JSON.toJSONString(bo).getBytes(),
17                     "ExQ1",
18                     "send",
19                     10000);
20             System.out.println("成功添加一个延时任务");
21         
22     
23 
24 

 1 public class Consumer1 
 2 
 3     public static void main(String[] args) 
 4         // 模拟分布式环境，处理到期的延时任务
 5         RabbitMQUtil.bindConsumer("q2",
 6                 new DemoTaskHandler(),
 7                 "consumer1");
 8 
 9     
10 
11 

1 public class Consumer2 
2 
3     public static void main(String[] args) 
4         // 模拟分布式环境，处理到期的延时任务
5         RabbitMQUtil.bindConsumer("q2",
6                 new DemoTaskHandler(),
7                 "consumer2");
8     
9