欢迎来到尧图网

客户服务 关于我们

您的位置：首页 > 财经 > 创投人物 > RabbitMQ实战-JavaDemo

RabbitMQ实战-JavaDemo

2025/4/19 13:53:52 来源：https://blog.csdn.net/StaticKing/article/details/141497026  浏览:  次   关键词：RabbitMQ实战-JavaDemo

目录

前言

消息生产者

消息消费者

消息确认机制

消息持久化

Maven 依赖

总结

前言

在使用 RabbitMQ 进行消息传递时，了解如何在代码中创建和发布消息（生产者）、接收和处理消息（消费者），以及配置消息确认机制和持久化，是确保系统可靠性和效率的关键。以下将详细解读这些概念，并提供相应的 Java 示例代码和 Maven 依赖。

注：如果是跟着笔者一步步认识RabbitMQ以及手动搭建Rabbit，则需要注意，在搭建RabbitMQ时，笔者运行RabbitMQ使用了指定的vhosts=all，但在本示例中笔者并未指定虚拟主机(vhost)，则默认使用/ ，如果你没有创建这个则会报错503，可参考https://blog.csdn.net/StaticKing/article/details/141467806?spm=1001.2014.3001.5502 此篇文章解决，如果你可以正常登录控制台则可在登录控台后，通过页面进行手动添加

消息生产者

消息生产者负责创建并发送消息到 RabbitMQ 服务器的交换机或队列中。

如何在代码中创建和发布消息：

  • 步骤概述：
    1. 创建 ConnectionFactory 并配置 RabbitMQ 服务器的连接参数。
    2. 创建 Connection 并从中创建 Channel。
    3. 声明要使用的交换机或队列（如果尚未存在）。
    4. 使用 basicPublish 方法发布消息。

Java 示例：生产者发布消息到队列

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;public class Producer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("123456");// 2. 创建连接和通道try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 3. 声明队列（确保队列存在）boolean durable = true; // 队列持久化channel.queueDeclare(QUEUE_NAME, durable, false, false, null);// 4. 创建消息String message = "Hello World!";// 5. 发布消息channel.basicPublish("", QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN, // 消息持久化message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}

说明：

  • queueDeclare 方法用于声明一个队列。如果队列已存在且参数匹配，该方法不会有任何效果。
  • basicPublish 方法将消息发送到指定的交换机和路由键。在这个例子中，我们使用默认交换机（空字符串），并将消息直接发布到指定队列。
  • MessageProperties.PERSISTENT_TEXT_PLAIN 设置消息为持久化，以确保在 RabbitMQ 重启后消息不会丢失。

消息消费者

消息消费者负责从 RabbitMQ 服务器的队列中接收并处理消息。

如何在代码中创建和消费消息：

  • 步骤概述：
    1. 创建 ConnectionFactory 并配置连接参数。
    2. 创建 Connection 并从中创建 Channel。
    3. 声明要消费的队列（确保队列存在）。
    4. 设置消费回调，定义如何处理接收到的消息。
    5. 使用 basicConsume 方法开始消费。

Java 示例：消费者接收并处理消息

import com.rabbitmq.client.*;public class Consumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("123456");// 2. 创建连接和通道Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 3. 声明队列（确保队列存在）boolean durable = true;channel.queueDeclare(QUEUE_NAME, durable, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 4. 设置消费者回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");try {// 模拟消息处理doWork(message);} finally {System.out.println(" [x] Done");// 5. 手动确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};// 6. 取消自动确认（设置为 false）boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });}private static void doWork(String task) {// 模拟耗时操作try {Thread.sleep(1000); // 模拟处理时间} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}
}

说明：

  • DeliverCallback 接口用于定义接收到消息时的处理逻辑。
  • basicConsume 方法开始消费消息，参数 autoAck 设置为 false，表示手动确认消息。

消息确认机制

消息确认机制用于确保消息被可靠地处理。RabbitMQ 提供了自动确认和手动确认两种方式，以及在处理失败时的消息重发机制。

确认类型：

  • 自动确认（Auto Acknowledge）：

    • 消费者在接收消息后，RabbitMQ 自动将消息标记为已被处理，无需等待消费者的确认。
    • 如果消费者在处理消息时崩溃或失败，消息可能会丢失。
    • 适用于对消息可靠性要求较低的场景。

  • 手动确认（Manual Acknowledge）：

    • 消费者在成功处理消息后，显式发送确认信号给 RabbitMQ。
    • 如果消费者处理消息失败或在处理过程中崩溃，RabbitMQ 会重新将消息投递给其他消费者。
    • 提高消息的可靠性，适用于需要确保每条消息都被处理的场景。

Java 示例：手动确认机制

在上述消费者示例中，autoAck 被设置为 false，并在 finally 块中调用了 channel.basicAck 方法进行手动确认。下面是对手动确认机制的更详细说明和示例：

import com.rabbitmq.client.*;public class ManualAckConsumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();boolean durable = true;channel.queueDeclare(QUEUE_NAME, durable, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");try {// 处理消息doWork(message);// 处理成功，发送确认channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);System.out.println(" [x] Done and Acknowledged");} catch (Exception e) {// 处理失败，不发送确认，消息会被重新投递System.out.println(" [x] Error processing message, not acknowledged");}};boolean autoAck = false; // 手动确认channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });}private static void doWork(String task) throws Exception {// 模拟处理逻辑if (task.contains("fail")) {throw new Exception("Simulated processing failure");}Thread.sleep(1000);}
}

说明：

  • 在消费者处理逻辑中，如果消息处理成功，调用 basicAck 方法发送确认。
  • 如果在处理过程中出现异常，消费者不发送确认，RabbitMQ 将重新投递消息。

消息持久化

为了确保消息在 RabbitMQ 崩溃或重启后不会丢失，需要配置消息持久化。消息持久化包括：

  • 持久化队列：确保队列在服务器重启后仍然存在。
  • 持久化消息：确保消息在服务器重启后不会丢失。

如何实现消息持久化：

  • 持久化队列：在声明队列时，将 durable 参数设置为 true。
  • 持久化消息：在发布消息时，将消息属性的 deliveryMode 设置为 2（持久化）。

Java 示例：持久化队列和消息

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;public class PersistentProducer {private final static String QUEUE_NAME = "persistent_task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("123456");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明持久化队列boolean durable = true;channel.queueDeclare(QUEUE_NAME, durable, false, false, null);String message = "Persistent Message";// 发布持久化消息channel.basicPublish("", QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}

说明：

  • channel.queueDeclare 中将 durable 设置为 true，声明持久化队列。
  • 使用 MessageProperties.PERSISTENT_TEXT_PLAIN 设置消息为持久化。

消费者端：

消费者端无需特殊设置来处理持久化消息，只需要正确配置队列即可。

import com.rabbitmq.client.*;public class PersistentConsumer {private final static String QUEUE_NAME = "persistent_task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();boolean durable = true;channel.queueDeclare(QUEUE_NAME, durable, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");try {doWork(message);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (Exception e) {System.out.println(" [x] Error processing message");}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });}private static void doWork(String task) throws Exception {// 模拟处理Thread.sleep(1000);}
}

总结：

通过配置队列和消息的持久化机制，可以确保即使在 RabbitMQ 崩溃或重启后，消息仍然不会丢失。持久化对于需要高可靠性的消息传递系统尤为重要。

Maven 依赖

为了在 Java 项目中使用 RabbitMQ，需要添加 RabbitMQ Java 客户端库的 Maven 依赖。在 pom.xml 文件中添加以下内容：

<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.15.0</version> <!-- 使用最新稳定版本 --></dependency>
</dependencies>

说明：

  • com.rabbitmq:amqp-client 是 RabbitMQ 提供的官方 Java 客户端库。
  • 确保使用最新的稳定版本，以获取最新的功能和安全更新。

总结

理解 RabbitMQ 的消息生产与消费机制，以及相关的确认和持久化机制，对于构建可靠的消息传递系统至关重要。通过正确配置生产者和消费者，并应用手动确认和消息持久化机制，可以确保消息在分布式系统中可靠、安全地传递和处理。结合 Docker 部署 RabbitMQ 和 Maven 集成 Java 客户端，开发者可以快速搭建和测试基于 RabbitMQ 的消息传递解决方案。

版权声明:

本网仅为发布的内容提供存储空间，不对发表、转载的内容提供任何形式的保证。凡本网注明“来源：XXX网络”的作品，均转载自其它媒体，著作权归作者所有，商业转载请联系作者获得授权，非商业转载请注明出处。

我们尊重并感谢每一位作者，均已注明文章来源和作者。如因作品内容、版权或其它问题，请及时与我们联系，联系邮箱：809451989@qq.com，投稿邮箱：809451989@qq.com

相关资讯

热文排行

最新新闻

推荐新闻

热搜词