RabbitMQ队列

### RabbitMQ 队列使用教程与示例代码 #### 什么是 RabbitMQ？ RabbitMQ 是一个开源的消息队列系统，它实现了高级消息队列协议（AMQP），并基于生产者-消费者模型运行。该工具支持多种消息传递模式，包括点对点、发布/订阅以及工作队列[^1]。 #### 声明队列和交换机 在 Java 中可以通过以下两种方式声明队列和交换机： 1. **通过 `Channel` 对象直接创建** 使用 `channel.queueDeclare()` 方法来定义一个新的队列。 2. **绑定到特定的 Exchange 上** 这种情况下需要先声明 Exchange 并将其与 Queue 绑定在一起[^4]。 下面是一段简单的 Java 示例代码用于演示如何声明队列： ```java import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class DeclareQueueExample { public static void main(String[] args) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { String queueName = "example_queue"; // 创建持久化队列 channel.queueDeclare(queueName, true, false, false, null); System.out.println("Queue declared successfully."); } } } ``` #### 设置队列 TTL 和延迟插件 为了给队列中的每条消息设置生存时间（TTL），或者实现延迟队列功能，可以借助官方社区提供的插件——`rabbitmq_delayed_message_exchange`。此插件允许开发者发送带有延迟属性的消息，在指定的时间之后才被投递给消费者[^5]。 ##### 安装延迟插件步骤 1. 访问 GitHub 页面下载最新版插件文件：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/latest/download/rabbitmq_delayed_message_exchange-[version].ez 2. 将解压后的 `.ez` 文件复制至 RabbitMQ 插件目录下，默认路径为 `/usr/lib/rabbitmq/lib/rabbitmq_server-[version]/plugins/` 3. 启用插件执行命令：`sudo rabbitmq-plugins enable rabbitmq_delayed_message_exchange` 启用成功后即可按照文档说明配置 Delayed Message Exchange 类型，并测试其效果。 #### PHP 脚本管理 RabbitMQ 队列状态 如果希望查看当前服务器上的所有队列及其状态信息，则可通过 CLI 工具完成操作。具体指令如下所示[^2]: ```bash /opt/modules/rabbitmq/sbin/rabbitmqctl list_queues status ``` 上述命令会返回一系列关于各个队列的状态数据，比如内存占用情况、未确认消息数量等等。 #### C# 开发环境下的 RabbitMQ 实现方案 对于 .NET 应用程序而言，集成 RabbitMQ 的过程相对简便。只需利用 NuGet 包管理器引入依赖项 “RabbitMQ.Client”，随后参照官方样例编写消费端逻辑即可[^3]。 这里给出一段基础框架供参考： ```csharp using RabbitMQ.Client; using System.Text; class Program { static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using(var connection = factory.CreateConnection()) using(var channel = connection.CreateModel()) { string queueName = "hello_world"; Console.WriteLine($" [*] Waiting for messages on '{queueName}'..."); while(true){ BasicDeliverEventArgs ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); byte[] body = ea.Body.ToArray(); string message = Encoding.UTF8.GetString(body); Console.WriteLine($">>> Received: {message}"); // 手动应答已处理完毕的消息 channel.BasicAck(ea.DeliveryTag, multiple:false); } } } } ``` ---