DefaultMessageListenerContainer 不从 IBM MQ 读取消息 [英] DefaultMessageListenerContainer Not Reading Messages from IBM MQ

问题描述

我正在尝试创建一些使用 DefaultMessageListenerContainer 来侦听来自 IBM MQ 的消息的 Spring Boot 代码.

I am trying to create some Spring Boot code that uses DefaultMessageListenerContainer to listen to messages from IBM MQ.

我可以创建 MQQueueConnectionFactory 并使用 JmsTemplate 发送和接收消息，但这是为了实现高吞吐量并且想要使用侦听器而不是轮询.

I can create the MQQueueConnectionFactory and send and receive messages using JmsTemplate but this is intended to by high throughput and want use a listener instead of polling.

我已经将大部分代码整合到一个组件中，所以我希望我拥有所有相关的内容.

I have consolidate much of the code down to a Component so I hope I have all that is relevant.

如果我安排了 receiveMessage 方法，它会选择排队的消息，所以我知道 sendMessage 方法正在排队消息.

If I schedule the receiveMessage method, it picks up queued messages so I know the sendMessage method is queuing messages.

@Component
class AllInOneTest {

    private MessagingConfiguration.QueueConfig config;
    private MQQueueConnectionFactory           connectionFactory;
    private JmsTemplate                        jmsTemplate;
    private DefaultMessageListenerContainer    listenerContainer;
    private final Logger                       logger = LoggerFactory.getLogger(getClass());

    public AllInOneTest(MessagingManager manager) throws JMSException {
        String detailsName = "default";
        config = manager.getMessagingDetails(detailsName).getConfig();

        logger.debug("AllInOneTest Initializing Connection Factory: {}", detailsName);
        connectionFactory = new MQQueueConnectionFactory();
        connectionFactory.setHostName(config.getHost());
        connectionFactory.setPort(config.getPort());
        connectionFactory.setTransportType(config.getTransportType());
        connectionFactory.setQueueManager(config.getQueueManager());
        connectionFactory.setChannel(config.getChannel());

        logger.debug("AllInOneTest Initializing Message Listener: {}", detailsName);
        DefaultMessageListenerContainer defaultListener = new DefaultMessageListenerContainer();
        defaultListener.setConnectionFactory(connectionFactory);

        defaultListener.setExceptionListener((ee) -> {
            logger.warn(String.format("AllInOneTest Message Listener Error: %s", detailsName), ee);
        });

        defaultListener.setDestinationResolver((session, name, pubSub) -> {
            Destination ret = session.createQueue(name);

            logger.debug("AllInOneTest Created Listener Destination: {}", ret);

            return ret;
        });

        defaultListener.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                logger.info("AllInOneTest Listening For Message: {}", message);
            }
        });

        // TODO Configure subscription.
        // defaultListener.setSubscriptionDurable(true);
        // defaultListener.setSubscriptionName("masher-service");

        // TODO Configure concurrency.
        // defaultListener.setConcurrency(config.getConcurrency());

        // TODO Configure transaction.
        // defaultListener.setSessionTransacted(config.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE);

        listenerContainer = defaultListener;

        logger.debug("AllInOneTest Initializing JMS Template: {}", detailsName);
        jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setMessageConverter(new SpringToJMSMessageConverter());
        jmsTemplate.setReceiveTimeout(1000L);
        jmsTemplate.setDefaultDestinationName(config.getOutputQueue());
        jmsTemplate.setDestinationResolver((session, name, pubSub) -> {
            Destination ret = session.createQueue(name);

            logger.debug("AllInOneTest Created JMS Template Destination: {}", ret);

            return ret;
        });

        listenerContainer.setDestinationName(config.getOutputQueue());

        logger.debug("AllInOneTest Starting Message Listener: {} on {}", detailsName, config.getOutputQueue());
        listenerContainer.start();
    }

    // @Scheduled(fixedRate = 500L)
    public void receiveMessage() {
        Object message = jmsTemplate.receiveAndConvert();
        if (message != null) {
            logger.info("AllInOneTest Received: {}", message);
        }
    }

    @Scheduled(fixedRate = 1500L)
    public void sendMessage() {
        int count = counter.incrementAndGet();
        org.springframework.messaging.Message<String> message = MessageBuilder.withPayload(String.format("JMS Masher Message %d %s %s", count,
                new SimpleDateFormat("HH:mm:ss.SSS").format(new Date()), UUID.randomUUID().toString())).build();

        logger.info("AllInOneTest Sending: {} [{}]", message.getPayload(), message.getHeaders());

        jmsTemplate.convertAndSend(config.getInputQueue(), message);
    }

}

我正在调用 DefaultMessageListenerContainer.start() 但我感觉它不是开始"，我一定是遗漏了什么.

I am calling DefaultMessageListenerContainer.start() but I get the sense it is not "starting" and I must be missing something.

为 JmsTemplate 而不是为 DefaultMessageListenerContainer 调用 DestinationResolver.

The DestinationResolver is called for the JmsTemplate but not the DefaultMessageListenerContainer.

我在控制台中没有看到任何异常.

I don't see any exceptions in the console.

感谢您的帮助，韦斯.

推荐答案

DefaultMessageListenerContainer defaultListener = new DefaultMessageListenerContainer();

DefaultMessageListenerContainer defaultListener = new DefaultMessageListenerContainer();

当你以编程方式创建容器时，而不是让 Spring 作为 @Bean 管理它，你必须调用 afterPropertiesSet()(在你设置所有属性之后，然后在你start()之前.

When you create the container programmatically, rather than having Spring manage it as a @Bean, you have to call afterPropertiesSet() (after you have set all the properties, and before you start() it).

对于许多 Spring 组件来说都是如此.通常最好让 Spring 管理它们.

This is true for many Spring components. It's generally better to let Spring manage them.

这篇关于DefaultMessageListenerContainer 不从 IBM MQ 读取消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！