Runtime.getRuntime().addShutdownHook(new Thread() {
 public void run() {
 //關(guān)閉訂閱者
 //關(guān)閉線程池
 //退出
 }
});

 //注冊(cè)linux kill信號(hào)量 kill -12
Signal sig = new Signal("USR2");
Signal.handle(sig, new SignalHandler() {
 @Override
 public void handle(Signal signal) {
 //關(guān)閉訂閱者
 //關(guān)閉線程池
 //退出
 }
});

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 {

 //模擬消費(fèi)線程池 同時(shí)4個(gè)線程處理
 private static final ThreadPoolExecutor THREAD_POOL = (ThreadPoolExecutor) Executors.newFixedThreadPool(4);
 
 //模擬消息生產(chǎn)任務(wù)
 private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor();
 
 //用于判斷是否關(guān)閉訂閱
 private static volatile boolean isClose = false;

 public static void main(String[] args) throws InterruptedException {
 
 //注冊(cè)鉤子方法
 Runtime.getRuntime().addShutdownHook(new Thread() {
  public void run() {
  close();
  }
 });

 BlockingQueue <String> queue = new ArrayBlockingQueue<String>(100);
 producer(queue);
 consumer(queue);

 }

 //模擬消息隊(duì)列生產(chǎn)者
 private static void producer(final BlockingQueue queue){
 //每200毫秒向隊(duì)列中放入一個(gè)消息
 SCHEDULED_EXECUTOR_SERVICE.scheduleAtFixedRate(new Runnable() {
  public void run() {
  queue.offer("");
  }
 }, 0L, 200L, TimeUnit.MILLISECONDS);
 }

 //模擬消息隊(duì)列消費(fèi)者 生產(chǎn)者每秒生產(chǎn)5個(gè) 消費(fèi)者4個(gè)線程消費(fèi)1個(gè)1秒 每秒積壓1個(gè)
 private static void consumer(final BlockingQueue queue) throws InterruptedException {
 while (!isClose){
  getPoolBacklogSize();
  //從隊(duì)列中拿到消息
  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();
   }
   }
  });
  }
 }
 }

 //查看線程池堆積消息個(gè)數(shù)
 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;
 }

 private static void close(){
 System.out.println("收到kill消息，執(zhí)行關(guān)閉操作");
 //關(guān)閉訂閱消費(fèi)
 isClose = true;
 //關(guān)閉線程池，等待線程池積壓消息處理
 THREAD_POOL.shutdown();
 //判斷線程池是否關(guān)閉
 while (!THREAD_POOL.isTerminated()) {
  try {
  //每200毫秒 判斷線程池積壓數(shù)量
  getPoolBacklogSize();
  TimeUnit.MILLISECONDS.sleep(200L);
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
 }
 System.out.println("訂閱者關(guān)閉，線程池處理完畢");
 }

 static {
 String osName = System.getProperty("os.name").toLowerCase();
 if(osName != null && osName.indexOf("window") == -1) {
  //注冊(cè)linux kill信號(hào)量 kill -12
  Signal sig = new Signal("USR2");
  Signal.handle(sig, new SignalHandler() {
  @Override
  public void handle(Signal signal) {
   close();
  }
  });
 }
 }

}

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

ps -fe|grep MsgClient