Observer - BlockingQueue

Question

我使用觀察者模式和BlockingQueue來添加一些實例。現在，在另一種方法我用了隊列，但它似乎採取（）永遠等待，即使我做這件事是這樣的：

/** {@inheritDoc} */ 
@Override 
public void diffListener(final EDiff paramDiff, final IStructuralItem paramNewNode, 
    final IStructuralItem paramOldNode, final DiffDepth paramDepth) { 
    final Diff diff = 
     new Diff(paramDiff, paramNewNode.getNodeKey(), paramOldNode.getNodeKey(), paramDepth); 
    mDiffs.add(diff); 
    try { 
     mDiffQueue.put(diff); 
    } catch (final InterruptedException e) { 
     LOGWRAPPER.error(e.getMessage(), e); 
    } 
    mEntries++; 

    if (mEntries == AFTER_COUNT_DIFFS) { 
     try { 
      mRunner.run(new PopulateDatabase(mDiffDatabase, mDiffs)); 
     } catch (final Exception e) { 
      LOGWRAPPER.error(e.getMessage(), e); 
     } 
     mEntries = 0; 
     mDiffs = new LinkedList<>(); 
    } 
} 

/** {@inheritDoc} */ 
@Override 
public void diffDone() { 
    try { 
     mRunner.run(new PopulateDatabase(mDiffDatabase, mDiffs)); 
    } catch (final Exception e) { 
     LOGWRAPPER.error(e.getMessage(), e); 
    } 
    mDone = true; 
} 

而mDiffQueue是的LinkedBlockingQueue，我使用它像這樣的：

while (!(mDiffQueue.isEmpty() && mDone) || mDiffQueue.take().getDiff() == EDiff.INSERTED) {} 

但我認爲第一個表達式進行檢查，而mDone是不正確的，那麼也許mDone設置爲true（觀察者始終是多線程的？），但它已經調用mDiffQueue.take（）？ ： -/

編輯：我真的不明白現在。我最近把它改成：

synchronized (mDiffQueue) { 
    while (!(mDiffQueue.isEmpty() && mDone)) { 
     if (mDiffQueue.take().getDiff() != EDiff.INSERTED) { 
      break; 
     } 
    } 
} 

如果我在調試器等待它的工作原理有點時間，但也應該在「實時」的工作，因爲mDone被初始化爲假，並因此而條件應是真的，身體應該被執行。

如果mDiffQueue爲空且mDone爲true，它應該跳過while循環的主體（這意味着隊列不再被填充）。

編輯：看來，這是：

synchronized (mDiffQueue) { 
    while (!(mDiffQueue.isEmpty() && mDone)) { 
     if (mDiffQueue.peek() != null) { 
      if (mDiffQueue.take().getDiff() != EDiff.INSERTED) { 
       break; 
      } 
     } 
    } 
} 

即使我不明白爲什麼偷看（）是強制性的。

編輯：

我想跳過所有插入的節點是什麼，我做的是遍歷樹和：

for (final AbsAxis axis = new DescendantAxis(paramRtx, true); axis.hasNext(); axis.next()) { 
    skipInserts(); 
    final IStructuralItem node = paramRtx.getStructuralNode(); 
    if (node.hasFirstChild()) { 
     depth++; 
     skipInserts(); 
     ... 

基本上計算樹的最大深度或水平，而不考慮其中的節點已經在樹的另一個版本中被刪除（用於比較Sunburst可視化），但是好的，這可能超出範圍。只是爲了說明我正在做一些尚未插入的節點，即使它只是調整最大深度。

問候，

約翰內斯

Answer 1

首先建議：不要synchronized (mDiffQueue)。如果LinkedBlockingQueue有一些​​方法，則會發生死鎖;這裏並非如此，但這是一種你應該避免的做法。無論如何，我不明白爲什麼你在這個時候同步。

你要「喚醒」定期在等待檢查mDone已設置：

while (!(mDiffQueue.isEmpty() && mDone)) { 
    // poll returns null if nothing is added in the queue for 0.1 second. 
    Diff diff = mDiffQueue.poll(0.1, TimeUnit.SECONDS); 
    if (diff != null) 
     process(diff); 
} 

這是關於與使用peek，但peek基本上等待納秒代替。使用peek被稱爲「忙等待」（您的線程不停地運行while循環），並使用pool被稱爲「半忙等待」（讓線程休眠）。

我想你的情況下process(diff)將脫離循環，如果diff不是類型EDiff.INSERTED。我不確定這是否是你想要完成的。這看起來很奇怪，因爲你基本上只是拖延消費者線程，直到你得到一個正確類型的單個元素，然後你什麼都不做。由於您不在while循環中，因此無法接收未來的傳入元素。

Answer 2

take()是一個 「堵呼叫」。這意味着它會阻止（等到永遠），直到隊列中有東西，然後它會返回添加的內容。當然，如果隊列中有東西，它會立即返回。

您可以使用peek()返回什麼將由take()返回 - 也就是說，peek()返回的下一個項目不從隊列中刪除，或返回null如果沒有什麼隊列。嘗試使用peek()代替您的測試（但也檢查null）。