个人中心
gpt4 book ai didi

java - 确定 MQ 消息队列何时为空(所有消息均已读取)

转载 作者：行者123 更新时间：2023-12-01 10:07:41 25 4
gpt4 key购买 nike

我正在使用 Spring JMS 创建错误监视工具，目前正在从队列中读取消息并记录错误。我想要的角度是将错误队列中的所有消息存储到映射中。 map 的键基本上是错误信息，值是计数，因为我们想知道有多少个。当所有消息都被读取并且队列上有 0 条消息时，我想运行应用程序的文件创建部分并最终终止应用程序。

我遇到过这些帖子: Check MQ queue depth , How to check whether have message in the queue , Websphere 7 MQueue: how to access queue depth from Java?

我尝试了其中每一个的大杂烩，也查看了IBM的知识中心，但没有取得任何成功。这是我目前拥有的:

public int depthOfQueue() throws MQException {
        queueManager = new MQQueueManager(queueManagerName); // fails here
        MQQueue queue = queueManager.accessQueue(queueName, CMQC.MQOO_OUTPUT);
        int depth = queue.getCurrentDepth();
        queueManager.disconnect();
        queue.close();
        return depth;
    }

但是在我的日志中我得到了如下堆栈跟踪:

java.lang.NoClassDefFoundError: Could not initialize class com.ibm.mq.internal.MQCommonServices
    at com.ibm.mq.MQSESSION.getJmqiEnv(MQSESSION.java:142) ~[com.ibm.mq-7.0.1.9.jar:7.0.1.9 - k701-109-120705]
    at com.ibm.mq.MQQueueManagerFactory.<init>(MQQueueManagerFactory.java:85) ~[com.ibm.mq-7.0.1.9.jar:7.0.1.9 - k701-109-120705]
    at com.ibm.mq.MQQueueManagerFactory.getInstance(MQQueueManagerFactory.java:112) ~[com.ibm.mq-7.0.1.9.jar:7.0.1.9 - k701-109-120705]
    at com.ibm.mq.MQQueueManager.<clinit>(MQQueueManager.java:156) ~[com.ibm.mq-7.0.1.9.jar:7.0.1.9 - k701-109-120705]
    at org.mycom.casts.MQReader.depthOfQueue(MQReader.java:56) ~[classes/:?]
    at org.mycom.casts.MQReader.processOrder(MQReader.java:36) ~[classes/:?]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_74]
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:1.8.0_74]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:1.8.0_74]
    at java.lang.reflect.Method.invoke(Unknown Source) ~[?:1.8.0_74]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:198) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:90) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:66) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:721) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:681) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:651) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:315) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:253) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1150) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1142) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1039) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at java.lang.Thread.run(Unknown Source) [?:1.8.0_74]

和

java.lang.NoClassDefFoundError: Could not initialize class com.ibm.mq.MQQueueManager
    at org.mycom.casts.MQReader.depthOfQueue(MQReader.java:56) ~[classes/:?]
    at org.mycom.casts.MQReader.processOrder(MQReader.java:36) ~[classes/:?]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_74]
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:1.8.0_74]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:1.8.0_74]
    at java.lang.reflect.Method.invoke(Unknown Source) ~[?:1.8.0_74]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:198) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:90) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:66) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:721) ~[spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:681) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:651) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:315) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:253) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1150) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1142) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1039) [spring-jms-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at java.lang.Thread.run(Unknown Source) [?:1.8.0_74]

我读到我可能缺少 MQ 的依赖项，但我的 pom.xml 中有我需要的东西。我还了解到我可以下载 MQ 的客户端版本(或其他东西)，但无法做到这一点。

正如有人在这篇文章中指出的那样:IBM MQ getCurrentDepth requires accessQueue

Finally, there should NEVER be a need to use the getCurrentDepth method in an MQ application. You either get a single message or loop until an MQException is thrown with MQRC_NO_MSG_AVAILABLE (2033) reason code.

所以我的问题是。如何解决抛出的异常(也许我忽略了某些内容)，或者检查 MQException MQRC_NO_MSG_AVAILABLE (2033) 的最佳方法是什么？因为这也应该是队列为空的信号。

最后澄清一下，我使用带有注释的 Spring JMS 来设置和读取队列，所有这些都正常工作，我只需要知道队列何时为空。

为了以防万一，这里是监听器方法:

@JmsListener(destination = "${mq.queueName}")
public void processOrder(String message) throws MQException {
    int messageCount = depthOfQueue();  //problem line
    String messageType = "";
    if(message.startsWith("MSH|") && message.contains("ZER|")){
        messageType = "HL7";
        manager.filterMessage(messageType, message);
    } else if (message.contains(""/*ADD XML CHECK HERE*/)) {
        messageType = "XML";
        manager.filterMessage(messageType, message);
    } else {
        logger.error("Message:\n"+message+"\n does not meet conditions for error");
    }
}

最佳答案

tod​​ji 对此解决方案指出的重要说明:

Note: You're user needs to be given inq permissions on the queue for this to work: setmqaut -m QM1 -n Q1 -t queue -p mq-app +inq

实际上，对于任何研究同一问题的人来说，我都从这两种方法中获得了我需要的功能

public int depthOfQueue() throws MQException {
        queueManager = createQueueManager();
        int openOptions = MQConstants.MQOO_INPUT_AS_Q_DEF | MQConstants.MQOO_OUTPUT | MQConstants.MQOO_INQUIRE;
        MQQueue queue = queueManager.accessQueue(queueName, openOptions);
        int depth = queue.getCurrentDepth();
        queueManager.disconnect();
        queue.close();
        return depth;
    }

和

@SuppressWarnings("unchecked")
    public MQQueueManager createQueueManager() throws MQException {
        MQEnvironment.port = port;
        MQEnvironment.hostname = host;
        MQEnvironment.channel = channel; 
        MQEnvironment.properties.put(MQConstants.TRANSPORT_PROPERTY, MQConstants.TRANSPORT_MQSERIES);
        return new MQQueueManager(queueManagerName);
    }

使用这些:

import com.ibm.mq.*;import com.ibm.mq.constants.MQConstants;

并且您需要在 pom 中提供适当的依赖项:

<dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq.headers</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq.commonservices</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq.pcf</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mqjms</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq.jmqi</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>com.ibm.mq</groupId>
            <artifactId>dhbcore</artifactId>
            <version>7.0.1.9</version>
        </dependency>
        <dependency>
            <groupId>javax.resource</groupId>
            <artifactId>connector</artifactId>
            <version>1.0</version>
        </dependency>

可能是更好的方法，但这可以完成工作。

关于java - 确定 MQ 消息队列何时为空(所有消息均已读取)，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/36336087/

25 4 0
文章推荐： Java正则表达式提取两个字符串之间的特殊字符串
文章推荐： java - SocketChannel 读取时 CPU 负载较高
文章推荐： for-loop - 进行迭代计划的最佳方法
文章推荐： java - Mockito Mock 运行 Autowired 类对象
行者123
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com