java - 无法使用基于 JMS 的代码和 amqp 1.0 访问 ActiveMQ

Question

我正在尝试使用 AMQP 1.0 连接到 ActiveMQ 代理，但我想在我的应用程序代码中使用 JMS。我对使用 JMS 感兴趣主要是因为我希望开发人员能够使用他们已经熟悉的 API。

我在本地主机上运行 ActiveMQ 5.14.0 和以下代码:

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

    Connection connection = null;
    try {
        // Producer
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("amqp://localhost:5672");

        connection = connectionFactory.createConnection();
        connection.start();     

        Session session = connection.createSession(false,
                                                   Session.AUTO_ACKNOWLEDGE);

        Topic topic = session.createTopic("customerTopic");     

        // Publish
        MessageProducer producer = session.createProducer(topic);
        for ( int i = 0; i < 10; i++) {
            Message msg = session.createTextMessage("Task : " + i);

            producer.send(msg);

        }
        session.close();
    } finally {
        if (connection != null) {
            connection.close();
        }

    }


}

代码总是以同样的方式失败，堆栈跟踪中的根本原因如下:

由 org.apache.activemq.transport.InactivityIOException 引起: channel 处于非 Activity 状态的时间太长(>30000):tcp://127.0.0.1:5672

这发生在 connection.start() 方法调用上。

如果我针对 ActiveMQ tcp 端点运行相同的代码，那么它会按预期执行。

我的 pom 文件依赖项如下(我怀疑这是我的问题的根源，因为我发现依赖项的文档非常难以理解)

<dependencies>
    <dependency>
        <groupId>org.apache.qpid</groupId>
        <artifactId>qpid-amqp-1-0-client-jms</artifactId>
        <version>0.32</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-amqp</artifactId>
        <version>5.14.0</version>
    </dependency>

</dependencies>

我直接的问题是“为什么这不起作用？”。

我的补充(基于意见)问题是“是否值得尝试使用 AMQP 1.0 之上的 JMS 抽象，或者我应该放弃学习提供者特定的 API？”

Answer 1

与jndi合作更好

public static void main(String[] args) throws JMSException, InterruptedException, NamingException {
    Connection connection = null;
    try {
        Properties props = new Properties();
        props.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        props.setProperty("connectionfactory.myFactoryLookup",
                "amqp://localhost:5672");
        props.put("topic." + "MyTOPIC", "customerTopic");
        InitialContext ic = new InitialContext(props);
        ConnectionFactory cf1 = (ConnectionFactory) ic.lookup("myFactoryLookup");
        Topic topic = (Topic) ic.lookup("MyTOPIC");
        connection = cf1.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(topic);
        connection.start();
        for (int i = 0; i < 10; i++) {
            Message msg = session.createTextMessage("Task : " + i);
            producer.send(msg);
        }
        session.close();
    } finally {
        if (connection != null) {
            connection.close();
        }
    }
}

替换

 <dependency>
    <groupId>org.apache.qpid</groupId>
    <artifactId>qpid-amqp-1-0-client-jms</artifactId>
    <version>0.32</version>
</dependency>

由

    <dependency>
        <groupId>org.apache.qpid</groupId>
        <artifactId>qpid-jms-client</artifactId>
        <version>0.9.0</version>
    </dependency>

在经纪商方面您需要添加:

 <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?transport.transformer=jms"/>

引用http://activemq.apache.org/amqp.html