java - Java中使用等待/通知机制终止生产者-消费者线程

Question

我正在维护的应用程序(经过许多编码员)使用等待/通知机制实现了生产者-消费者问题。

使用者在应用程序的“服务器”端等待消息，然后将“客户端”端的消息转发到 LDAP 服务器。

问题在于建立/终止多个连接时。生产者线程只是不断地增加，并且永远不会在应该终止的时候终止。

当连接终止时，生产者/消费者线程也应该终止。由于建立/终止的连接数量较多，内存使用量变得巨大。

代码:

class Producer extends Thread {
    public void run() {
        long previous = 0;
        long last = 0;
        long sleeptime = 1;

        while (alive) {
            try{
                last = System.currentTimeMillis();

                byte[] aux;
                if ((aux = cliente.readmessage()) != null){

                    sleeptime = 1;
                    previous = last;

                    synchronized (list) {
                        while (list.size() == MAX)
                            try {
                                list.wait();
                            } catch (InterruptedException ex) {
                            }
                        list.addFirst(new Messagetimestamped(aux, System
                                .currentTimeMillis()));
                        list.notifyAll();
                    }
                }
                else{
                    if (last-previous > 1000)
                        sleeptime = 1000;
                    else
                        sleeptime = 1;
                    sleep(sleeptime);
                }
            }
            catch (Exception e){
                if (lives()){
                    System.out.println("++++++++++++++++++ Basic Process - Producer");
                    kill();
                    nf.notify(false, processnumber);
                }
                return;
            }
        }
    }
}


class Consumer extends Thread{

    public void run() {
        while (alive) {
            byte[] message = null;
            Messagetimestamped  mt;
            synchronized(list) {
                while (list.size() == 0) {
                    try {
                        list.wait(); //HANGS HERE!
                        if (!alive) return;
                        sleep(1);
                    }
                    catch (InterruptedException ex) {}
                }
                mt = list.removeLast();
                list.notifyAll();
            }
            message = mt.mensaje;

            try{
                long timewaited = System.currentTimeMillis()-mt.timestamp;

                if (timewaited < SLEEPTIME)
                    sleep (SLEEPTIME-timewaited);

                if ( s.isClosed() || s.isOutputShutdown() ){
                    System.out.println("++++++++++++++++++++ Basic Process - Consumer - Connection closed!(HLR)");
                    kill();
                    nf.notify(false, processnumber);
                }
                else {
                    br.write(message);
                    br.flush();
                }
            } catch(SocketException e){
                return;
            } catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

基本上，在alive设置为false之后，生产者实际上被终止了。消费者则不然。它只是卡在 list.wait() 行上。显然，来自 Producer 的 list.notify() (或 list.notifyAll()？)在终止后不会传递，因此 Consumer 永远无法检查 alive boolean 值。

如何通过尽可能少的修改来解决这个问题？

谢谢。

Answer 1

我只会使用 ExecutorService 来包装队列，管理您的线程并为您处理关闭。如果这样做，几乎所有代码都会消失。

但是为了回答你的问题，我建议发送一颗毒丸。一个特殊的对象，消费者收到它后将关闭它。