0

我想就我在 ActiveMQ 中的一个主题征求意见。

我正在使用 ActiveMQ 5.15。我正在尝试修改 xml 配置文件以使用转发到另一个队列/主题的 CompositeQueues 添加虚拟目标。在此组件的 ActiveMQ 文档中，架构如下：

<compositeQueue name="IncomingOrders"> 
   <forwardTo>
    <topic physicalName="Notifications" />
   </forwardTo>
 </compositeQueue>

我已经能够转发来自现有队列的消息，例如request.typeA.classC。但是，我有几个使用相同前缀request.typeA 的队列。因此，我的意图是使用通配符，不必为具有该前缀的每个现有队列定义复合队列，并使其更易于维护。

我需要这样的东西：

<compositeQueue name="request.typeA.>"> 
   <forwardTo>
    <topic physicalName="Notifications" />
   </forwardTo>
 </compositeQueue>

但是那段代码不起作用，我怀疑这是因为它根本不受支持（至少目前还不支持）。我已尝试在physicalName属性中成功使用通配符，但在name中没有。

  • 我的一个先决条件是我必须保留使用相同前缀的不同队列（不能将它们合并为一个）。

  • 我的另一个先决条件是我不能通过代码动态创建新的队列/主题（由于服务器权限）。这就是我对修改xml配置文件感兴趣的原因。

所以我想知道你们中是否有人知道是否可以在name属性中使用通配符（我没有在文档中阅读过任何证据），如果可以，我该怎么做。如果您确定使用当前的 ActiveMQ 版本无法做到这一点，我会感谢您确认。

我也很感激您可以为我打算的相同目的提出其他替代方案/建议，并满足我之前提到的先决条件。我还阅读了有关Mirrored Queues的信息，但是这是一个影响所有现有队列的设置（我只是对其中的一小部分感兴趣）并且可能会对性能产生相当大的影响。

非常感谢您的宝贵时间和最诚挚的问候。

4

2 回答 2

0

我正在使用 5.15.11，我看到通配符适用于复合队列。这是我的复合队列配置。

 <compositeQueue name="email.>">
   <forwardTo>
     <queue physicalName="MY_MAIN_QUEUE" />
   </forwardTo>
 </compositeQueue>
于 2020-02-06T03:21:27.270 回答
0

最后，我找到了一种解决方法，允许我只为队列的一个子集创建 MirroredQueues，给定一个前缀。

我所做的是创建自己的 DestinationInterceptor，以便仅为我感兴趣的那些队列创建一个镜像队列，并排除其余队列（因为默认的MirroredQueue实现镜像了系统中创建的所有队列）。

我是怎么做到的。我将 MirroredQueue.java 类实现从库中复制到了一个名为 CustomMirroredQueue 的新类中，并向名为mirroring的类添加了一个新属性。我从接口DestinationInterceptor修改了intercept(final Destination destination)实现，考虑到 if 语句中的这个新属性（我为它创建了一个名为isPrefixMirrored的辅助方法）：

/*
* This method is responsible for intercepting all the queues/topics that are created in the system.
* In this particular case we are interested only in the queues, in order we can mirror *some* of them and get
* a copy of the messages that are sent to them (With the topics this mirroring is not necessary since we would just
* subscribe to that topic for receiving the same message).
* */
public Destination intercept(final Destination destination) {
    if (destination.getActiveMQDestination().isQueue()) {
        if (isPrefixMirrored(destination) && (!destination.getActiveMQDestination().isTemporary() || brokerService.isUseTempMirroredQueues())) {
            try {
                //we create a mirrored queue for that destination
                final Destination mirrorDestination = getMirrorDestination(destination);
                if (mirrorDestination != null) {
                    return new DestinationFilter(destination) {
                        public void send(ProducerBrokerExchange context, Message message) throws Exception {
                            message.setDestination(mirrorDestination.getActiveMQDestination());
                            mirrorDestination.send(context, message);

                            if (isCopyMessage()) {
                                message = message.copy();
                            }
                            message.setDestination(destination.getActiveMQDestination());
                            message.setMemoryUsage(null); // set this to null so that it will use the queue memoryUsage instance instead of the topic.
                            super.send(context, message);
                        }
                    };
                }
            } catch (Exception e) {
                LOG.error("Failed to lookup the mirror destination for: {}", destination, e);
            }
        }
    }
    return destination;
}

/*
* @returns true if the destination passed as parameter will be mirrored. If the value for the attribute "mirroring"
* is an empty string "" then all the queues will be mirrored by default.
**/
private boolean isPrefixMirrored(Destination destination) {
    if (mirroring.equals("")) {
        return true;
    }
    List<String> mirroredQueuesPrefixes = Arrays.asList(mirroring.split(","));
    final String destinationPhysicalName = destination.getActiveMQDestination().getPhysicalName();
    return mirroredQueuesPrefixes.stream().map(String::trim).anyMatch(destinationPhysicalName::contains);
}

我生成了一个仅包含此自定义类和依赖项（为此使用 gradle）的 .jar，并将其添加到 ActimeMQ 代理安装中的 lib 文件夹中。然后我可以在 ActiveMQ 配置 XML 文件中使用这个标签作为bean ：

<destinationInterceptors>
    <bean xmlns="http://www.springframework.org/schema/beans" class="package.CustomMirroredQueue" id="CustomMirroredQueue">
        <property name="copyMessage" value="true"/>
        <property name="postfix" value=""/>
        <property name="prefix" value="mirror."/>
        <property name="mirroring" value="PREFIX_1, QUEUE2, QUEUE3"/>
    </bean>
</destinationInterceptors>

该类必须包含该类在 ActiveMQ 库文件夹中的路径。属性 copyMessage、postfix 和 prefix 来自默认的 MirroredQueue 实现。并且镜像属性将是一个列表，其中包含要镜像的所有特定队列/前缀（并且只有那些）。

于 2020-02-07T09:16:45.140 回答