java – 优雅地结束正在阻塞队列中等待的线程

每个连接都由Session类管理.此类维护2个连接线程,一个DownstreamThread和一个UpstreamThread.

UpstreamThread在客户端套接字上阻塞,并将所有传入的字符串编码为消息,以传递到另一层进行处理. DownstreamThread阻塞BlockingQueue,其中插入了客户端的消息.当队列中有消息时,下游线程将消息从队列中取出,将其转换为字符串并将其发送到客户端.在最终系统中,应用程序层将对传入的消息进行操作,并将传出的消息推送到服务器以发送到适当的客户端,但是现在我只有一个简单的应用程序,在传回消息之前会休息一秒钟作为附加时间戳的传出消息.

我遇到的问题是当客户端断开连接时,整个事情都会正常关闭.我正在争论的第一个问题是正常的断开连接,客户端让服务器知道它正在结束与QUIT命令的连接.基本的伪代码是：

while (!quitting) {
    inputString = socket.readLine () // blocks
    if (inputString != "QUIT") {
        // forward the message upstream
        server.acceptMessage (inputString);
    } else {
        // Do cleanup
        quitting = true;
        socket.close ();
    }
}

上游线程的主循环查看输入字符串.如果是QUIT,则线程设置一个标志,表示客户端已结束通信并退出循环.这导致上游线程很好地关闭.

只要未设置连接关闭标志,下游线程的主循环就会等待BlockingQueue中的消息.如果是,则下游线程也应该终止.然而,它没有,它只是坐在那里等待.它的伪代码看起来像这样：

while (!quitting) {
    outputMessage = messageQueue.take (); // blocks
    sendMessageToClient (outputMessage);
}

当我测试这个时,我注意到当客户端退出时,上游线程关闭,但下游线程没有.

经过一番搔痒之后,我意识到下游线程仍在阻塞BlockingQueue,等待永远不会到来的传入消息.上游线程不会在链上向前转发QUIT消息.

如何正常关闭下游线程？想到的第一个想法是在take()调用上设置超时.我不是太热衷于这个想法,因为无论你选择什么价值,它一定不会完全令人满意.要么它太长了,僵尸线程在那里停留了很长时间才关闭,或者它太短了,连接已经闲置了几分钟但仍然有效将会被杀死.我确实想过将QUIT消息发送到链中,但是这要求它完全往返服务器,然后是应用程序,然后再次返回服务器,最后返回到会话.这似乎也不是一个优雅的解决方案.

我确实查看了Thread.stop()的文档,但显然已经弃用了,因为它无论如何都没有正常工作,所以看起来它也不是一个真正的选项.我的另一个想法是强制在下游线程中以某种方式触发异常并让它在其最终块中清理,但这让我觉得这是一个可怕而又笨拙的想法.

我觉得两个线程都应该能够自己正常关闭,但我也怀疑如果一个线程结束它还必须发出信号通知另一个线程以更加主动的方式结束,而不是简单地为另一个线程设置一个标志来检查.由于我对Java仍然不是很有经验,所以我现在很缺乏想法.如果有人有任何建议,将不胜感激.

为了完整起见,我在下面列出了Session类的真实代码,不过我相信上面的伪代码片段涵盖了问题的相关部分.全班约250行.

import java.io.*;
import java.net.*;
import java.util.concurrent.*;
import java.util.logging.*;

/**
 * Session class
 * 
 * A session manages the individual connection between a client and the server. 
 * It accepts input from the client and sends output to the client over the 
 * provided socket.
 * 
 */
public class Session {
    private Socket              clientSocket    = null;
    private Server              server          = null;
    private Integer             sessionId       = 0;
    private DownstreamThread    downstream      = null;
    private UpstreamThread      upstream        = null;
    private boolean             sessionEnding   = false;

    /**
     * This thread handles waiting for messages from the server and sending
     * them to the client
     */
    private class DownstreamThread implements Runnable {
        private BlockingQueue<DownstreamMessage>    incomingMessages    = null;
        private OutputStreamWriter                  streamWriter        = null;
        private Session                             outer               = null;

        @Override
        public void run () {
            DownstreamMessage message;
            Thread.currentThread ().setName ("DownstreamThread_" + outer.getId ());

            try {
                // Send connect message
                this.sendMessageToClient ("Hello,you are client " + outer.getId ());

                while (!outer.sessionEnding) {
                    message = this.incomingMessages.take ();
                    this.sendMessageToClient (message.getPayload ());
                }

                // Send disconnect message
                this.sendMessageToClient ("Goodbye,client " + getId ());

            } catch (InterruptedException | IOException ex) {
                Logger.getLogger (DownstreamThread.class.getName ()).log (Level.SEVERE,ex.getMessage (),ex);
            } finally {
                this.terminate ();
            }
        }

        /**
         * Add a message to the downstream queue
         * 
         * @param message
         * @return
         * @throws InterruptedException 
         */
        public DownstreamThread acceptMessage (DownstreamMessage message) throws InterruptedException {
            if (!outer.sessionEnding) {
                this.incomingMessages.put (message);
            }

            return this;
        }

        /**
         * Send the given message to the client
         * 
         * @param message
         * @throws IOException 
         */
        private DownstreamThread sendMessageToClient (CharSequence message) throws IOException {
            OutputStreamWriter osw;
            // Output to client
            if (null != (osw = this.getStreamWriter ())) {
                osw.write ((String) message);
                osw.write ("rn");
                osw.flush ();
            }

            return this;
        }

        /**
         * Perform session cleanup
         * 
         * @return 
         */
        private DownstreamThread terminate () {
            try {
                this.streamWriter.close ();
            } catch (IOException ex) {
                Logger.getLogger (DownstreamThread.class.getName ()).log (Level.SEVERE,ex);
            }
            this.streamWriter   = null;

            return this;
        }

        /**
         * Get an output stream writer,initialize it if it's not active
         * 
         * @return A configured OutputStreamWriter object
         * @throws IOException 
         */
        private OutputStreamWriter getStreamWriter () throws IOException {
            if ((null == this.streamWriter) 
            && (!outer.sessionEnding)) {
                BufferedOutputStream os = new BufferedOutputStream (outer.clientSocket.getOutputStream ());
                this.streamWriter       = new OutputStreamWriter (os,"UTF8");
            }

            return this.streamWriter;
        }

        /**
         * 
         * @param outer 
         */
        public DownstreamThread (Session outer) {
            this.outer              = outer;
            this.incomingMessages   = new LinkedBlockingQueue ();
            System.out.println ("Class " + this.getClass () + " created");
        }
    }

    /**
     * This thread handles waiting for client input and sending it upstream
     */
    private class UpstreamThread implements Runnable {
        private Session outer   = null;

        @Override
        public void run () {
            StringBuffer    inputBuffer = new StringBuffer ();
            BufferedReader  inReader;

            Thread.currentThread ().setName ("UpstreamThread_" + outer.getId ());

            try {
                inReader    = new BufferedReader (new InputStreamReader (outer.clientSocket.getInputStream (),"UTF8"));

                while (!outer.sessionEnding) {
                    // Read whatever was in the input buffer
                    inputBuffer.delete (0,inputBuffer.length ());
                    inputBuffer.append (inReader.readLine ());
                    System.out.println ("Input message was: " + inputBuffer);

                    if (!inputBuffer.toString ().equals ("QUIT")) {
                        // Forward the message up the chain to the Server
                        outer.server.acceptMessage (new UpstreamMessage (sessionId,inputBuffer.toString ()));
                    } else {
                        // End the session
                        outer.sessionEnding = true;
                    }
                }

            } catch (IOException | InterruptedException e) {
                Logger.getLogger (Session.class.getName ()).log (Level.SEVERE,e.getMessage (),e);
            } finally {
                outer.terminate ();
                outer.server.deleteSession (outer.getId ());
            }
        }

        /**
         * Class constructor
         * 
         * The Core Java volume 1 book said that a constructor such as this 
         * should be implicitly created,but that doesn't seem to be the case!
         * 
         * @param outer 
         */
        public UpstreamThread (Session outer) {
            this.outer  = outer;
            System.out.println ("Class " + this.getClass () + " created");
        }
    }

    /**
     * Start the session threads
     */
    public void run () //throws InterruptedException 
    {
        Thread upThread     = new Thread (this.upstream);
        Thread downThread   = new Thread (this.downstream);

        upThread.start ();
        downThread.start ();
    }

    /**
     * Accept a message to send to the client
     * 
     * @param message
     * @return
     * @throws InterruptedException 
     */
    public Session acceptMessage (DownstreamMessage message) throws InterruptedException {
        this.downstream.acceptMessage (message);
        return this;
    }

    /**
     * Accept a message to send to the client
     * 
     * @param message
     * @return
     * @throws InterruptedException 
     */
    public Session acceptMessage (String message) throws InterruptedException {
        return this.acceptMessage (new DownstreamMessage (this.getId (),message));
    }

    /**
     * Terminate the client connection
     */
    private void terminate () {
        try {
            this.clientSocket.close ();
        } catch (IOException e) {
            Logger.getLogger (Session.class.getName ()).log (Level.SEVERE,e);
        }
    }

    /**
     * Get this Session's ID
     * 
     * @return The ID of this session
     */
    public Integer getId () {
        return this.sessionId;
    }

    /**
     * Session constructor
     * 
     * @param owner The Server object that owns this session
     * @param sessionId The unique ID this session will be given
     * @throws IOException 
     */
    public Session (Server owner,Socket clientSocket,Integer sessionId) throws IOException {

        this.server         = owner;
        this.clientSocket   = clientSocket;
        this.sessionId      = sessionId;
        this.upstream       = new UpstreamThread (this);
        this.downstream     = new DownstreamThread (this);

        System.out.println ("Class " + this.getClass () + " created");
        System.out.println ("Session ID is " + this.sessionId);
    }
}