基础简单消息队列应用
在上一课中，我们已经学习到了什么是消息队列，有哪些消息队列，以及我们会用到哪个消息队列。今天，就直接进入主题，学习第一种，最简单，但也是最常用，最好用的消息队列模式。
最简单的队列功能
最简单的队列功能，无非就是将我们在数据结构与算法中学过的那个队列结构，变成一个外部功能组件。让各种语言和各种应用程序都可以通过这个队列来进行数据操作。这样的一个队列系统就称之为“消息队列中间件”。
一般，我们会将生产消息的程序，或者说，将数据放入到队列的一方称为 P （生产者，Producer）；然后将队列称为Q（Queue）；最后，将守候在队列前，等待从队列中获取数据的应用、程序或者代码段称为 C（消费者/客户端，Consumer）。
这是 RabbitMQ 官网手册上的图，后面的相关图示我们也将直接使用它们的。在这个图中，有字母的部分就不多解释了。中间红色的一格一格的部分代表的就是 Q 。
通常来说，P 只管将数据放到 Q 中，Q 负责中间存储数据，然后 C 取出数据。数据的进出方式遵循经典数据结构队列中的规则，也就是 FIFO 先进先出。
是不是就和我们之前说过的一样，最简单的理解，它就是将队列这个数据结构抽成了一个第三方组件。这样就可以跨平台、跨应用、跨语言地进行数据存取了。
RebbitMQ 实现
好了，先来看 RabbitMQ 的实现。你需要先安装好 RabbitMQ ，我这里是使用的 Docker 安装的。
代码语言：javascript复制docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management
然后，也要安装好 PHP 的 Composer 组件，用于操作 RabbitMQ 消息队列，PHP 需要开启 sockets 扩展。
代码语言：javascript复制composer require php-amqplib/php-amqplib
5672 是 RabbitMQ 的服务端口，15672 则是它自带的一个管理工具的访问端口。具体的内容大家可以到官方文档中进行更加深入的学习。当然，也可以使用虚拟机方式来搭建测试环境，这个大家看自己的喜好吧。
接下来，我们先实现 P 端的代码，也就是生产者向消息队列中添加数据。
代码语言：javascript复制// 2.rq.p.php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;
// 建立连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel(); // 获取频道
// 定义队列
$channel->queue_declare('hello', false, false, false, false);
// 创建消息
$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'hello'); // 将消息放入队列中
echo "生产者向消息队列中发送信息：Hello World！";
$channel->close();
$connection->close();
注释很清晰了吧，如果你之前没有用 PHP 操作过 RabbitMQ 的话，那么直接复制这段代码就可以了。其中比较特殊的是 channel ，它是共享单个 TCP 连接的轻量级信道。这个概念可能是 RabbitMQ 相较于其它消息队列系统比较特别的。然后就是消息内容是通过一个 AMQPMessage 对象承载的，这个 AMQP 其实就是 RabbitMQ 的核心协议。协议是通信双方能够相互看明白对方的基础，能够建立通信的基础，就像我们之前在 Redis 中学习过的 RESP 协议一样。这里大家只需要知道 RabbitMQ 是使用这个协议就好了，而且它也支持其它的一些协议。发送完消息之后，记得关闭连接哦。
好了，接下来是我们的消费者/客户端实现。
代码语言：javascript复制// 2.rq.c.php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLibConnectionAMQPStreamConnection;
// 建立连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel(); // 获取频道
// 定义队列
$channel->queue_declare('hello', false, false, false, false);
echo "等待消息，或者使用 Ctrl C 退出程序。", PHP_EOL;
// 定义接收数据的回调函数
$callback = function ($msg) {
echo '接收到数据： ', $msg->body, PHP_EOL;
};
// 消费队列，获取到数据将调用 callback 回调函数
$channel->basic_consume('hello', '', false, true, false, false, $callback);
// 频道是开启状态时，挂起程序，不停地执行
while ($channel->is_open()) {
// 等待并监听频道中的队列信息
// 发现上方 basic_consume 定义的队列有消息后
// 就调用它对应的 callback
$channel->wait();
}
看着好像多很多东西呀？其实上面一半和生产者是一样的。从中间开始，我们使用的是 Channel 对象的 basic_consume() 方法，这个方法最后有一个回调函数参数。然后在下面通过 wait() 方法持续监听队列中是否有数据。如果有数据了，就调用指定的回调函数。并将消息内容交给回调函数的参数。
注意哦，一般来说，消息队列的消费者，或者说是客户端，或者说是 C 端。大部分情况下可能都会是这样通过一个死循环挂起的。目的就是当我们运行起程序之后，可以不停地，不间断地一直处理队列中的消息数据。之前在学习 Swoole 时，另外如果你学习过 Go 语言的话，也会发现它们的 Http 服务中也是有类似的死循环代码来实现服务端挂起的。这个大家可以到我的 Swoole 系列中看看哦。
测试一下吧，运行一下生产者代码。
代码语言：javascript复制➜ source git:(master) ✗ php 2.rq.p.php
生产者向消息队列中发送信息：Hello World！%
执行结束了，只是输出了一句话，没啥别的效果。那么我们再来运行一下消费者代码。
代码语言：javascript复制> php 2.rq.c.php
等待消息，或者使用 Ctrl C 退出程序。
接收到数据： Hello World!
可以看到，消费者先是输出了接收到的数据，这个数据其实是上一步我们运行生产者插入到队列中的数据。现在，这条数据打印出来了，其实就是相当于已经被我们消费了。当然，在实际业务中，你可能会对这些数据进行更复杂的业务操作。但在演示时，我这里只是打印了一下。然后，消费者会继续挂在这里等待下一条消息的到来。这时，你可以再次运行生产者代码，然后就会看到消费者这边直接就已经消费了。
Redis 实现
对于 Redis 的实现，其实非常简单，我们之前也已经学过的，那就是使用 List 这个数据结构。先来看生产者，直接 push 数据就好了。
代码语言：javascript复制// 2.rs.p.php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
echo "生产者向消息队列中发送信息：Hello World！";
$redis->lpush('hello', 'Hello World！');
消费者呢？当然就是我们之前已经学过的 pop 啦。
代码语言：javascript复制// 2.rs.p.php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
echo "等待消息，或者使用 Ctrl C 退出程序。", PHP_EOL;
while(1){
$data = $redis->rpop('hello');
if ($data){
echo '接收到数据： ', $data, PHP_EOL;
}
}
在这里，我们使用的是 lpush() rpop() 的方式，当然你也可以使用 rpush() lpop() 的方式，大家思考一下，如果是 lpush() lpop() ，是实现的什么数据结构呢？
同样地，在 Redis 的消费者中，我们也需要通过一个死循环挂起消费者，然后不停地获取数据进行处理。剩下的测试过程就和上面的 RabbitMQ 一样了。
我的实践
之前我就说过，我的消息队列实践不多。唯一的业务场景下实现的高并发消息队列应用其实就是上面的 Redis 这两段代码。真的，核心就是这点东西。但为了抗高并发，我是使用的 Swoole ，生产者是在 Hyperf 框架中通过控制器接收到数据后，直接就放到 Redis 里。然后消费者就是一个命令行，接着开 100 个协程，将获取到的消息数据丢给协程处理。
业务应用是游戏的日志上报，最高并发 20000 ，入库日志量 3000万 。使用的是阿里云最便宜的那个 Redis 服务，4G 大小单实例的那款。
是不是见到了消息队列的恐怖能力。这个量级，对于任何消息队列应用来说问题都不大，RabbitMQ 被认为是比较慢的，但是，它的处理能力是每秒几万次请求。而 Redis ，就是以 Redis 的读写性能为基础的，大概每秒11万的读和8万的写。这个在之前的 Redis 学习中都已经说过了。
总结
今天通过代码，我们其实就已经学习到了整个消息队列中最核心的内容。没错，消息队列就是这么地简单，但又这么地实用。我的业务例子其实是异步解耦的一种实现。而对于另一种常见的场景，秒杀，大家想一想，是不是也可以直接通过这样一个简单的队列就能够实现了。当然，可能还比较简陋，也需要考虑更多的东西，但是，一秒内处理几万条请求和一秒内让几万条请求入队，这个差别可大了去了。
其实，从队列的思想就可以看出，我们用数据库也可以实现队列，插入数据是入队，然后倒序查询出来一条就可以视为出队。但是呢，数据库的性能往往和专业的消息队列以及 NoSQL 工具都是有很大的差距的。因此，其实还是那句话，把握本质和思想，工具用啥都好说。
测试代码：
https://github.com/zhangyue0503/dev-blog/blob/master/消息队列/source/2.rq.c.php
https://github.com/zhangyue0503/dev-blog/blob/master/消息队列/source/2.rq.c.php
https://github.com/zhangyue0503/dev-blog/blob/master/消息队列/source/2.rq.c.php
https://github.com/zhangyue0503/dev-blog/blob/master/消息队列/source/2.rq.c.php
