一个生产者和一个震荡者的Spring JMS负载平衡

Question

我已经对JMS负载平衡进行了很长时间的研究。我们可以为负载平衡JMS消息创建多生产者和多消费者。但我想了解，我们如何在一个生产者和一个消费者之间负载平衡JMS消息。我不能像ApacheCamel那样在我的项目中添加更多依赖项。

@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 类，我无法在其中进行重大的配置更改，例如引入更多 JMS 模板或更多连接工厂。我的制片人如下所示

@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;
}
}

Answer 1

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

只要目标是队列（而不是主题），这是隐含的，因为您有＜code＞工厂。setPubSubDomain（false）则它将正常工作。它是JMS合同的一部分。如果同一队列中有多个消费者，则消息将分布在这些消费者之间；只有一个消费者将接收特定消息。

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

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