Java 不使用迭代实现发布订阅

Question

大多数发布订阅实现都使用某种形式的迭代来通知订阅者，方法是使用 for/while 循环调用他们的监听器方法，或者将感兴趣的新主题放在他们每个人的阻塞队列中。

我想实现一个发布订阅，而不是“监视”某物，即位置、变量或文件或其他东西，然后从其存储位置提取新主题。我希望他们都“大约”同时收到通知。

这是因为以前的解决方案很慢，因为我有大量的订阅者并且遍历它们需要时间。

有什么想法吗？

Answer 1

你可以这样做:

public class Producer {
    public static Object publishedObject;
    public static ConcurrentLinkedDeque<Object> previousObjects;
    private int size;

    public publishObject(Object object) {
        previousObjects.offerLast(publishedObject);
        publishedObject = object;
        size++;
        if(size > 50) {
            previousObjects.removeFirst();
        }
    }
}

public class Consumer implements Runnable {
    private Object lastConsumedObject;
    public void run() {
        while(true) {
            if(Producer.publishedObject == lastConsumedObject) {
                sleep(500);
            } else {
                Iterator<Object> iterator = Producer.descendingIterator();
                Object next = iterator.next();
                Object newLastConsumedObject = next;
                do {
                    this.process(next);
                } while(iterator.hasNext() && 
                    (next = iterator.next()) != lastConsumedObject);
                lastConsumedObject = newLastConsumedObject;   
            }
        }
    }
}

Producer 有一个Consumers 可以轮询的“发布位置”，它还将以前发布的对象存储在LinkedList 以防 Consumer 没有注意到。消费者轮询发布位置并在没有变化时 hibernate ，否则它会处理已发布的对象并遍历先前发布的对象以防错过发布事件。

半忙等待可能会扼杀你的表现。另一种方法是拥有两类消费者:SinkConsumer 和 RelayConsumer。每个硬件线程将有一个 RelayConsumer(例如，如果您有四个内核和超线程，则有 8 个 RelayConsumer)。 Publisher 将消息发布到 RelayConsumers，然后每个 RelayConsumers 将消息发布到 numberOfSinkConsumers/numberOfRelayConsumers SinkConsumers。