为什么这个观察者在运行时不更新？答案

【问题标题】：Why does this Observer not update when it's running?为什么这个观察者在运行时不更新？
【发布时间】：2012-08-29 14:47:13
【问题描述】：

我有一个实现 Observer 和 Runnable 的类，如下所示（我知道这个例子可能很笨拙）：

public class Triage implements Observer,Runnable{
    Observable obsrvbl;
    private BlockingQueue<String> messages; 
    volatile static boolean interrupted=false;  
    double updated;

    Triage(Observable obsrvbl, BlockingQueue messages){
    this.obsrvbl=obsrvbl;
    this.messages = messages;
    obsrvbl.addObserver(this);
    }

    public void update(Observable o, Object arg){
        updated += ((Double)arg).doubleValue();
        System.out.println("updated");
    }

    public void run(){
        String msg; 
        while(!interrupted){
            msg=messages.take();
            if(msg!=null){    
                //do something with message         
            }
        }
    }
}

正在查看的队列在 Observable 调用 notifyObservers() 的同时填充。当 Queue 上没有任何内容时，在 Observer 上成功调用 update()，但如果 Queue 上有要处理的消息，则永远不会调用 update()。这是预期的行为吗？

我见过this，但这似乎是一个不同的问题。

这是 Observable - 有点做作：

public class Producer extends Observable implements Runnable {
    volatile static boolean interrupted=false;
    private BlockingQueue<String> quotes;
    Producer(BlockingQueue quotes){
        this.quotes=quotes; 
    }
    public void run(){
        String msg;     
        while(!interrupted){    
            msg=quotes.take();
            if(msg!=null){
                setChanged();
                notifyObservers(Double.valueOf(3.0));   
            }
        }
    }
}

【问题讨论】：

是的。当队列未填充时，观察者会更新，因此可观察的一面似乎没问题。
注意：if((msg=messages.peek())!=null){ msg=messages.poll(); 可以替换为msg = messages.take(); 以避免一些忙等待。
我会使用messages.take() 来等待消息，或者使用结合了队列和线程的ExecutorService。我不明白 Observable 增加了什么。（你不能用消息做）
向我们展示 Observable 的代码。你确定它打电话给notifyObservers()？
"当队列中没有任何内容时" - 哪个队列？ quotes 还是 messages？还是他们都是同一个队列？

标签： java observer-pattern

【解决方案1】：

弄清楚这有什么问题 - 我的疏忽。正如@slim 在评论中所建议的那样，两个队列是相同的，因此 Triage 的 run() 在 Producer 的 run() 之前消耗了消息，因此没有收到通知。无论如何，fwiw - 这是一个完整的工作示例：

public class Triage implements Observer,Runnable{
    Observable obsrvbl;
    private BlockingQueue<String> messages; 
    volatile static boolean interrupted=false;  
    Integer updated = 0;
    private static Random rand = new Random(47);

    Triage(Observable obsrvbl, BlockingQueue messages){
        this.obsrvbl=obsrvbl;
        this.messages = messages;
        obsrvbl.addObserver(this);
    }

    public void update(Observable o, Object arg){
        updated += ((Integer)arg);
        System.out.println("Updated: " + updated);
    }

    public void run(){
        String msg; 
        while(!interrupted){
            try{
                msg=messages.take();
                System.out.println("Run: " + msg);
            }catch(InterruptedException ie){}
        }
    }

    public static void main(String[] args){
        BlockingQueue<String> q1 = new LinkedBlockingQueue<String>();
        BlockingQueue<String> q2 = new LinkedBlockingQueue<String>();
        Producer p = new Producer(q1);
        new Thread(p).start();
        new Thread(new Triage(p,q2)).start();
        for(int i=0;i<20;i++){      
            int next = rand.nextInt(10)*500;
            System.out.println("Populating: " + next);
            q1.add((Integer.valueOf(next)).toString());
            q2.add((Integer.valueOf(next)).toString());         
        }
    }
}

class Producer extends Observable implements Runnable {
    volatile static boolean interrupted=false;
    private BlockingQueue<String> quotes;
    Producer(BlockingQueue quotes){
        this.quotes=quotes; 
    }
    public void run(){
        String msg;     
        while(!interrupted){    
            try{
                msg=quotes.take();
                if(msg!=null){
                    System.out.println("Notifying: " + msg);
                    setChanged();
                    notifyObservers(Integer.valueOf(msg));  
                }
            }catch(InterruptedException ie){}
        }
    }
}

【讨论】：