java - 是否可以 peek() 查看 Java ConcurrentLinkedQueue 的元素

Question

我有一个程序，其中有多个 PublisherTask 和 SubscriberTask 类型的对象。给定订阅者可以订阅一个或多个发布者。
为了描述我的问题，我将发布一些代码......

abstract class Publication {
    // some published information
}

class ConcretePublicationA extends Publication {

}

class ConcretePublicationB extends Publication {

}

abstract class Subscription {
    private final long id;
    private final Subscriber s;
    // PLUS some other members relating to the subscription

    protected Subscription(long id, Subscriber s){
        this.id = id;
        this.s =s;
    }

    public Subscriber getSubscriber() {
        return this.s;
    }


}

class ConcreteSubscriptionA extends Subscription {

    protected ConcreteSubscriptionA(long id, Subscriber s) {
        super(id, s);
        // TODO Auto-generated constructor stub
    }

}

class ConcreteSubscriptionB extends Subscription {

    protected ConcreteSubscriptionB(long id, Subscriber s) {
        super(id, s);
        // TODO Auto-generated constructor stub
    }

}

interface Subscriber {
    public void update(Publication pub);
}

interface Publisher {
    public Subscription subscribe(Subscriber subscriber);
}

abstract class PublisherTask implements Runnable, Publisher {
    private final ConcurrentHashMap<Long, Subscription> subscribers =
        new ConcurrentHashMap<Long, Subscription>();
    Long subscriptionId = 0L;

    @Override
    public void run() {
        /*obviously this is a different variable in a real program*/
        boolean some_condition = true;

        while(some_condition) {
            // do some work
            Publication pub = /* new ConcretePublication(....) */ null;

            for (Subscription s : subscribers.values()) {
                s.getSubscriber().update(pub);
            }
        }

    }

    @Override
    public Subscription subscribe(Subscriber subscriber) {
        Subscription sub;

        synchronized(subscriptionId) {
            /* the lines below are in a function in the sub-class,
             *  but for brevity I'm showing them here
             */
            sub = new ConcreteSubscriptionA(++subscriptionId, subscriber);
                    subscribers.put(subscriptionId, sub);
        }
        return sub ;
    }

}


abstract class SubscriberTask implements Runnable, Subscriber {

    protected ConcurrentLinkedQueue<Publication> newPublications =
        new ConcurrentLinkedQueue<Publication>();

    @Override
    public void run() {
        /*obviously this is a different variable in a real program*/
        boolean some_condition = true;

        while(some_condition) {
            // do some work
            Publication pub = newPublications.peek();

        /* the lines below are in a function in the sub-class,
         *  but for brevity I'm showing them here
         */
        {
            if (pub instanceof ConcretePublicationA) {
                // Do something with the published data
            } else if (pub instanceof ConcretePublicationB) {
                // Do something with the published data
            }
        }
        }
    }

    @Override
    public void update(Publication pub) {

        /* My question relates to this method:
             * Bascially to avoid memory issues I would like existing
             * unprocessed publications **Of Tth Same Type As The New One**
             * to be discarded
             */
        Publication existing = null;

        do {
            //This won't work coz peek() only looks at the head of the queue
                    existing = newPublications.peek();

            if ((existing != null) && (existing.getClass().equals(pub))) {
                newPublications.remove(existing);
            }
        } while (existing != null);
        newPublications.add(pub);
    }

好的，现在您已经有机会仔细检查我的代码了。我想请教以下问题:
在上面显示的 update 方法中，是否可以查看 ConcurrentLinkedQueue 中的所有元素并删除给定类型的元素？
另外，如果您认为可以对类(class)以及它们之间的交互方式进行改进，请随时告诉我。
谢谢

Answer 1

是的，可以使用迭代器查看 ConcurrentLinkedQueue 的所有元素。

Iterator<Publication> itr = newPublications.iterator();
while (itr.hasNext()) {
   Publication existing = itr.next();
   if (existing.getClass().equals(pub)) {
      itr.remove();
   }
}

因为 ConcurrentLinkedQueue 返回的迭代器保证遍历迭代器构造时存在的元素，并且可能(但不保证)反射(reflect)构造后的任何修改，因此您可能需要外部锁定:

newPublications

总的来说，虽然这似乎不是很有效，所以应该研究一种完全删除重复出版物的替代解决方案。