基础简单消息队列应用

在上一课中，我们已经学习到了什么是消息队列，有哪些消息队列，以及我们会用到哪个消息队列。今天，就直接进入主题，学习第一种，最简单，但也是最常用，最好用的消息队列模式。

最简单的队列功能

最简单的队列功能，无非就是将我们在数据结构与算法中学过的那个队列结构，变成一个外部功能组件。让各种语言和各种应用程序都可以通过这个队列来进行数据操作。这样的一个队列系统就称之为“消息队列中间件”。

一般，我们会将生产消息的程序，或者说，将数据放入到队列的一方称为 P （生产者，Producer）；然后将队列称为Q（Queue）；最后，将守候在队列前，等待从队列中获取数据的应用、程序或者代码段称为 C（消费者/客户端，Consumer）。

这是 RabbitMQ 官网手册上的图，后面的相关图示我们也将直接使用它们的。在这个图中，有字母的部分就不多解释了。中间红色的一格一格的部分代表的就是 Q 。

通常来说，P 只管将数据放到 Q 中，Q 负责中间存储数据，然后 C 取出数据。数据的进出方式遵循经典数据结构队列中的规则，也就是 FIFO 先进先出。

是不是就和我们之前说过的一样，最简单的理解，它就是将队列这个数据结构抽成了一个第三方组件。这样就可以跨平台、跨应用、跨语言地进行数据存取了。

RebbitMQ 实现

好了，先来看 RabbitMQ 的实现。你需要先安装好 RabbitMQ ，我这里是使用的 Docker 安装的。

docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management

然后，也要安装好 PHP 的 Composer 组件，用于操作 RabbitMQ 消息队列，PHP 需要开启 sockets 扩展。

composer require php-amqplib/php-amqplib

5672 是 RabbitMQ 的服务端口，15672 则是它自带的一个管理工具的访问端口。具体的内容大家可以到官方文档中进行更加深入的学习。当然，也可以使用虚拟机方式来搭建测试环境，这个大家看自己的喜好吧。

接下来，我们先实现 P 端的代码，也就是生产者向消息队列中添加数据。

// 2.rq.p.php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// 建立连接
$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 是使用这个协议就好了，而且它也支持其它的一些协议。发送完消息之后，记得关闭连接哦。

好了，接下来是我们的消费者/客户端实现。

// 2.rq.c.php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

// 建立连接
$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 系列中看看哦。

测试一下吧，运行一下生产者代码。

➜  source git:(master) ✗ php 2.rq.p.php 
生产者向消息队列中发送信息：Hello World！%

执行结束了，只是输出了一句话，没啥别的效果。那么我们再来运行一下消费者代码。

> php 2.rq.c.php 
等待消息，或者使用 Ctrl+C 退出程序。
接收到数据： Hello World!

可以看到，消费者先是输出了接收到的数据，这个数据其实是上一步我们运行生产者插入到队列中的数据。现在，这条数据打印出来了，其实就是相当于已经被我们消费了。当然，在实际业务中，你可能会对这些数据进行更复杂的业务操作。但在演示时，我这里只是打印了一下。然后，消费者会继续挂在这里等待下一条消息的到来。这时，你可以再次运行生产者代码，然后就会看到消费者这边直接就已经消费了。

Redis 实现

对于 Redis 的实现，其实非常简单，我们之前也已经学过的，那就是使用 List 这个数据结构。先来看生产者，直接 push 数据就好了。

// 2.rs.p.php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

echo "生产者向消息队列中发送信息：Hello World！";

$redis->lpush('hello', 'Hello World！');

消费者呢？当然就是我们之前已经学过的 pop 啦。

// 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/%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/source/2.rq.c.php

https://github.com/zhangyue0503/dev-blog/blob/master/%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/source/2.rq.c.php

https://github.com/zhangyue0503/dev-blog/blob/master/%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/source/2.rq.c.php

https://github.com/zhangyue0503/dev-blog/blob/master/%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97/source/2.rq.c.php