RxJava解决方案，以防止堵塞在多线程环境

Question

任务：

我有一组唯一值的 - ["A","B","C","D"] 多个线程试图从中删除值。 当它变为空时，代码应该通知一些监听者。

在Java中，我每次移除元素时都需要阻塞线程，并检查集合是否为空，以防止一次发送2个相同的事件。

那么我可以用RxJava以某种方式实现它而不阻止执行？

P.S.在真正的节目中有很多这样的组合。

Answer 1

你的问题有点含糊，但听起来像你想要的东西是这样的：

final class RxSet { 
    final Set<String> elements = new HashSet<String>(); 

    final AsyncSubject<Object> onEmpty = AsyncSubject.create(); 

    public RxSet(String... initialValues) { 
     elements.addAll(Arrays.asList(initialValues)); 
    } 

    public Observable<Object> onEmpty() { 
     return onEmpty; 
    } 

    public boolean take(String item) { 
     boolean wasEmpty = false; 
     boolean isEmpty = false; 
     boolean success = false; 

     synchronized (this) { 
      wasEmpty = set.isEmpty(); 
      success = set.remove(item); 
      isEmpty = set.isEmpty(); 
     } 

     if (!wasEmpty && isEmpty) { 
      onEmpty.onNext("became empty"); 
      onEmpty.onComplete(); 
     } 

     return success; 
    } 
}