java - 使用 Spring Boot 和 AMQP 订阅 Azure 服务总线主题

Question

我设置了一个名为“state-changed”的 Azure 服务总线主题，并且它有一个名为“reverb”的订阅。我正在尝试使用 @JmsListener 设置一个方法来订阅该主题，但收到错误:

2017-03-22 18:34:41.049  WARN 23356 --- [enerContainer-6] o.s.j.l.DefaultMessageListenerContainer  : Setup of JMS message listener invoker failed for destination 'state-changed' - trying to recover. Cause: The messaging entity 'sb://[MySERVICEBUS].servicebus.windows.net/state-changed' could not be found. TrackingId:d2b442f79e0f44bdb449861ea57155ce_G44, SystemTracker:gateway6, Timestamp:3/22/2017 6:34:37 PM

javax.jms.JMSException: The messaging entity 'sb://[MySERVICEBUS].servicebus.windows.net/state-changed' could not be found. TrackingId:d2b442f79e0f44bdb449861ea57155ce_G44, SystemTracker:gateway6, Timestamp:3/22/2017 6:34:37 PM
    at org.apache.qpid.amqp_1_0.jms.impl.TopicSubscriberImpl.createClientReceiver(TopicSubscriberImpl.java:111) ~[qpid-amqp-1-0-client-jms-0.32.jar:0.32]
    at org.apache.qpid.amqp_1_0.jms.impl.MessageConsumerImpl.<init>(MessageConsumerImpl.java:129) ~[qpid-amqp-1-0-client-jms-0.32.jar:0.32]
    at org.apache.qpid.amqp_1_0.jms.impl.TopicSubscriberImpl.<init>(TopicSubscriberImpl.java:46) ~[qpid-amqp-1-0-client-jms-0.32.jar:0.32]
    at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl.createDurableSubscriber(SessionImpl.java:544) ~[qpid-amqp-1-0-client-jms-0.32.jar:0.32]
    at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl.createDurableSubscriber(SessionImpl.java:59) ~[qpid-amqp-1-0-client-jms-0.32.jar:0.32]
    at org.springframework.jms.listener.AbstractMessageListenerContainer.createConsumer(AbstractMessageListenerContainer.java:870) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.createListenerConsumer(AbstractPollingMessageListenerContainer.java:215) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.initResourcesIfNecessary(DefaultMessageListenerContainer.java:1189) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1165) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1158) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1055) ~[spring-jms-4.3.6.RELEASE.jar:4.3.6.RELEASE]
    at java.lang.Thread.run(Unknown Source) [na:1.8.0_77]

我一直在使用这篇博文来尝试让一切正常运行:http://ramblingstechnical.blogspot.co.uk/p/using-azure-service-bus-with-spring-jms.html

我可以使用 JmsTemplate 将消息添加到主题，并使用 Azure 文档中概述的普通旧 Java JMS 库从中读取消息:https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-java-how-to-use-jms-api-amqp所以我知道我的主题确实有效并且可以访问，只是当我使用 Spring 配置它时，我似乎做错了一些事情。

我的配置如下:

@Configuration
public class JmsConfiguration
{

    @Bean
    public JmsListenerContainerFactory topicJmsListenerContainerFactory() throws NamingException
    {
        DefaultJmsListenerContainerFactory returnValue = new DefaultJmsListenerContainerFactory();

        Context context = context();
        ConnectionFactory cf = connectionFactory(context);

        returnValue.setConnectionFactory(cf);
        returnValue.setSubscriptionDurable(Boolean.TRUE);
        return returnValue;
    }

    private Context context() throws NamingException
    {
        Hashtable<String, String> env = new Hashtable<String, String>();
        env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory");
        env.put(Context.PROVIDER_URL, "src/main/resources/servicebus.properties");
        Context context = new InitialContext(env);
        return context;
    }



    /**
     * @param context
     * @return
     * @throws NamingException
     */
    private ConnectionFactory connectionFactory(Context context) throws NamingException
    {
        ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
        return cf;
    }

}

servicebus.properties(用户名和密码等已编辑):

# servicebus.properties - sample JNDI configuration

# Register a ConnectionFactory in JNDI using the form:
# connectionfactory.[jndi_name] = [ConnectionURL]
connectionfactory.SBCF=amqps://[USER]:[PASSWORD]@[MYSERVICEBUS]

# Register some queues in JNDI using the form
# queue.[jndi_name] = [physical_name]
# topic.[jndi_name] = [physical_name]
queue.workflow = workflow
topic.state-changed = stage-changed

最后是我的监听器类:

@Component
public class TestListener
{
    Logger logger = LoggerFactory.getLogger(LoggingWorkflowEventHandler.class);

    @JmsListener(destination = "state-changed", containerFactory = "topicJmsListenerContainerFactory", subscription = "reverb")
    public void onMessage(String message)
    {
        logger.info("Received message from topic: {}", message);
    }
}

如果有人设法让这个工作正常运行，我将不胜感激。

Answer 1

您的错误消息表明未找到您的目的地名称(未找到消息实体)。请注意，您需要以如下特定方式告诉 Azure 您的订阅名称:

<TopicName>/Subscriptions/<SubscriptionName>

就您而言:

state-changed/Subscriptions/reverb

希望有帮助

干杯 塞布