• 理解消息队列原来这么简单
  • 发布于 2个月前
  • 125 热度
    0 评论
  • Spring
  • 0 粉丝 33 篇博客
  •   
基础简单消息队列应用
在上一课中,我们已经学习到了什么是消息队列,有哪些消息队列,以及我们会用到哪个消息队列。今天,就直接进入主题,学习第一种,最简单,但也是最常用,最好用的消息队列模式。

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

一般,我们会将生产消息的程序,或者说,将数据放入到队列的一方称为 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 工具都是有很大的差距的。因此,其实还是那句话,把握本质和思想,工具用啥都好说。
用户评论