Java メッセージキュータスクのスムーズなシャットダウン-＆＃＆チュートリアル-php.cn

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

高洛峰

リリース： 2016-11-17 12:46:53

オリジナル

1349 人が閲覧しました

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

ログイン後にコピー

Java メッセージキュータスクのスムーズなシャットダウン別のターミナルを開いて、ps を通じてプロセス番号を確認します。コマンドを実行するか、nohup を通じて Java プロセスを開始してプロセス ID を取得します

ps -fe|grep MsgClient

ログイン後にコピー

Java メッセージキュータスクのスムーズなシャットダウン kill -12 pid を実行すると、シャットダウンのビジネスロジックが表示されます

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