从 MQ 异步接收多条消息

Question

我在我的应用程序中使用 Spring + Hibernate + JPA。

我需要从Websphere MQ 读取消息并将消息插入数据库。 有时可能会有连续的消息可用，有时消息的数量会非常少，有时我们可能无法期待来自Queue 的消息。

目前我正在一一阅读消息并将它们插入数据库。但在性能方面并没有多大帮助。

我的意思是当我有大量消息时（例如队列中的 30 万条消息），我无法更快地插入它们。每秒插入数据库的实体数量不是很高。因为我确实为每一个实体做出了承诺。

我想使用休眠批处理，以便我可以在单个提交中插入实体列表。 （例如：每次提交 30 到 40 条消息）

问题：

1. 如何从队列接收多条消息？ （我已经检查过 BatchMessageListenerContainer 可能会有所帮助。但我无法获得一些参考）

2. 我应该将 db 插入过程与 onMessage 方法分开吗？那么该线程将被释放到池中并可用于从队列中挑选下一条消息？

3. 并行线程使用情况？

当前实现：

消息监听器：

<bean id="myMessageListener" class="org.mypackage.MyMessageListener">

<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="destinationName" ref="queue"/>
    <property name="messageListener" ref="myMessageListener"/>
    <property name ="concurrentConsumers" value ="10"/>
    <property name ="maxConcurrentConsumers" value ="50"/>        
</bean>

监听类：

package org.mypackage.MyMessageListener;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;

import org.mypackage.service.MyService;

public class MyMessageListener implements MessageListener {

    @Autowired
    private MyService myService;

    @Override
    public void onMessage(Message message) {
        try {
             TextMessage textMessage = (TextMessage) message;
             // parse the message
             // Process the message to DB
        } catch (JMSException e1) {
             e1.printStackTrace();
        }
    }
}

Answer 1

不清楚您的要求是什么。

Spring Batch 项目提供了一个BatchMessageListenerContainer。

消息侦听器容器适用于通过配置提供的建议来拦截消息接收。 要在单个事务中启用消息批处理，请在建议链中使用 TransactionInterceptor 和 RepeatOperationsInterceptor（在基类中设置或不设置事务管理器）。然后容器将使用RepeatOperations 在同一个线程中接收多条消息，而不是接收一条消息并对其进行处理。与 RepeatOperations 和事务拦截器一起使用。如果事务拦截器使用 XA，则使用 XA 连接工厂或 TransactionAwareConnectionFactoryProxy 将 JMS 会话与正在进行的事务同步（打开失败后重复消息的可能性）。在后一种情况下，您不需要在基类中提供事务管理器 - 它只会在途中阻止 JMS 会话与数据库事务同步。