问服务器中的死锁监视狗，以防止编写不当的扩展。

EN

ReceivedMessageTask task = new ReceivedMessageTask(sessionId, message, handler, this);
task.setMaxHandlingTimeout(maxHandlingTimeout);
ListenableFuture<Boolean> future = (ListenableFuture<Boolean>) executor.submitListenable(new ListenableFutureTask<Boolean>(task));

public final class ReceivedMessageTask implements Callable<Boolean> {

    private final static Logger log = LoggerFactory.getLogger(ReceivedMessageTask.class);

    private final RTMPConnection conn;

    private final IRTMPHandler handler;

    private final String sessionId;

    private Packet message;

    private AtomicBoolean done = new AtomicBoolean(false);

    private DeadlockGuard guard;

    private long maxHandlingTime = 500L;

    public ReceivedMessageTask(String sessionId, Packet message, IRTMPHandler handler) {
        this(sessionId, message, handler, (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId));
    }

    public ReceivedMessageTask(String sessionId, Packet message, IRTMPHandler handler, RTMPConnection conn) {
        this.sessionId = sessionId;
        this.message = message;
        this.handler = handler;
        this.conn = conn;
    }   

    public Boolean call() throws Exception {
        Red5.setConnectionLocal(conn);
        try {
            if (maxHandlingTime <= 0) {
                if (!Red5.isDebug()) {
                    guard = new DeadlockGuard();
                    new Thread(guard, String.format("DeadlockGuard#%s", sessionId)).start();
                }
            }
            handler.messageReceived(conn, message);
        } catch (Exception e) {
            log.error("Error processing received message {} on {}", message, sessionId, e);
        } finally {
            Red5.setConnectionLocal(null);
            done.set(true);
            if (guard != null) {
                guard.join();
            }
        }
        return Boolean.valueOf(done.get());
    }

    public void setMaxHandlingTimeout(long maxHandlingTimeout) {
        this.maxHandlingTime = maxHandlingTimeout;
    }

    private class DeadlockGuard implements Runnable {

        private Thread taskThread;

        private Thread guardThread = null;

        AtomicBoolean sleeping = new AtomicBoolean(false);

        DeadlockGuard() {
            this.taskThread = Thread.currentThread();
        }

        public void run() {
            try {
                this.guardThread = Thread.currentThread();
                if (log.isDebugEnabled()) {
                    log.debug("Threads - task: {} guard: {}", taskThread.getName(), guardThread.getName());
                }
                sleeping.compareAndSet(false, true);
                Thread.sleep(maxHandlingTime);
            } catch (InterruptedException e) {
                log.debug("Deadlock guard interrupted on {} during sleep", sessionId);  
            } finally {
                sleeping.set(false);
            }
            if (!done.get()) {
                if (!taskThread.isInterrupted()) {
                    if (taskThread.isAlive()) {
                        log.warn("Interrupting unfinished active task on {}", sessionId);
                        taskThread.interrupt();
                    }               
                } else {
                    log.debug("Unfinished active task on {} already interrupted", sessionId);                   
                }
            }
        }

        public void join() {
            if (sleeping.get()) {
                guardThread.interrupt();
            }
        }
    }
}