通过 QPID JMS 从 azure servicebus 接收来自主题订阅的消息

Question

我正在关注此tutorial 以读取来自给定订阅的消息。我已经创建了一个这样的接收类 -

public class Receive implements MessageListener{

    private static boolean runReceiver = true;
    private Connection connection;
    private Session sendSession;
    private Session receiveSession;
    private MessageProducer sender;
    private MessageConsumer receiver;
    private static Random randomGenerator = new Random();

    public Receive() throws Exception {
        Hashtable<String, String> env = new Hashtable<String, String>();

        env.put("connectionfactory.SBCF", "amqps://All:[shared-access-key]@[namespace].servicebus.windows.net?amqp.idleTimeout=120000");
        env.put("topic.TOPIC", "job/Subscriptions/job-test-subscription");

        env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");

        Context context = new InitialContext(env);

        // Look up ConnectionFactory and Queue
        ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
        Destination queue = (Destination) context.lookup("TOPIC");

        // Create Connection
        connection = cf.createConnection();

        // Create sender-side Session and MessageProducer
        sendSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        sender = sendSession.createProducer(queue);

        if (runReceiver) {
            // Create receiver-side Session, MessageConsumer,and MessageListener
            receiveSession = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            receiver = receiveSession.createConsumer(queue);
            receiver.setMessageListener(this);
            connection.start();
        }
    }
    public void close() throws JMSException {
        connection.close();
    }

    public void onMessage(Message message) {
        try {
            System.out.println("Received message with JMSMessageID = " + message.getJMSMessageID());
            message.acknowledge();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            Receive simpleReceiver = new Receive();
    } catch (Exception e) {
        e.printStackTrace();
    }
}   

在执行这个程序时我收到了这个错误 -

    log4j:WARN No appenders could be found for logger (io.netty.util.internal.logging.InternalLoggerFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
javax.jms.JMSException: hostname can't be null
    at org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:86)
    at org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:108)
    at org.apache.qpid.jms.JmsConnection.connect(JmsConnection.java:172)
    at org.apache.qpid.jms.JmsConnectionFactory.createConnection(JmsConnectionFactory.java:204)
    at org.apache.qpid.jms.JmsConnectionFactory.createConnection(JmsConnectionFactory.java:191)
    at Receive.<init>(Receive.java:41)
    at Receive.main(Receive.java:63)
Caused by: java.io.IOException: hostname can't be null
    at org.apache.qpid.jms.util.IOExceptionSupport.create(IOExceptionSupport.java:45)
    at org.apache.qpid.jms.provider.amqp.AmqpProvider$2.run(AmqpProvider.java:217)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: hostname can't be null
    at java.net.InetSocketAddress.checkHost(InetSocketAddress.java:149)
    at java.net.InetSocketAddress.createUnresolved(InetSocketAddress.java:254)
    at io.netty.bootstrap.Bootstrap.connect(Bootstrap.java:126)
    at org.apache.qpid.jms.transports.netty.NettyTcpTransport.connect(NettyTcpTransport.java:167)
    at org.apache.qpid.jms.provider.amqp.AmqpProvider$2.run(AmqpProvider.java:195)
    ... 7 more

错误指向这条线 - connection = cf.createConnection(); 也想知道像这样提供我的主题和订阅名称 -job/Subscriptions/job-test-subscription 是否可以。

很抱歉发了这么长的帖子。

Answer 1

根据here 中的Service Bus entity address 小节，job/Subscriptions/job-test-subscription 值是您的订阅地址，而不是主题。请更改您的 url 和 topic 属性，然后重试。

connectionfactory.SBCF=amqps://All:[shared-access-key]@[namespace].servicebus.windows.net/job/Subscriptions/job-test-subscription?amqp.idleTimeout=120000
topic.TOPIC=job

希望对你有帮助。