时间戳

首页 > 解决方案 > Spring JMS 负载均衡，一生产者一消费者

spring - Spring JMS 负载均衡，一生产者一消费者

问题描述

关于 JMS 负载平衡，我研究了很长时间。我们可以创建多个生产者和多个消费者来对 JMS 消息进行负载平衡。但我想了解，我们如何使用一个生产者和一个消费者对 JMS 消息进行负载平衡。我无法在我的项目中添加更多依赖项，例如 Apache Camel。

@Configuration
@EnableJms
@ComponentScan({"com.jmsloadbalance.jms"})
@Bean
public class JmsConfig {
public JmsTemplate getJmsTemplate() {
    JmsTemplate template = new JmsTemplate();
    template.setConnectionFactory(connectionFactory());
    template.setDefaultDestination(new ActiveMQQueue("default.topic");
    template.setExplicitQosEnabled(true);
    template.setDeliveryPersistent(false);
    template.setTimeToLive(60000);
    template.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
    template.setMessageConverter(getMessageConverter());
    return template;
}

@Bean
public DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setPubSubDomain(false);
    factory.setDestinationResolver(new DynamicDestinationResolver());
    factory.setConcurrency("1");
    factory.setMessageConverter(getMessageConverter());
    return factory;
}

private ActiveMQConnectionFactory connectionFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    factory.setBrokerURL("vm://localhost");
    return factory;
}

private MessageConverter getMessageConverter() {
    MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    converter.setTypeIdPropertyName("JMSType");
    return converter;
}
}

这是我的 JmsConfig 类，我无法在其中进行大的配置更改，例如引入更多 JMSTemplate 或更多 ConnectionFactory。我的制作人如下所示

@Service("accountJmsProducer")
public class AccountJmsProducer {

private static Logger LOG = Logger.getLogger(AccountJmsProducer.class);

@Autowired
private JmsTemplate template;

private Destination destination;

public Account create(Account account) {
    if (this.destination == null) {
        this.destination = new ActiveMQQueue("account.create");
    }
    template.convertAndSend(destination, account);
    return null;
}
}

我的消费者如下所示：

@Service("accountJmsConsumer")
public class AccountJmsConsumer {

private static final Logger LOG = Logger.getLogger(AccountJmsConsumer.class);

@Autowired
@Qualifier("accountService")
private AccountService accountService;

private Account lastReceived;

@JmsListener(containerFactory = "defaultJmsListenerContainerFactory", destination = "account.create")
public Account create(Account account) {
    LOG.warn("Received " + account);
    setLastReceived(account);
    return accountService.create(account);
}
public synchronized Account getLastReceived() {
    return lastReceived;
}
public synchronized void setLastReceived(Account lastReceived) {
    this.lastReceived = lastReceived;
}
}

标签： springactivemqload-balancingspring-jms

解决方案

目前尚不清楚当有一个消费者时负载平衡是什么意思，但根据您对我对您问题的评论的评论：

只要目的地是一个队列（不是一个主题）并且这是隐含的，因为你有factory.setPubSubDomain(false)那么它就可以工作。它是 JMS 合同的一部分。如果同一个队列上有多个消费者，消息将在这些消费者之间分发；只有一个消费者会收到特定的消息。

如果交付失败，它可能会或可能不会重新交付给同一消费者。

大多数代理（包括 ActiveMQ）都提供某种预取机制。IIRC，使用 ActiveMQ 默认为 1000。如果您的消息少于此数量，则可能有一个消费者处于空闲状态；如果是这样，请减少预取以调整分布。

推荐阅读