基於RabbitMQ實現分布式延時任務調度

一.分布式延時任務

傳統做法是將延時任務插入數據庫，使用定時去掃描，比對任務是否到期，到期則執行並設置任務狀態為完成。這種做法在分布式環境下還需要對定時掃描做特殊處理（加分布式鎖）避免任務被重復執行。

然而使用RabbitMQ實現延時任務可以天然解決分布式環境下重復執行的問題（利用mq中消息只會被一個消費者消費這一特性可以讓延時任務只會被一個消費者執行）。基於RabbitMQ做延時任務的核心是利用RabbitMQ的消息到期轉發特性。發送消息時設置消息到期時間，等消息到期未被消費時會將消息轉發到一個新的隊列，新隊列的消費者收到消息后再處理，利用這種時間差特性實現任務的延時觸發。

二.准備RabbitMQ並設置延時任務用到的相關隊列

1.安裝erlang和RabbitMQ(注意erlang與RabbitMQ的版本對應關系)

2.開啟rabbitmq_management

打開RabbitMQ Command Prompt輸入命令：rabbitmq-plugins enable rabbitmq_management

3.創建兩個Exchange

創建一個Exchange用於添加延時任務，相關配置如下

 

 再創建一個Exchange用於接收到期的延時任務，相關配置如下

 

4.創建兩個Queue

創建第一個Queue，用於添加延時任務，相關配置如下

 

上面配置創建了一個隊列q1,設置到期消息被轉移的目的地Exchange(dlx)和Route key(dlx_rk)

接下來配置q1綁定的Exchange為ExQ1，Route key為send

 

 

 再創建第二個Queue，用於接收隊列q1中到期被轉移的任務，相關配置如下

 

並綁定到Exchange：dlx，Route key：dlx_rk

 

通過上面兩個Exchange和兩個Queue的配置，讓RabbitMQ支持q1中的消息到期后轉移到q2中。所以業務上我們只用將延時任務發送到q1，讓任務到期觸發執行的業務代碼去監聽（消費）q2。這樣基本上就實現了分布式環境下延時任務的創建以及到期調度觸發執行。

三.具體代碼實現

1.創建簡單maven項目，添加如下依賴

    <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>

2.封裝用到的RabbitMQ操作

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 1.連接RabbitMQ
 * 2.添加延時任務
 * 3.消費延時任務
 */
public class RabbitMQUtil {

    private static Connection conn;
    private static Channel channel;

    /**
     * 初始化RabbitMQ連接與channel
     */
    static {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        factory.setHost("localhost");
        factory.setPort(5672);

        try {
            conn = factory.newConnection();
            channel = conn.createChannel();
        } catch (IOException e) {
            System.out.println("獲取RabbitMQ連接失敗");
        } catch (TimeoutException e) {
            System.out.println("獲取RabbitMQ連接超時");
        }
    }

//    public static void close() throws IOException, TimeoutException {
//        if (Objects.nonNull(channel))
//            channel.close();
//        if (Objects.nonNull(conn))
//            conn.close();
//    }

    /**
     * 向指定exchange下route key發送延時任務
     * @param msg 延時任務JSON bytes
     * @param exchangeName
     * @param routingKey
     * @param expiration 延時時間
     */
    public static void addTask(byte[] msg, String exchangeName, String routingKey, int expiration) {
        try {
            channel.basicPublish(exchangeName, routingKey,
                    new AMQP.BasicProperties.Builder()
                            .expiration(String.valueOf(expiration))
                            .build(), msg);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 消費指定queue的消息（延時任務）
     * @param queueName
     * @param handler 任務處理器
     * @param consumerTag 消費者標簽（多個消費者同時消息同一queue時可以使用consumerTag作區分）
     */
    public static void bindConsumer(String queueName, DemoTaskHandler handler, String consumerTag) {
        try {
            channel.basicConsume(queueName, false, consumerTag,
                    new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag,
                                                   Envelope envelope,
                                                   AMQP.BasicProperties properties,
                                                   byte[] body)
                                throws IOException {
                            long deliveryTag = envelope.getDeliveryTag();
                            // (process the message components here ...)
                            handler.execute(body, consumerTag);
                            channel.basicAck(deliveryTag, false); // 應答，告知queue成功收到消息
                        }
                    });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

3.模擬延時任務POJO

import java.io.Serializable;

public class DemoTask implements Serializable {

    private int id;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }
}

4.延時任務處理器

import com.alibaba.fastjson.JSON;

public class DemoTaskHandler {

    public void execute(byte[] body, String consumerTag) {
        DemoTask task = JSON.parseObject(new String(body), DemoTask.class);
        System.out.println(consumerTag + "收到延時任務id：" + task.getId() + " 並處理完畢");
    }
}

5.設計一個主程序往q1隊列發送延時任務

import com.alibaba.fastjson.JSON;

import java.util.Scanner;

public class Producer {

    public static void main(String[] args) {
        // 添加延時任務
        System.out.println("按下鍵盤添加延時任務");
        Scanner sc = new Scanner(System.in);
        int i = 1;
        while (sc.hasNextLine()) {
            sc.nextLine();
            DemoTask bo = new DemoTask();
            bo.setId(i++);
            RabbitMQUtil.addTask(JSON.toJSONString(bo).getBytes(),
                    "ExQ1",
                    "send",
                    10000);
            System.out.println("成功添加一個延時任務");
        }
    }

}

6.創建兩個消費者（處理延時任務的業務）消費延時任務，模擬分布式環境

public class Consumer1 {

    public static void main(String[] args) {
        // 模擬分布式環境，處理到期的延時任務
        RabbitMQUtil.bindConsumer("q2",
                new DemoTaskHandler(),
                "consumer1");

    }

}

public class Consumer2 {

    public static void main(String[] args) {
        // 模擬分布式環境，處理到期的延時任務
        RabbitMQUtil.bindConsumer("q2",
                new DemoTaskHandler(),
                "consumer2");
    }
}

7.運行Producer，Consumer1，Consumer2觀察結果

 

 

 

 

 

 通過觀察發現，每次發送一個延時任務后，過10秒會被consumer1或者consumer2消費，以上就基本實現了分布式延時任務調度。