从 ActiveMQ 并行消费消息

Question

每当我第一次将消息发布到队列时，消息被毫无问题地提取，但是当我删除第二个文件时，消息处于“待处理”状态，线程休眠时间（2 分钟）。为了测试在 ActiveMQ 中工作的并发性，我添加了名为 ThreadService 的 bean。

我在 JMSConfig.java 中有如下代码

@Bean
public ActiveMQConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    connectionFactory.setBrokerURL("tcp://localhost:61616");
    connectionFactory.setPassword("admin");
    connectionFactory.setUserName("admin");
    connectionFactory.setTrustedPackages(Arrays.asList("com.jms.domain", "java.util"));
    connectionFactory.setMaxThreadPoolSize(1);
    return connectionFactory;
}

@Bean(destroyMethod = "stop", initMethod = "start")
@Primary
public PooledConnectionFactory pooledConnectionFactory(ConnectionFactory connectionFactory) {
    PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
    pooledConnectionFactory.setConnectionFactory(connectionFactory);
    pooledConnectionFactory.setMaxConnections("8");
    pooledConnectionFactory.setMaximumActiveSessionPerConnection("10");
    return pooledConnectionFactory;
}

@Bean
public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory, DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    configurer.configure(factory, connectionFactory);
    factory.setConcurrency("1-5");
    return factory;
}

CamelRouter.java

from("file://E:/Camel")
    .bean(ThreadService)        
    .to("activemq:MessageQueue");

ThreadService.java

public void ThreadService throws Exception {
    Thread.sleep(120000);
}

如何在 ActiveMQ 中实现并发，使处于待处理状态的消息并行出列？

Answer 1

我很困惑，因为您的问题主题是关于消费，而您的路线正在生产到 ActiveMQ

并行消费者

如果您想并行消费来自 JMS 队列，您通常配置多个消费者。

如果您想为个人消费者执行此操作，您可以将其附加到端点 URI

from("activemq:queue:myQueue?concurrentConsumers=5"

如果您想将此作为每个消费者的默认设置，您可以在您的 bean 设置中进行配置

@Bean
public JmsConfiguration jmsConfiguration() {
    JmsConfiguration jmsConfiguration = new JmsConfiguration();
    jmsConfiguration.setConnectionFactory(pooledConnectionFactory());
    jmsConfiguration.setConcurrentConsumers(5);
    return jmsConfiguration;
}

@Bean(name = "activemq")
public ActiveMQComponent activeMq() {
    ActiveMQComponent activeMQComponent = new ActiveMQComponent();
    activeMQComponent.setConfiguration(jmsConfiguration()); 
    return activeMQComponent;
}

并行生产者

嗯，您的 JMS 生产路线有一个文件使用者，它是按定义为单线程的，以避免处理具有多个使用者的同一个文件。

但是，您可以在使用 Threads DSL of Camel 文件消费后将您的路由转为多线程

from("file://E:/Camel")
    .threads(5) // continue asynchronous from here with 5 threads
    .bean(ThreadService)        
    .to("activemq:MessageQueue");

像这样，ThreadService 中的“长时间运行的任务”不应再阻塞其他文件，因为 threads 语句中的 路由继续与 5 个线程异步。文件使用者保持单线程。

但请注意！ threads 语句中断当前事务。文件使用者将消息交给一个新线程。如果稍后发生错误，文件使用者看不到它。