PHP使用rabbitMQ

在 PHP 中使用 RabbitMQ 通常是为了处理惩罚异步任务、队列、消息推送等场景，特别是在高并发、分布式体系中，RabbitMQ 提供了可靠的消息队列服务。RabbitMQ 是基于 AMQP 协议的消息中间件，具有高效、可靠、可扩展的特点。
下面将先容怎样在 PHP 中使用 RabbitMQ，并结合场景给出实现方法。
一、RabbitMQ 底子概念

• Producer（生产者）：发送消息的客户端，负责将消息发送到消息队列。
  
• Consumer（消费者）：接收消息的客户端，从队列中获取消息并处理惩罚。
  
• Queue（队列）：存储消息的中间容器，生产者将消息发送到队列，消费者从队列中取出消息处理惩罚。
  
• Exchange（交换机）：接收生产者发来的消息，并根据路由规则分发到队列。
  
• Binding（绑定）：将队列绑定到交换机，使得消息可以或许根据规则路由到对应的队列。
  

二、安装和设置 RabbitMQ

• 安装 RabbitMQ 服务器 你可以通过包管理工具（如 apt-get 或 yum）来安装 RabbitMQ，也可以在官方提供的 Docker 镜像上运行 RabbitMQ 服务器：
  
  1. docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

  复制代码
• 管理 RabbitMQ RabbitMQ 提供了一个管理插件，可以通过 Web UI 进行管理。访问 http://localhost:15672，默认账号暗码都是 guest。
  
• 安装 PHP 客户端库 使用 php-amqplib 来连接 RabbitMQ。在项目中可以通过 Composer 安装：
  
  1. composer require php-amqplib/php-amqplib

  复制代码

三、PHP 使用 RabbitMQ 实例

1. 生产者发送消息

1. <?php
2. require_once __DIR__ . '/vendor/autoload.php';
4. use PhpAmqpLib\Connection\AMQPStreamConnection;
5. use PhpAmqpLib\Message\AMQPMessage;
7. // 创建连接到 RabbitMQ 服务器
8. $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
9. $channel = $connection->channel();
11. // 声明队列
12. $channel->queue_declare('task_queue', false, true, false, false);
14. // 发送的消息内容
15. $data = "Hello World!";
16. $msg = new AMQPMessage($data, array('delivery_mode' => 2)); // 设置消息为持久化
18. // 发布消息到队列
19. $channel->basic_publish($msg, '', 'task_queue');
21. echo " [x] Sent 'Hello World!'\n";
23. // 关闭连接
24. $channel->close();
25. $connection->close();
26. ?>

复制代码

表明：

• queue_declare 用于声明队列。假如队列不存在，则会创建。假如队列已经存在，则不会做任何操作。
  
• basic_publish 用于将消息发布到队列中。
  
• 使用 delivery_mode 设置消息恒久化（2 表示恒久化）。
  

2. 消费者接收消息

1. <?php
2. require_once __DIR__ . '/vendor/autoload.php';
4. use PhpAmqpLib\Connection\AMQPStreamConnection;
6. // 创建连接到 RabbitMQ 服务器
7. $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
8. $channel = $connection->channel();
10. // 声明队列
11. $channel->queue_declare('task_queue', false, true, false, false);
13. echo " [*] Waiting for messages. To exit press CTRL+C\n";
15. // 回调函数，用于处理接收到的消息
16. $callback = function ($msg) {
17.     echo ' [x] Received ', $msg->body, "\n";
18.     // 模拟任务处理
19.     sleep(substr_count($msg->body, '.'));
20.     echo " [x] Done\n";
21.     // 手动确认消息已经处理
22.     $msg->ack();
23. };
25. // 告诉 RabbitMQ 在同一时间不要发送多于一条消息给一个消费者
26. $channel->basic_qos(null, 1, null);
28. // 告诉 RabbitMQ 使用回调函数来接收消息，并手动确认消息
29. $channel->basic_consume('task_queue', '', false, false, false, false, $callback);
31. // 等待消息进入队列
32. while ($channel->is_consuming()) {
33.     $channel->wait();
34. }
36. // 关闭连接
37. $channel->close();
38. $connection->close();
39. ?>

复制代码

表明：

• 消费者通过 basic_consume 来订阅队列的消息，并指定一个回调函数 callback 来处理惩罚收到的消息。
  
• basic_qos(null, 1, null) 是为了设置“公平分发”，确保在消费者处理惩罚完前一条消息后才会接收到下一条消息。
  
• msg->ack() 是用来手动确认消息已经处理惩罚完毕，RabbitMQ 才会将该消息从队列中删除。
  

3. RabbitMQ 工作队列的应用场景

• 异步任务处理惩罚：好比发送邮件、天生报告、图片处理惩罚等，客户端将任务发送到消息队列，消费者异步处理惩罚任务。
  
• 负载均衡：多个消费者监听同一个队列，RabbitMQ 会将消息匀称分发到每个消费者上，实现任务的负载均衡。
  
• 任务重试：结合 ack 和 nack 确认机制，假如消费者在处理惩罚消息时出现错误，可以拒绝该消息并重新入队列处理惩罚。
  

四、RabbitMQ 高级特性

1. 消息恒久化

消息恒久化是为了防止 RabbitMQ 崩溃或重启时消息丢失。要做到恒久化，需要将队列和消息都设置为恒久化。

• 恒久化队列：在声明队列时，设置 durable 参数为 true。
  
• 恒久化消息：在创建消息时，设置 delivery_mode 参数为 2。
  

2. 延迟队列（延时任务）

通过 RabbitMQ 的 x-delayed-message 插件，可以实现延时任务。消息发送后不会立刻被消费，而是经过一段时间后才会被放入队列中消费。
3. 消息确认机制

消息确认机制是为了确保消息被正确处理惩罚，避免消息丢失。RabbitMQ 提供了手动和自动两种消息确认方式。手动确认（如上述消费者中的 ack）可以确保消息在处理惩罚失败时不会丢失。
五、总结

RabbitMQ 是一个强大且机动的消息队列体系，结合 PHP 可以实现许多高级应用场景，如异步任务、任务重试、负载均衡等。在高并发和分布式体系中，RabbitMQ 可以有效提高体系的可扩展性和稳定性。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。