本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
连接您的 JMS 应用程序
本教程向您展示如何使用 RabbitMQ JMS 客户端将 JMS 应用程序连接到 Amazon MQ for RabbitMQ 代理。您将学习如何创建用于发送消息的创建器和用于接收来自 RabbitMQ 队列的消息的使用者。
在开始之前,请将相应的 RabbitMQ JMS 依赖项添加到你的 Maven 项目中:
对于 JMS 1.1 和 2.0:
<dependencies> <dependency> <groupId>com.rabbitmq.jms</groupId> <artifactId>rabbitmq-jms</artifactId> <version>2.12.0</version> </dependency> </dependencies>
对于 JMS 3.1:
<dependencies> <dependency> <groupId>com.rabbitmq.jms</groupId> <artifactId>rabbitmq-jms</artifactId> <version>3.5.0</version> </dependency> </dependencies>
创建制作人
以下代码示例显示了如何使用 JMS 写入 RabbitMQ 队列:
import jakarta.jms.*; import com.rabbitmq.jms.admin.*; // Setting the connection factory RMQConnectionFactory factory = new RMQConnectionFactory(); factory.setHost(envProps.getProperty("RABBITMQ_HOST", "localhost")); factory.setPort(Integer.parseInt(envProps.getProperty("RABBITMQ_PORT", "5672"))); factory.setUsername(envProps.getProperty("RABBITMQ_USERNAME", "guest")); factory.setPassword(envProps.getProperty("RABBITMQ_PASSWORD", "guest")); factory.setVirtualHost(envProps.getProperty("RABBITMQ_VIRTUAL_HOST", "/")); factory.useSslProtocol(); connection = factory.createConnection(); connection.start(); String queueName = "test-queue-jms"; Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); RMQDestination destination = new RMQDestination(queueName, true, false); // Send the message to the queue MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.PERSISTENT); String msg_content = "Hello World!!"; TextMessage textMessage = session.createTextMessage(msg_content); producer.send(textMessage); System.out.printf("Published to AMQP queue '%s': %s", queueName, msg_content);
创建消费者
以下代码示例显示了如何使用 JMS 从 RabbitMQ 队列中读取:
import jakarta.jms.*; import com.rabbitmq.jms.admin.*; // Setting the connection factory RMQConnectionFactory factory = new RMQConnectionFactory(); factory.setHost(envProps.getProperty("RABBITMQ_HOST", "localhost")); factory.setPort(Integer.parseInt(envProps.getProperty("RABBITMQ_PORT", "5672"))); factory.setUsername(envProps.getProperty("RABBITMQ_USERNAME", "guest")); factory.setPassword(envProps.getProperty("RABBITMQ_PASSWORD", "guest")); factory.setVirtualHost(envProps.getProperty("RABBITMQ_VIRTUAL_HOST", "/")); factory.useSslProtocol(); // Establish the connection and session jakarta.jms.Connection connection = factory.createConnection(); String queueName = "test-queue-jms"; Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); RMQDestination destination = new RMQDestination(); destination.setDestinationName(queueName); destination.setAmqp(true); destination.setAmqpQueueName(queueName); // Initialize consumer MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(message -> { try { if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; System.out.printf("Message: %s%n", textMessage.getText()); } else if (message instanceof BytesMessage) { BytesMessage bytesMessage = (BytesMessage) message; byte[] bytes = new byte[(int) bytesMessage.getBodyLength()]; bytesMessage.readBytes(bytes); String content = new String(bytes); System.out.printf("Message: %s%n", content); } else { System.out.printf("Message: [%s]%n", message.getClass().getSimpleName()); } } catch (JMSException e) { System.err.printf("Error processing message: %s%n", e.getMessage()); } }); connection.start();