使用ActiveMQ创建持久的主题和订户spring boot jm

Question

我需要为ActiveMQ创建一个主题和一个持久的订阅者，我的问题是我不知道在哪里指定它。我可以创建主题并使用消息，但当我关闭订阅者，然后继续发送消息并再次打开订阅者时，它不会阅读它们。

这就是我到目前为止所知道的：

发送消息：

    JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
    jmsTemplate.setPubSubDomain(true);
    jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
    jmsTemplate.setDeliveryPersistent(true);
    jmsTemplate.convertAndSend("venta.topic",venta);

接收消息：

@JmsListener(destination = "venta.topic",id = "comercial",subscription = "venta.topic")
public void receiveMessage(Venta venta) {
    logger.log(Level.INFO, "RECEIVED : {0}",venta);      
    repository.save(venta);
}

我已经阅读了this article，并且我知道我需要创建持久订阅者。

我也读过spring docs

我认为这与DefaultJmsListenerContainerFactory有关(我没有实现它，我使用的是默认配置)，文档显示：

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory =
            new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setDestinationResolver(destinationResolver());
    factory.setConcurrency("3-10");
    return factory;
}

但是我似乎找不到在哪里创建持久会话。我的生产者和订阅者都连接到一个独立的activemq二进制文件。

我希望你能帮助我，提前谢谢。

Answer 1

正如前面的回答所指出的，有必要在工厂上设置客户端id和持久订阅：

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory =
            new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setDestinationResolver(destinationResolver());
    factory.setConcurrency("3-10");
    factory.setClientID("brokerClientId");
    factory.setSubscriptionDurable(true);
    return factory;
}

但这并不能单独将客户端注册为持久订阅者，这是因为JMSListener需要指定的containerFactory，否则它只会采用默认值：

@JmsListener(
destination = "venta.topic",
id = "comercial",
subscription = "venta.topic",
//this was also needed with the same name as the bean above
containerFactory = "jmsListenerContainerFactory" 
)
public void receiveMessage(Venta venta) {
            logger.log(Level.INFO, "RECEIVED : {0}",venta);      
            repository.save(venta);
}

值得一提的是，this post让我认识到了我的错误。

我希望这能帮助其他人

Answer 2

DefaultJmsListenerContainerFactory应该有唯一的clientId和持久的sub。true set类似于下面的代码：

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory =
            new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setDestinationResolver(destinationResolver());
    factory.setConcurrency("3-10");
    factory.setClientID("brokerClientId");
    factory.setSubscriptionDurable(true);
    return factory;
}

Answer 3

很难说，但这个问题的一个常见原因是忘记了在connectionFactory bean上配置唯一的clientId。它必须是唯一的，并且是经纪人区分每个客户端的方式。