当前位置：首页 > news >正文

RabbitMQ 消息队列：生产者与消费者实现详解

news 来源：原创 2024/9/30 4:52:53

在分布式系统中，消息队列（Message Queue, MQ）是一种重要的组件，用于解耦系统、异步处理任务以及实现系统间的通信。RabbitMQ 是一个流行的开源消息代理软件，它实现了高级消息队列协议（AMQP）。在本文中，我们将通过 Java 示例来演示如何使用 RabbitMQ 的生产者和消费者模型。

1. 环境准备

首先，确保你已经安装了 RabbitMQ 服务器，并且 Java 环境已经配置妥当。你还需要在项目中引入 RabbitMQ 的 Java 客户端库。通常，这可以通过 Maven 或 Gradle 等构建工具来实现。

2. 生产者（Producer）

生产者负责发送消息到队列。以下是生产者的 Java 实现代码及其注释：

package com.qcby.rabbitmq.mq1;  import com.qcby.rabbitmq.connection.RabbitMQConnection; // 自定义的连接管理类  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  import java.io.IOException;  
import java.util.concurrent.TimeoutException;  // 生产者类  
public class Producer {  private static final String QUEUE_NAME = "boyatopMember"; // 队列名称  public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接  // 通过自定义的 RabbitMQConnection 类获取连接  Connection connection = RabbitMQConnection.getConnection("/boyavirtualHosts"); // 假设这是连接到特定虚拟主机的连接  // 2. 设置通道  // 通道是大多数 RabbitMQ API 调用的入口点  Channel channel = connection.createChannel();  // 3. 设置消息  String msg = "hello world"; // 待发送的消息  System.out.println("msg:" + msg); // 在控制台输出消息内容  // 4. 发送消息到队列  // 第一个参数是交换机名称，这里使用空字符串表示默认交换机（direct类型）  // 第二个参数是队列名称  // 第三个参数是消息的其他属性，这里传null表示默认  // 第四个参数是消息体，需要是字节数组形式  channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());  // 5. 关闭通道和连接  channel.close();  connection.close();  }  
}

3. 消费者（Consumer）

消费者负责从队列中接收并处理消息。以下是消费者的 Java 实现代码及其注释：

package com.qcby.rabbitmq.mq1;  import com.qcby.rabbitmq.connection.RabbitMQConnection; // 自定义的连接管理类  
import com.rabbitmq.client.*;  import java.io.IOException;  
import java.util.concurrent.TimeoutException;  // 消费者类  
public class Consumer {  private static final String QUEUE_NAME = "boyatopMember"; // 队列名称  public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接  Connection connection = RabbitMQConnection.getConnection("/boyavirtualHosts"); // 连接到特定虚拟主机  // 2. 设置通道  Channel channel = connection.createChannel();  // 3. 定义消费者  // 使用 DefaultConsumer 类并覆盖 handleDelivery 方法来处理接收到的消息  DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {  @Override  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  String msg = new String(body, "UTF-8"); // 将字节数组转换为字符串  System.out.println("消费者获取消息:" + msg); // 在控制台输出消息内容  }  };  // 4. 监听队列  // 第一个参数是队列名称  // 第二个参数是是否自动确认消息，true表示自动确认，false表示需要手动确认  // 第三个参数是消费者实例  channel.basicConsume(QUEUE_NAME, true, defaultConsumer);  // 注意：这里的代码实际上会阻塞等待消息的到来。  // 在实际应用中，你可能需要保持这个消费者程序持续运行，直到你显式地停止它。  }  
}

4. 自定义连接（connection）

package com.qcby.rabbitmq.connection;import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class RabbitMQConnection {/*** 获取连接** @return*/public static Connection getConnection(String virtualHost) throws IOException, TimeoutException, TimeoutException {// 1.创建连接ConnectionFactory connectionFactory = new ConnectionFactory();// 2.设置连接地址connectionFactory.setHost("127.0.0.1");// 3.设置端口号:connectionFactory.setPort(5672);// 4.设置账号和密码connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 5.设置VirtualHostconnectionFactory.setVirtualHost(virtualHost);return connectionFactory.newConnection();}
}