Java 메시지 대기열 작업의 원활한 종료

1. 문제 배경

메시지 큐 모니터링을 위해 일반적으로 Java를 사용하여 독립적인 프로그램을 작성하고 Linux 서버에서 실행합니다. 프로그램이 시작된 후 메시지는 메시지 큐 클라이언트를 통해 수신되고 비동기 처리를 위해 스레드 풀에 배치되므로 빠른 동시 처리가 가능합니다.

그러면 문제는 프로그램을 수정하고 작업을 다시 시작해야 할 때 메시지가 손실되지 않도록 하려면 어떻게 해야 할까요?

일반적으로 구독자 프로그램이 종료된 후 메시지는 발신자 대기열에 누적되어 구독자의 다음 구독 소비를 기다리므로 받지 못한 메시지는 손실되지 않습니다. 손실될 수 있는 유일한 메시지는 대기열에서 제외되었지만 종료 시 아직 처리되지 않은 메시지입니다.

따라서 재시작 시 메시지가 정상적으로 처리될 수 있도록 원활한 종료 메커니즘이 필요합니다.

2. 문제 분석

원활한 종료 아이디어는 다음과 같습니다.

프로그램 종료 시 먼저 메시지 구독을 종료합니다. 모두 송신자 대기열에 있습니다

로컬 메시지 처리 스레드 풀을 닫습니다(로컬 스레드 풀의 메시지가 처리될 때까지 기다립니다)

프로그램이 종료됩니다

메시지 닫기 구독: 일반적으로 메시지 대기열 클라이언트는 연결을 닫는 메서드를 제공합니다. 자세한 내용은

API를 확인하여 스레드 풀을 닫을 수 있습니다. Java의 ThreadPoolExecutor 스레드 풀은 shutdown() 및 shutdownNow()라는 두 가지 메서드를 제공합니다. 차이점은 전자는 스레드 풀의 메시지가 처리될 때까지 기다리는 반면, 후자는 직접 스레드 실행을 중지하고 목록 컬렉션을 반환한다는 것입니다. 종료하려면 shutdown() 메서드를 사용해야 하고, 스레드 풀이 닫혔는지 확인하려면 isTerminating() 메서드를 사용해야 하기 때문입니다.

그러면 다시 질문이 뜹니다. 프로그램에 이를 어떻게 알릴 수 있을까요? 종료 작업을 수행해야 합니까?

Linux에서는 kill -9 pid를 사용하여 프로세스를 종료할 수 있으며, -9 외에도 kill -l을 사용하여 kill 명령의 다른 세마포어를 볼 수 있습니다. , 예를 들어 12) SIGUSR2 세마포어

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초입니다. 초당 메시지.

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

-12 pid를 실행하면 종료 비즈니스 로직을 볼 수 있습니다

3. 문제 요약

부서의 실제 비즈니스에서는 메시지 큐의 메시지 볼륨은 여전히 ​​상당히 큽니다. 일부 비즈니스 피크에서는 초당 수백 개의 메시지가 있으므로 메시지 백로그를 방지하려면 메시지 처리 속도를 보장해야 합니다. 로드를 통해.

일부 비즈니스 시나리오에서는 메시지 무결성에 대한 요구 사항이 그다지 높지 않으므로 다시 시작하는 동안 손실을 고려할 필요가 없습니다. 오히려 신중한 사고와 디자인이 필요합니다.