针对Quarkus JMS在连接ActiveMQ时遇到的AMQP协议不匹配错误，本文详细阐述了问题根源在于ActiveMQ默认OpenWire协议与Quarkus JMS（基于Qpid JMS）所依赖的AMQP协议不符。教程将指导用户如何通过配置ActiveMQ服务器启用AMQP传输连接器，并确保Quarkus客户端连接到正确的AMQP端口，从而实现Quarkus应用与ActiveMQ的顺畅集成，并简要提及消息监听器的配置方式。
引言：Quarkus JMS与ActiveMQ协议概述
Quarkus作为一种现代的、云原生的Java框架，其JMS（Java Message Service）扩展是基于Apache Qpid JMS实现的。Qpid JMS原生支持AMQP 1.0协议，这是一种国际标准化的消息传输协议，旨在提供跨平台、跨语言的互操作性。
ActiveMQ，作为Apache基金会下的一个流行的开源消息代理，支持多种消息协议，其中包括其原生的OpenWire协议和AMQP 1.0协议。默认情况下，ActiveMQ通常在端口61616上启用OpenWire协议。理解这些协议差异以及Quarkus JMS对AMQP的依赖，是成功集成两者的关键。
理解“AMQP SASL header mismatch”错误
当Quarkus应用尝试使用其内置的JMS扩展连接到ActiveMQ时，如果遇到类似AMQP SASL header mismatch value 0, expecting 41. In state: HEADER0的错误，这通常意味着AMQP客户端尝试连接到了一个非AMQP协议的端口。
这个错误信息揭示了问题的本质：
expecting 41：AMQP协议头部的第一个字节通常是ASCII字符’A’，其十进制值为41。Quarkus JMS客户端期望接收到这个值，以确认它连接到了一个AMQP服务。value 0：然而，客户端实际接收到的第一个字节是0。这通常是ActiveMQ OpenWire协议头部的特征。
因此，这个错误明确指出Quarkus JMS客户端（一个AMQP客户端）连接到了ActiveMQ的OpenWire端口，导致协议不匹配。以下代码片段对比了两种连接方式：
// 尝试使用Quarkus JMS提供的ConnectionFactory (基于Qpid JMS, 期望AMQP)// 在Quarkus应用中，通常会注入ConnectionFactory，但如果直接new，且Quarkus JMS扩展激活，// 可能会尝试使用其默认配置或Qpid的ConnectionFactory实现。// 如果连接到OpenWire端口，此处会抛出 AMQP SASL header mismatch 错误。// ConnectionFactory connectionFactory = new ConnectionFactory();// connection = connectionFactory.createConnection();// 使用ActiveMQConnectionFactory (原生OpenWire, 连接61616)// 此方式直接指定了OpenWire协议，因此可以正常连接到61616端口。ActiveMQConnectionFactory activeMQFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");connection = activeMQFactory.createConnection("admin", "admin"); // 此处工作正常
上述对比清晰地表明，问题的核心在于客户端所使用的协议与ActiveMQ服务器监听的协议不一致。
解决方案：配置ActiveMQ服务器以支持AMQP
要解决协议不匹配问题，核心在于配置ActiveMQ服务器以启用AMQP传输连接器，并确保Quarkus应用连接到这个AMQP端口。
对于ActiveMQ Classic（5.x版本），您需要在其配置文件conf/activemq.xml中添加或修改部分，以包含一个AMQP连接器。
ActiveMQ Classic 配置示例：
九歌
九歌–人工智能诗歌写作系统
322 查看详情
在上述配置中，我们添加了一个名为amqp的transportConnector，它监听在标准AMQP端口5672上。请注意，如果您使用的是ActiveMQ Artemis，其配置方式会有所不同，但同样需要确保配置一个AMQP协议的连接器。
完成配置修改后，请务必重启ActiveMQ Broker，以使新的传输连接器生效。
Quarkus应用连接AMQP端口
一旦ActiveMQ服务器正确配置并启用了AMQP传输连接器（例如在5672端口），Quarkus应用就可以通过其JMS扩展连接到该端口。您需要在Quarkus项目的application.properties或application.yaml文件中指定正确的AMQP连接URL。
application.properties 配置示例：
# Quarkus JMS 配置quarkus.jms.url=amqp://localhost:5672quarkus.jms.username=adminquarkus.jms.password=admin# 可选：配置连接池quarkus.jms.pool.enabled=truequarkus.jms.pool.max-connections=10
配置完成后，Quarkus将自动提供一个jakarta.jms.ConnectionFactory或jakarta.jms.JMSContext的CDI bean，您可以在您的应用中直接注入并使用它们来生产或消费消息。
Quarkus应用中使用JMS的示例：
import jakarta.enterprise.context.ApplicationScoped;import jakarta.inject.Inject;import jakarta.jms.ConnectionFactory;import jakarta.jms.JMSContext;import jakarta.jms.Queue;import jakarta.jms.JMSException;@ApplicationScopedpublic class JmsProducerService { @Inject ConnectionFactory connectionFactory; // Quarkus会自动提供基于配置的ConnectionFactory public void sendMessage(String queueName, String messageContent) { try (JMSContext context = connectionFactory.createContext()) { Queue queue = context.createQueue(queueName); context.createProducer().send(queue, messageContent); System.out.println("Message sent to " + queueName + ": " + messageContent); } catch (JMSException e) { System.err.println("Error sending message: " + e.getMessage()); e.printStackTrace(); } }}
关于消息监听器（MessageListener）
在Quarkus中，一旦JMS连接成功建立，您可以使用标准的JMS API来创建消息消费者并注册MessageListener。虽然Quarkus不像Spring那样提供特定的@JMSListener注解，但您可以利用CDI生命周期管理和JMS API来实现类似的功能。
Quarkus应用中设置MessageListener的示例：
import jakarta.enterprise.context.ApplicationScoped;import jakarta.enterprise.event.Observes;import jakarta.inject.Inject;import jakarta.jms.ConnectionFactory;import jakarta.jms.JMSContext;import jakarta.jms.Queue;import jakarta.jms.MessageConsumer;import jakarta.jms.MessageListener;import jakarta.jms.Message;import jakarta.jms.JMSException;import io.quarkus.runtime.ShutdownEvent;import io.quarkus.runtime.StartupEvent;@ApplicationScopedpublic class JmsConsumerService { @Inject ConnectionFactory connectionFactory; private JMSContext context; private MessageConsumer consumer; // 在应用启动时设置消息监听器 void onStart(@Observes StartupEvent ev) { try { context = connectionFactory.createContext(); Queue queue = context.createQueue("inQue"); // 监听的队列名称 consumer = context.createConsumer(queue); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { try { String receivedMessage = message.getBody(String.class); System.out.println("Received message: " + receivedMessage); // 在此处处理接收到的消息 // 如果需要手动确认，请根据会话模式调用 message.acknowledge() } catch (JMSException e) { System.err.println("Error processing message: " + e.getMessage()); e.printStackTrace(); } } }); context.start(); // 启动连接以开始接收消息 System.out.println("JMS MessageListener started for queue 'inQue'."); } catch (JMSException e) { System.err.println("Failed to set up JMS MessageListener: " + e.getMessage()); e.printStackTrace(); } } // 在应用关闭时清理资源 void onStop(@Observes ShutdownEvent ev) { try { if (consumer != null) { consumer.close(); } if (context != null) { context.close(); } System.out.println("JMS MessageListener stopped."); } catch (JMSException e) { System.err.println("Error closing JMS resources: " + e.getMessage()); e.printStackTrace(); } }}
注意事项： 在生产环境中，对于长时间运行的消息监听器，需要确保JMS Connection和Session（或JMSContext）的生命周期得到妥善管理，防止资源泄露
以上就是Quarkus与ActiveMQ JMS集成指南：解决AMQP协议不匹配问题的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1031088.html
微信扫一扫 
支付宝扫一扫 
