public class AlarmMgr {
    private final BlockingQueue<AlarmInfo> alarms = new LinkedBlockingQueue<AlarmInfo>();
    //告警系统客户端API
    private final AlarmAgent alarmAgent = new AlarmAgent();
    //告警发送线程
    private final AbstractTerminatableThread alarmSendingThread;

    private boolean shutdownRequested = false;

    private static final AlarmMgr INSTANCE = new AlarmMgr();

    private AlarmMgr() {
        alarmSendingThread = new AbstractTerminatableThread() {
            @Override
            protected void doRun() throws Exception {
                if (alarmAgent.waitUntilConnected()) {
                    AlarmInfo alarm;
                    alarm = alarms.take();
                    terminationToken.reservations.decrementAndGet();
                    try {
                        alarmAgent.sendAlarm(alarm);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

            @Override
            protected void doCleanup(Exception exp) {
                if (null != exp) {
                    exp.printStackTrace();
                }
                alarmAgent.disconnect();
            }

        };

        alarmAgent.init();
    }

    public static AlarmMgr getInstance() {
        return INSTANCE;
    }

    public void sendAlarm(AlarmType type, String id, String extraInfo) {
        final TerminationToken terminationToken = alarmSendingThread.terminationToken;
        if (terminationToken.isToShutdown()) {
            // log the alarm
            System.err.println("rejected alarm:" + id + "," + extraInfo);
            return;

        }
        try {
            AlarmInfo alarm = new AlarmInfo(id, type);
            alarm.setExtraInfo(extraInfo);
            terminationToken.reservations.incrementAndGet();
            alarms.add(alarm);
        } catch (Throwable t) {
            t.printStackTrace();
        }
    }

    public void init() {
        alarmSendingThread.start();
    }

    public synchronized void shutdown() {
        if (shutdownRequested) {
            throw new IllegalStateException("shutdown already requested!");
        }
        
        alarmSendingThread.terminate();
        shutdownRequested = true;
    }

    public int pendingAlarms() {
        return alarmSendingThread.terminationToken.reservations.get();
    }

}

class AlarmAgent {
    // 省略其它代码
    private volatile boolean connectedToServer = false;

    public void sendAlarm(AlarmInfo alarm) throws Exception {
        // 省略其它代码
        System.out.println("Sending " + alarm);
        try {
            Thread.sleep(50);
        } catch (Exception e) {

        }
    }

    public void init() {
        // 省略其它代码
        connectedToServer = true;
    }

    public void disconnect() {
        // 省略其它代码
        System.out.println("disconnected from alarm server.");
    }

    public boolean waitUntilConnected() {
        // 省略其它代码
        return connectedToServer;
    }
}

public abstract class AbstractTerminatableThread extends Thread 
     implements Terminatable {
    public final TerminationToken terminationToken;

    public AbstractTerminatableThread() {
        super();
        this.terminationToken = new TerminationToken();
    }

      /**
     * 
     * @param terminationToken 线程间共享的线程终止标志实例
    */
    public AbstractTerminatableThread(TerminationToken terminationToken) {
        super();
        this.terminationToken = terminationToken;
    }

    protected abstract void doRun() throws Exception;

    protected void doCleanup(Exception cause) {}

    protected void doTerminiate() {}

    @Override
    public void run() {
        Exception ex = null;
        try {
            while (true) {
                                    /*
                 * 在执行线程的处理逻辑前先判断线程停止的标志。
                 */
                if (terminationToken.isToShutdown()
                    && terminationToken.reservations.get() <= 0) {
                    break;
                }
                doRun();
            }

        } catch (Exception e) {
            // Allow the thread to terminate in response of a interrupt invocation
            ex = e;
        } finally {
            doCleanup(ex);
        }
    }

    @Override
    public void interrupt() {
        terminate();
    }

    @Override
    public void terminate() {
        terminationToken.setToShutdown(true);
        try {
            doTerminiate();
        } finally {
            // 若无待处理的任务，则试图强制终止线程
            if (terminationToken.reservations.get() <= 0) {
                super.interrupt();
            }
        }
    }

}

public class TerminationToken {
        //使用volatile修饰，以保证无需显示锁的情况下该变量的内存可见性
    protected volatile boolean toShutdown = false;
    public final AtomicInteger reservations = new AtomicInteger(0);

    public boolean isToShutdown() {
        return toShutdown;
    }

    protected void setToShutdown(boolean toShutdown) {
        this.toShutdown = true;
    }

}

public class ProducerConsumerStop {
    class SampleConsumer<P> {
        private final BlockingQueue<P> queue = new LinkedBlockingQueue<P>();

        private AbstractTerminatableThread workThread 
                = new AbstractTerminatableThread() {
            @Override
            protected void doRun() throws Exception {
                terminationToken.reservations.decrementAndGet();
                P product = queue.take();
                // ...
                System.out.println(product);
            }

        };

        public void placeProduct(P product) {
            if (workThread.terminationToken.isToShutdown()) {
                throw new IllegalStateException("Thread shutdown");
            }
            try {
                queue.put(product);
                workThread.terminationToken.reservations.incrementAndGet();
            } catch (InterruptedException e) {

            }
        }

        public void shutdown() {
            workThread.terminate();
        }

        public void start() {
            workThread.start();
        }
    }

    public void test() {
        final SampleConsumer<String> aConsumer = new SampleConsumer<String>();

        AbstractTerminatableThread aProducer = new AbstractTerminatableThread() {
            private int i = 0;

            @Override
            protected void doRun() throws Exception {
                aConsumer.placeProduct(String.valueOf(i));
            }

            @Override
            protected void doCleanup(Exception cause) {
                // 生产者线程停止完毕后再请求停止消费者线程
                aConsumer.shutdown();
            }

        };

        aProducer.start();
        aConsumer.start();
    }
}