ServiceBus for Windows Server上的Java-选项？

Question

使用Windows服务总线开发Java应用程序有哪些选项？

>

JavaMessage Broker API-这需要ACS来使用，SB for Win不支持。

AMQP-这似乎不适用于SB for Windows，我一直收到错误

组织。阿帕奇。qpid。amqp\u 1\u 0。客户发件人$SenderCreationException：对等方未为链接创建远程终结点，目标：

而同样的代码适用于Azure SB。那么，SB for Windows上的AMQP似乎没有完全正常工作？

如果我遗漏了什么，请纠正我？

更新以在本地计算机上测试AMQP，这就是我所做的

>

连接工厂。SBCF=amqps：//

队列。QUEUE=queue1

代码使用证书更新。

在运行时我得到这个错误

javax.jms.JMSException: Peer did not create remote endpoint for link, target: queue1
    at org.apache.qpid.amqp_1_0.jms.impl.MessageProducerImpl.<init>(MessageProducerImpl.java:77)
    at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl.createProducer(SessionImpl.java:348)
    at org.apache.qpid.amqp_1_0.jms.impl.SessionImpl.createProducer(SessionImpl.java:63)
    at com.stringcompany.Analyzer.SimpleSenderReceiver.<init>(SimpleSenderReceiver.java:70)
    at com.stringcompany.Analyzer.SimpleSenderReceiver.main(SimpleSenderReceiver.java:95)
Caused by: org.apache.qpid.amqp_1_0.client.Sender$SenderCreationException: Peer did not create remote endpoint for link, target: queue1
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:171)
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:104)
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:97)
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:83)
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:69)
    at org.apache.qpid.amqp_1_0.client.Sender.<init>(Sender.java:63)
    at org.apache.qpid.amqp_1_0.client.Session.createSender(Session.java:74)
    at org.apache.qpid.amqp_1_0.client.Session.createSender(Session.java:66)
    at org.apache.qpid.amqp_1_0.jms.impl.MessageProducerImpl.<init>(MessageProducerImpl.java:72)
    ... 4 more
javax.jms.JMSException: Session remotely closed

使用相同的代码如果我通过设置SB命名空间和队列指向Azure服务总线，如下所示

连接工厂。SBCF=amqps：//

这样就可以进行信息交换。

如果有人想试试，下面是代码

package com.stringcompany.Analyzer;

//SimpleSenderReceiver.java

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Hashtable;
import java.util.Random;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class SimpleSenderReceiver 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 SimpleSenderReceiver() throws Exception {
    // Configure JNDI environment
    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, "D:\\Java\\Azure\\workspace\\Analyzer\\src\\main\\resources\\servicebus.properties");
    Context context = new InitialContext(env);

    // Lookup ConnectionFactory and Queue
    ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");

    System.out.println("cf:"+cf);

    // Create Connection
    connection = cf.createConnection();
    System.out.println("connection :"+connection);

    connection.setExceptionListener(new ExceptionListener() {

        public void onException(JMSException arg0) {
            System.err.println(arg0);

        }
    });
    connection.start();



    // Create sender-side Session and MessageProducer
    sendSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    System.out.println("Session open");


    Destination queue = (Destination) context.lookup("QUEUE");
    System.out.println("queue:"+queue);

    sender = sendSession.createProducer(queue);
    Queue q=(Queue) queue;
    System.out.println(sender.getDestination());
    System.out.println("sender:"+sender);
    if (runReceiver) {
        System.out.println("Waitng for new message");
        // Create receiver-side Session, MessageConsumer,and MessageListener
        receiveSession = connection.createSession(false,
                Session.CLIENT_ACKNOWLEDGE);
        receiver = receiveSession.createConsumer(queue);
        receiver.setMessageListener(this);
        connection.start();
    }
}

public static void main(String[] args) {
    try {

        if ((args.length > 0) && args[0].equalsIgnoreCase("sendonly")) {
            runReceiver = false;
        }

        //System.setProperty("javax.net.debug","ssl");
        System.setProperty("javax.net.ssl.trustStore","D:\\Java\\Azure\\workspace\\Analyzer\\src\\main\\resources\\SBKeystore.keystore");
        System.setProperty("log4j.configuration","D:\\Java\\Azure\\workspace\\Analyzer\\src\\main\\resources\\log4j.properties");
        SimpleSenderReceiver simpleSenderReceiver = new SimpleSenderReceiver();
        System.out
                .println("Press [enter] to send a message. Type 'exit' + [enter] to quit.");
        BufferedReader commandLine = new java.io.BufferedReader(
                new InputStreamReader(System.in));

        while (true) {
            String s = "Message";//commandLine.readLine();
            if (s.equalsIgnoreCase("exit")) {
                simpleSenderReceiver.close();
                System.exit(0);
            } else {
                simpleSenderReceiver.sendMessage();
            }
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

private void sendMessage() throws JMSException {
    TextMessage message = sendSession.createTextMessage();
    message.setText("Test AMQP message from JMS");
    long randomMessageID = randomGenerator.nextLong() >>> 1;
    message.setJMSMessageID("ID:" + randomMessageID);
    sender.send(message);
    System.out.println("Sent message with JMSMessageID = "
            + message.getJMSMessageID());
}

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();
    }
}

  }

Answer 1

windows server Service Bus 1.1支持AMQP 1.0。基本上，在ServiceBus中使用AMQP的cloud和on-prem有两个区别：1。寻址：您将需要构建AMQP连接字符串（如果您正在寻找HA，则需要DNS）2。身份验证：您需要使用加入域的帐户，因为ACS不存在on-prem。您还需要将SB证书分发给您的客户端。

Answer 2

这个问题最简单的答案是您应该对SASPolcyKey进行URL编码。

connectionfactory.SBCF = amqps://[SASPolicyName]:[SASPolicyKey]@[namespace].servicebus.windows.net

其中SASPolicyKey应为URL编码。

Answer 3

嗨，我们有同样的问题，谢天谢地MS更新了他们的留档，以展示如何正确地做到这一点。：http://msdn.microsoft.com/en-us/library/dn574799.aspx