Java Qpid Proton - ActiveMQ 代理无法分配请求的地址：绑定

Question

我真的需要这个问题的答案，这就是我编辑它的原因。

我使用此代码在我的连接中构建了一个 Apache ActiveMQ 代理

Broker.java

public class经纪人{

private BrokerService broker;

public Broker(String connector) {
    this.broker = new BrokerService();
    this.broker.setUseJmx(true);
    try {
        this.broker.addConnector(connector);

    } catch (URISyntaxException e) {
        e.printStackTrace();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public void addConnector(String connector){
    try {
        this.broker.addConnector(connector);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public void start() {
    try {
        this.broker.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public BrokerService getBroker() {
    return broker;
}

public void setBroker(BrokerService broker) {
    this.broker = broker;
}

}

这是我的问题

我使用 Qpid Proton 库（可在此处获取：Qpid Proton）。我有一个 class 来读取数据，这几乎是他们在 qpid webiste

上给你的例子

package messaging;

import java.io.IOException;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.engine.BaseHandler;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.message.Message;
import org.apache.qpid.proton.reactor.FlowController;
import org.apache.qpid.proton.reactor.Handshaker;

public class AMQPSubscriber extends BaseHandler {

    private String broker;
    private String topic;
    private String port;

    public AMQPSubscriber(String broker, String port, String topic) {
        this.broker = broker;
        this.port = port;
        this.topic = topic;
        this.add(new Handshaker());
        this.add(new FlowController());

    }

    @Override
    public void onReactorInit(Event event) {
        try {
            event.getReactor().acceptor(broker, Integer.parseInt(port), new AMQPSubscriber(broker, port, topic));
        } catch (NumberFormatException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onDelivery(Event event) {
        System.out.println("---------Message Received--------");
        Receiver recv = (Receiver) event.getLink();
        Delivery delivery = recv.current();
        if (delivery.isReadable() && !delivery.isPartial()) {
            int size = delivery.pending();
            byte[] buffer = new byte[size];
            int read = recv.recv(buffer, 0, buffer.length);
            recv.advance();

            Message msg = Proton.message();
            msg.decode(buffer, 0, read);
            System.out.println("Subject : " + msg.getProperties().getSubject());
            System.out.println("Text : " + ((AmqpValue) msg.getBody()).getValue());
        }
    }

}

这个class主要是调用:

public static void main (String[]args) throws IOException, TimeoutException, InterruptedException{

    Broker broker = new Broker("amqp://" + host + ":" + AMQPport);
    broker.start();

 AMQPSubscriber receiv = new AMQPSubscriber(host, "5672", topic);
        Reactor r;
        try {
            r = Proton.reactor(receiv);
            r.run();
        } catch (IOException e) {
            e.printStackTrace();
        }


}

但是当我执行这段代码时，我得到一个

 INFO | Loaded the Bouncy Castle security provider.
 INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[C:\Users\alexi\Documents\workspace-sts-3.7.3.RELEASE\IOT\activemq-data\localhost\KahaDB]
 INFO | JMX consoles can connect to service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi
 INFO | KahaDB is version 6
 INFO | Recovering from the journal @1:61115
 INFO | Recovery replayed 11 operations from the journal in 0.014 seconds.
 INFO | PListStore:[C:\Users\alexi\Documents\workspace-sts-3.7.3.RELEASE\IOT\activemq-data\localhost\tmp_storage] started
 INFO | Apache ActiveMQ 5.13.3 (localhost, ID:DESKTOP-UK0JIC4-52783-1467025817901-0:1) is starting
 INFO | Listening for connections at: amqp://127.0.0.1:5672
 INFO | Connector amqp://127.0.0.1:5672 started
 INFO | Apache ActiveMQ 5.13.3 (localhost, ID:DESKTOP-UK0JIC4-52783-1467025817901-0:1) started
 INFO | For help or more information please see: http://activemq.apache.org
 WARN | Store limit is 102400 mb (current store usage is 0 mb). The data directory: C:\Users\alexi\Documents\workspace-sts-3.7.3.RELEASE\IOT\activemq-data\localhost\KahaDB only has 7792 mb of usable space. - resetting to maximum available disk space: 7792 mb
 WARN | Temporary Store limit is 51200 mb (current store usage is 0 mb). The data directory: C:\Users\alexi\Documents\workspace-sts-3.7.3.RELEASE\IOT\activemq-data\localhost only has 7792 mb of usable space. - resetting to maximum available disk space: 7792 mb
java.net.BindException: Address already in use: bind
    at sun.nio.ch.Net.bind0(Native Method)
    at sun.nio.ch.Net.bind(Unknown Source)
    at sun.nio.ch.Net.bind(Unknown Source)
    at sun.nio.ch.ServerSocketChannelImpl.bind(Unknown Source)
    at java.nio.channels.ServerSocketChannel.bind(Unknown Source)
    at org.apache.qpid.proton.reactor.impl.AcceptorImpl.<init>(AcceptorImpl.java:102)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.acceptor(ReactorImpl.java:477)
    at messaging.AMQPSubscriber.onReactorInit(AMQPSubscriber.java:33)
    at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:209)
    at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
    at org.apache.qpid.proton.engine.impl.EventImpl.delegate(EventImpl.java:129)
    at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:114)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:307)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:275)
    at org.apache.qpid.proton.reactor.impl.ReactorImpl.run(ReactorImpl.java:343)
    at messaging.Main.main(Main.java:98)

当我使用 MQTT 和 Paho 时，这个代理工作正常，我希望它也能与 AMQP 一起工作。我知道绑定意味着端口已被使用，但我不知道如何在没有数据发送的端口上监听。

谢谢你的帮助。

亚历克斯

Answer 1

当有另一个应用程序创建了一个侦听您请求的目标端口的服务器套接字时，您会收到该错误。您需要检查您不是运行该端口上的另一个代理实例，或者有一些阻止访问的防火墙。

Answer 2

我找到了解决办法。

当我将连接器添加到 activeMQ 嵌入式代理时，它会将其添加为一次只允许一个连接的 TCP。

所以我创建了一个像这样的 UDP 连接器： broker.addConnector("udp://"+host+":"+AMQPport);

这个解决方案对我有用我希望它能在未来帮助其他开发者。

干杯，Alexi

Java Qpid Proton - ActiveMQ 代理无法分配请求的地址：绑定

Java Qpid Proton - ActiveMQ broker Cannot assign requested address: bind

java

activemq

bind

amqp

qpid