Java メッセージ キュー タスクのスムーズなシャットダウン

1. 問題の背景

メッセージキューの監視では、通常、Java を使用して別のプログラムを作成し、Linux サーバー上で実行します。プログラムが開始されると、メッセージはメッセージ キュー クライアントを通じて受信され、非同期処理のためにスレッド プールに入れられるため、高速な同時処理が可能になります。

それでは、プログラムを変更してタスクを再起動する必要がある場合、メッセージが失われないようにするにはどうすればよいでしょうか?

通常、サブスクライバー プログラムが終了すると、メッセージは送信者キューに蓄積され、サブスクライバーによる次のサブスクリプションの消費を待つため、未受信のメッセージが失われることはありません。失われる可能性があるメッセージは、キューから取り出されたものの、シャットダウン時にまだ処理されていないメッセージだけです。

そのため、再起動時にメッセージが正常に処理されることを保証するために、スムーズなシャットダウン メカニズムが必要です。

2. 問題の分析

スムーズなシャットダウンの考え方は次のとおりです:

プログラムを閉じるときは、まずメッセージ サブスクリプションを閉じます。この時点で、メッセージはすべて送信者キューにあります

ローカル メッセージを閉じます。スレッド プールの処理 (ローカル スレッド プールを待っています) メッセージが処理されます)

プログラムが終了します

メッセージ サブスクリプションを閉じる: 一般に、メッセージ キュー クライアントは、接続を閉じるためのメソッドを提供します。詳細については、API を自分で表示できます

。スレッド プールを閉じる: Java の ThreadPoolExecutor スレッド プールには shutdown() と shutdownNow() という 2 つのメソッドが用意されています。違いは、前者はスレッド プール内のメッセージが処理されるまで待機するのに対し、後者はスレッドの実行を直接停止し、リストコレクションを返します。シャットダウンするには shutdown() メソッドを使用し、スレッド プールが閉じられているかどうかを判断するには isTerminated() メソッドを使用する必要があるためです。

次に、シャットダウン操作が必要であることをプログラムにどのように通知するかという問題が再び発生します。 Linux では、kill -9 pid を使用してプロセスをシャットダウンできます。-9 に加えて、kill -l を使用して、12) SIGUSR2 セマフォの使用など、kill コマンドの他のセマフォを表示できます。

Javaプログラムの起動時に対応するシグナルを登録できます。セマフォが監視され、対応するkillオペレーションを受信すると、関連するビジネスオペレーションが実行されます。

疑似コードは次のとおりです

 //注册linux kill信号量  kill -12Signal sig = new Signal("USR2");
Signal.handle(sig, new SignalHandler() {    @Override
    public void handle(Signal signal) {        //关闭订阅者
        //关闭线程池
        //退出
    }
});

以下は、デモを通じて関連する論理操作をシミュレートします

最初にプロデューサーをシミュレートし、1秒あたり5つのメッセージを生成します

次にサブスクライバーをシミュレートし、メッセージを受信した後、それを処理用のスレッド プール、スレッド プールには固定数の 4 スレッドがあり、各メッセージの処理時間は 1 秒であるため、スレッド プールは 1 秒あたり 1 つのメッセージをバックログします。

package com.lujianing.demo;import sun.misc.Signal;import sun.misc.SignalHandler;import java.util.concurrent.*;/**
 * @author lujianing01@58.com
 * @Description:
 * @date 2016/11/14
 */public class MsgClient {    //模拟消息队列订阅者 同时4个线程处理
    private static final ThreadPoolExecutor THREAD_POOL = (ThreadPoolExecutor) Executors.newFixedThreadPool(4);    //模拟消息队列生产者
    private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();    //用于判断是否关闭订阅
    private static volatile boolean isClose = false;    public static void main(String[] args) throws InterruptedException {
        BlockingQueue <String> queue = new ArrayBlockingQueue<String>(100);
        producer(queue);
        consumer(queue);
    }    //模拟消息队列生产者
    private static void producer(final BlockingQueue  queue){        //每200毫秒向队列中放入一个消息
        SCHEDULED_EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {            public void run() {
                queue.offer("");
            }
        }, 0L, 200L, TimeUnit.MILLISECONDS);
    }    //模拟消息队列消费者 生产者每秒生产5个   消费者4个线程消费1个1秒  每秒积压1个
    private static void consumer(final BlockingQueue queue) throws InterruptedException {        while (!isClose){
            getPoolBacklogSize();            //从队列中拿到消息
            final String msg = (String)queue.take();            //放入线程池处理
            if(!THREAD_POOL.isShutdown()) {
                THREAD_POOL.execute(new Runnable() {                    public void run() {                        try {                            //System.out.println(msg);
                            TimeUnit.MILLISECONDS.sleep(1000L);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
    }    //查看线程池堆积消息个数
    private static long getPoolBacklogSize(){        long backlog = THREAD_POOL.getTaskCount()- THREAD_POOL.getCompletedTaskCount();
        System.out.println(String.format("[%s]THREAD_POOL backlog:%s",System.currentTimeMillis(),backlog));        return backlog;
    }    static {
        String osName = System.getProperty("os.name").toLowerCase();        if(osName != null && osName.indexOf("window") == -1) {            //注册linux kill信号量  kill -12
            Signal sig = new Signal("USR2");
            Signal.handle(sig, new SignalHandler() {                @Override
                public void handle(Signal signal) {
                    System.out.println("收到kill消息，执行关闭操作");                    //关闭订阅消费
                    isClose = true;                    //关闭线程池，等待线程池积压消息处理
                    THREAD_POOL.shutdown();                    //判断线程池是否关闭
                    while (!THREAD_POOL.isTerminated()) {                        try {                            //每200毫秒 判断线程池积压数量
                            getPoolBacklogSize();
                            TimeUnit.MILLISECONDS.sleep(200L);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println("订阅者关闭，线程池处理完毕");
                    System.exit(0);
                }
            });
        }
    }
}

サービスを実行すると、コンソールを通じて関連する出力情報を確認できます。デモでは、スレッド プール内のバックログ メッセージの数が出力されます

java -cp /home/work/lujianing/msg-queue-client/* com.lujianing.demo.MsgClient

別のターミナルを開いて、ps を通じてプロセス番号を確認します。コマンドを実行するか、nohup を通じて Java プロセスを開始してプロセス ID を取得します

ps -fe|grep MsgClient

kill -12 pid を実行すると、シャットダウンのビジネス ロジックが表示されます

3. 実際のビジネスでは部門の担当者は、メッセージ キューのメッセージ量は依然として非常に多く、ビジネスのピーク時には 1 秒あたり数百のメッセージが発生するため、単一のサブスクリプション ノードへの負荷を回避するために、メッセージの処理速度を確保する必要があります。負荷によっても解決できます。

一部のビジネス シナリオでは、メッセージの整合性の要件はそれほど高くないため、再起動時の損失を考慮する必要はありません。逆に、慎重な思考と設計が必要です。