Quarkus与ActiveMQ JMS集成指南：解决AMQP协议不匹配问题

// 尝试使用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"); // 此处工作正常

<brokers>
    <broker brokerName="localhost" dataDirectory="${activemq.data}">
        <!-- ... 其他配置 ... -->

        <transportConnectors>
           <!-- OpenWire 协议 (默认) -->
           <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>

           <!-- AMQP 1.0 协议 (推荐使用标准端口5672) -->
           <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>

        <!-- ... 其他配置 ... -->
    </broker>
</brokers>

# Quarkus JMS 配置
quarkus.jms.url=amqp://localhost:5672
quarkus.jms.username=admin
quarkus.jms.password=admin
# 可选：配置连接池
quarkus.jms.pool.enabled=true
quarkus.jms.pool.max-connections=10

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;

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

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;

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