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

Question

我维护的应用程序(通过许多编码器)使用等待/通知机制实现了Producer-Consumer problematic。

使用者在应用程序的“服务器”端等待消息，然后在“客户端”端将消息转发到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之后，Producer实际上会被终止。消费者不需要，它只是停留在list.wait()系列上。显然，list.notify() (或list.notifyAll()?)Producer在终止后不会交付，因此消费者永远不会检查alive布尔值。

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

谢谢。

Answer 1

我只会使用一个ExecutorService，它会包装队列，管理你的线程，并为你处理关机。如果你这样做了，几乎所有的代码都会消失。

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