连接到多个队列并将它们的消息路由到另一个队列的最有效方法答案

【问题标题】：Most efficient way to connect to multiple queues and route their messages to another queue连接到多个队列并将它们的消息路由到另一个队列的最有效方法
【发布时间】：2016-08-14 20:11:19
【问题描述】：

我正在使用 IBM 的 Websphere MQ 队列，我想连接到其中的多个，浏览每个消息，并将它们全部路由到另一个队列，不同的应用程序将从该队列中读取。

从“主”队列读取的应用程序是用 Java 编写的，并使用 Spring Boot 和 JMS 处理来自队列的消息。我发现用这个应用程序连接到多个队列有困难，所以我的想法是创建一个路由器应用程序，它可以连接到多个队列，清除“主”队列，并用收集到的队列中的消息填充它。

我也在 Java 中开始了“路由器”应用程序，并决定使用 Spring Boot，而不是 JMS，但在概念上仍然发现了一些困难。

所以我想知道 Java 是否适合这项工作（或者我可能缺少一些基本的东西）。我也在考虑 Perl 脚本，但我对这门语言不是很有经验（如果有的话），但我不想贬低它是一个可行的选择。

那么这两种语言中的哪一种可以被证明是：

1) 可维护

2) 有点容易阅读

3) 高效

完成我所描述的任务？

两者的优点和缺点都会很好。

【问题讨论】：

标签： java spring perl ibm-mq processing-efficiency

【解决方案1】：

Java 中最先进的 EIP 集成框架是 Apache Camel。我建议使用它。使用 Apache Camel，您可以编写类似于以下形式的路由器：

public class Routes extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("jms:InputOneQueue1").to("jms:destinationQueue");
        from("jms:InputOneQueue2").to("jms:destinationQueue");
        from("jms:InputOneQueue3").to("jms:destinationQueue");
    }
}

【讨论】：

你知道这是否有使用 Spring 的简单实现吗？因为每个队列的定义不会太冗长？
您可以将 apache camel 与 spring 一起使用。网上有简单的集成示例。您可以使用我的一个项目作为起点github.com/gmaslowski/dam-camel/tree/master/dam-camel-tw
太棒了，谢谢。我会检查一下，看看这是否是我想要前进的方向。

【解决方案2】：

Apache Camel 看起来很有前途，但与我设置 Spring Boot 应用程序的方式不太吻合。我最终使用了 IBM 的 MQ 库。

我基本上都是在设置好队列后使用这些选项

    putMsgOpts = new MQPutMessageOptions();

    getFirstMsgOpts = new MQGetMessageOptions();
    getFirstMsgOpts.options = MQConstants.MQGMO_BROWSE_FIRST;

    getNextMsgOpts = new MQGetMessageOptions();
    getNextMsgOpts.options = MQConstants.MQGMO_BROWSE_NEXT;

那就用我的方法

public void transferQueue(MQPropsManager q1) {
    String message = "";
    message = readFromQueue1(q1, getFirstMsgOpts);
    message = verifyMessage(message);   // just a check for empty or null
    writeToQueue2(message);
    q1.decrementMessagesLeftToProcess();  // decrement initial queue depth 

    while (q1.getMessagesLeftToProcess() > 0) {
        message = readFromQueue1(q1, getNextMsgOpts);
        message = verifyMessage(message);
        writeToQueue2(message);
        q1.decrementMessagesLeftToProcess();
    }
    closeQueue(q1);
}

然后是它调用的两个方法：

public String readFromQueue1(MQPropsManager q1,
            MQGetMessageOptions getMsgOpts) {
        MQMessage msg = new MQMessage();
        String message = "";
        try {
            q1.getQueue().get(msg, getMsgOpts);
            message = msg.readStringOfCharLength(msg.getMessageLength());
        } catch (IOException ioe) {
        //  Failed to read string retreived from queue: q1.getQueueName()
        } catch (MQException mqe) {
        //  Failed to retreive message from queue: q1.getQueueName()
        }
        return message;
    }

public void writeToQueue2(String message) {
        MQMessage mqMessage = new MQMessage();
        mqMessage.format = MQConstants.MQFMT_STRING;
        mqMessage.messageType = MQConstants.MQMT_DATAGRAM;

        try {
            mqMessage.writeString(message);
            q2.getQueue().put(mqMessage, putMsgOpts);
        } catch (IOException ioe) {
        //  Failed to write message: message 
        } catch (MQException mqe) {
        //  Failed to put message: message on to the queue2
        }
    }

【讨论】：