从 ActiveMQ DestinationSource.getQueues 响应中正确迭代队列

Question

由于某些原因，在以下代码中，destinationSource.getQueues() 返回一个 CopyOnWriteArraySet 而不是简单的 Set。这是一个问题，因为 for 循环在 Set 已满之前开始处理，并且由于 CopyOnWriteArraySet 的性质，它只会在循环之前处理 Set 中的项目。我知道我可以在那里扔一个Thread.sleep()，但这并不能解决根本问题。是否有任何理由将其作为CopyOnWriteArraySet 而不是Set 返回？还有什么方法可以迭代 CopyOnWriteArraySet 以确保覆盖所有项目，即使是在迭代期间添加的项目？

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
ActiveMQConnection activeMQConnection = (ActiveMQConnection) connectionFactory.createConnection();
activeMQConnection.start();
DestinationSource destinationSource = activeMQConnection.getDestinationSource();

Set<ActiveMQQueue> queues = destinationSource.getQueues();

for(ActiveMQQueue queue : queues) {
  queueNames.add(queue.getPhysicalName());
}

activeMQConnection.close()

编辑：这是我想出的解决方案，虽然它并不完美，但它确保您将获得所有队列，直到添加队列之间的时间超过 1 秒。

    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

    ActiveMQConnection activeMQConnection = (ActiveMQConnection) connectionFactory.createConnection();

    activeMQConnection.start();

    DestinationSource destinationSource = activeMQConnection.getDestinationSource();

    Set<ActiveMQQueue> queues = destinationSource.getQueues();

    do {
        for(ActiveMQQueue queue : queues) {
            String physcialName = queue.getPhysicalName();
            if(!queueNames.contains(physcialName)) {
                queueNames.add(physcialName);
            }
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log(e.toString());
        }
    }while(queueNames.size() < queues.size());

    activeMQConnection.close();

Answer 1

我在从连接中获取所有队列时遇到了同样的问题。 每当我从 DestinationSource 获得队列，然后在此集合上进行迭代（foreach）时， 我得到了不同数量的队列（在迭代循环中，我总是得到比初始集合更多的队列）。

DestinationSource ds = connection.getDestinationSource();
Set<ActiveMQQueue> queues = ds.getQueues();
log.debug("Found '" + queues.size() + "' queues");
for (ActiveMQQueue queue : queues) {...}

然后，我像这样向目标源添加了一个侦听器

DestinationSource ds = connection.getDestinationSource();
Set<ActiveMQQueue> queues = ds.getQueues();
// Add listener:
ds.setDestinationListener(event -> event.hashCode());
log.debug("Found '" + queues.size() + "' queues");
for (ActiveMQQueue queue : queues) {...}

从现在开始，我总是得到正确数量的队列，并且可以遍历整个集合。

虽然，我真的不知道为什么;）

Answer 2

正如你所说，CopyOnWriteArraySet 的本性就是这样。队列列表可以与您的线程同时更改。通过返回一个CopyOnWriteArraySet，ActiveMQ 为您提供了一个可以在您的线程中安全使用的数据结构（ConcurrentModificationException 没有变化）并且将保持最新。

由于可以随时添加新队列，因此无法“等待”直到它们全部完成。

如果您想知道何时添加了新队列，然后做一些响应，最好的方法是监听相应的 ActiveMQ advisory message。该工具将让您响应消息队列的添加、消费者和生产者的添加以及它们的删除。认为链接有一个代码示例。