Swoole 源码分析之 Channel 通道模块

news/发布时间2024/5/10 14:17:55

原文首发链接：Swoole 源码分析之 Channel 通道模块
大家好，我是码农先森。

引言

通道，用于协程间通讯，支持多生产者协程和多消费者协程。底层自动实现了协程的切换和调度。

通道与 PHP 的 Array 类似，仅占用内存，没有其他额外的资源申请，所有操作均为内存操作，无 IO 消耗。

底层使用 PHP 引用计数实现，无内存拷贝。即使是传递巨大字符串或数组也不会产生额外性能消耗 channel 基于引用计数实现，是零拷贝的。

源码拆解

Channel 通道需要在协程环境中使用，我们先看下面这段代码，使用 new Channel(1) 创建一个 channel 对象，然后在第一个协程中向通道中推送数据，在第二个协程获取到通道内的数据进行消费。

use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
use function Swoole\Coroutine\run;run(function(){// 创建 channel 通道对象$channel = new Channel(1);Coroutine::create(function () use ($channel) {for($i = 0; $i < 10; $i++) {Coroutine::sleep(1.0);// 向通道内推送数据$channel->push(['rand' => rand(1000, 9999), 'index' => $i]);echo "{$i}\n";}});Coroutine::create(function () use ($channel) {while(1) {// 从通道中获取数据$data = $channel->pop(2.0);if ($data) {var_dump($data);} else {assert($channel->errCode === SWOOLE_CHANNEL_TIMEOUT);break;}}});
});

在分析源代码之前，我们可以提前看一下源码整体的调用逻辑图，以便我们有个大致的印象。

这段代码主要是在 Swoole 的协程环境中创建 Channel 对象并初始化其容量的逻辑。

// swoole-src/ext-src/swoole-channel.cc:132
static PHP_METHOD(swoole_channel_coro, __construct) {zend_long capacity = 1;// 解析传入的参数ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 0, 1)Z_PARAM_OPTIONALZ_PARAM_LONG(capacity)ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);if (capacity <= 0) {capacity = 1;}// 当前对象对应的 ChannelObject 结构体指针ChannelObject *chan_t = php_swoole_channel_coro_fetch_object(Z_OBJ_P(ZEND_THIS));// 为该通道对象分配新的 Channel 实例，并设置其容量为传入的值。chan_t->chan = new Channel(capacity);zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("capacity"), capacity);
}

这段代码主要是在 Swoole 的协程环境中向通道中推送数据并对返回结果进行处理的逻辑。

// swoole-src/ext-src/swoole-channel.cc:149
static PHP_METHOD(swoole_channel_coro, push) {// 获取当前对象的 Channel 实例Channel *chan = php_swoole_get_channel(ZEND_THIS);zval *zdata;double timeout = -1;// 解析传入的参数ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 1, 2)Z_PARAM_ZVAL(zdata)Z_PARAM_OPTIONALZ_PARAM_DOUBLE(timeout)ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);Z_TRY_ADDREF_P(zdata);zdata = sw_zval_dup(zdata);// 向通道中推入数据if (chan->push(zdata, timeout)) {zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), Channel::ERROR_OK);RETURN_TRUE;} else {zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), chan->get_error());Z_TRY_DELREF_P(zdata);efree(zdata);RETURN_FALSE;}
}// swoole-src/coroutine/channel.cc:105
bool Channel::push(void *data, double timeout) {// 获取当前协程对象 current_coCoroutine *current_co = Coroutine::get_current_safe();// 如果通道已关闭if (closed) {// 设置错误并返回空指针error_ = ERROR_CLOSED;return false;}// 如果通道已满或生产者队列不为空，则设置超时消息，并根据传入的超时值添加定时器，等待生产者。if (is_full() || !producer_queue.empty()) {TimeoutMessage msg;msg.error = false;msg.timer = nullptr;if (timeout > 0) {msg.chan = this;msg.type = PRODUCER;msg.co = current_co;// 根据传入的超时值添加定时器msg.timer = swoole_timer_add(timeout, false, timer_callback, &msg);}// 挂起生产者协程yield(PRODUCER);// 如果设置了定时器，则在超时消息中删除定时器if (msg.timer) {swoole_timer_del(msg.timer);}// 如果当前协程被取消if (current_co->is_canceled()) {// 设置错误并返回空指针error_ = ERROR_CANCELED;return nullptr;}// 如果发生超时if (msg.error) {// 设置错误并返回空指针error_ = ERROR_TIMEOUT;return nullptr;}// 如果通道关闭且为空的情况if (closed && is_empty()) {// 设置相应的错误并返回空指针。error_ = ERROR_CLOSED;return nullptr;}}// 将数据压入数据队列。data_queue.push(data);swoole_trace_log(SW_TRACE_CHANNEL, "push data to channel, count=%ld", length());// 如果消费者队列不为空，则唤醒消费者协程。if (!consumer_queue.empty()) {Coroutine *co = pop_coroutine(CONSUMER);// 恢复消费者协程co->resume();}return true;
}

这段代码主要是在 Swoole 的协程环境中从通道中取出数据并对返回结果进行处理的逻辑。

// swoole-src/ext-src/swoole-channel.cc:175
static PHP_METHOD(swoole_channel_coro, pop) {// 获取当前对象的 Channel 实例Channel *chan = php_swoole_get_channel(ZEND_THIS);// 设置超时变量为-1double timeout = -1;// 解析一个超时参数ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 0, 1)Z_PARAM_OPTIONALZ_PARAM_DOUBLE(timeout)ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);// 从通道中取出数据，并返回一个 zval 指针zval *zdata = (zval *) chan->pop(timeout);// 如果返回的 zval 指针不为空if (zdata) {// 将其返回给 PHP 脚本，并释放内存RETVAL_ZVAL(zdata, 0, 0);efree(zdata);zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), Channel::ERROR_OK);} else {zend_update_property_long(swoole_channel_coro_ce, SW_Z8_OBJ_P(ZEND_THIS), ZEND_STRL("errCode"), chan->get_error());RETURN_FALSE;}
}// swoole-src/coroutine/channel.cc:55
void *Channel::pop(double timeout) {// 获取当前协程对象 current_coCoroutine *current_co = Coroutine::get_current_safe();// 如果通道已关闭且为空if (closed && is_empty()) {// 设置错误并返回空指针error_ = ERROR_CLOSED;return nullptr;}// 如果通道为空或者消费者队列不为空if (is_empty() || !consumer_queue.empty()) {TimeoutMessage msg;msg.error = false;msg.timer = nullptr;if (timeout > 0) {msg.chan = this;msg.type = CONSUMER;msg.co = current_co;// 根据传入的超时值添加定时器msg.timer = swoole_timer_add(timeout, false, timer_callback, &msg);}// 挂起消费者协程yield(CONSUMER);// 如果设置了定时器，则在超时消息中删除定时器if (msg.timer) {swoole_timer_del(msg.timer);}// 如果当前协程被取消if (current_co->is_canceled()) {// 设置错误并返回空指针error_ = ERROR_CANCELED;return nullptr;}// 如果发生超时if (msg.error) {// 设置错误并返回空指针error_ = ERROR_TIMEOUT;return nullptr;}// 如果通道关闭且为空的情况if (closed && is_empty()) {// 设置相应的错误并返回空指针。error_ = ERROR_CLOSED;return nullptr;}}// 从数据队列中弹出数据，并返回该数据。void *data = data_queue.front();data_queue.pop();// 如果生产者队列不为空，则唤醒生产者协程if (!producer_queue.empty()) {Coroutine *co = pop_coroutine(PRODUCER);// 恢复到生产者协程co->resume();}return data;
}

这段代码一是针对超时回调处理的处理逻辑，并恢复相关的协程操作。二是实现了协程的挂起操作，并根据不同的类型将当前协程放入不同的队列中，以便后续根据需要恢复执行。

// swoole-src/coroutine/channel.cc:22
void Channel::timer_callback(Timer *timer, TimerNode *tnode) {TimeoutMessage *msg = (TimeoutMessage *) tnode->data;msg->error = true;msg->timer = nullptr;if (msg->type == CONSUMER) {// 从消费者队列中移除该协程msg->chan->consumer_remove(msg->co);} else {// 从生产者队列中移除该协程msg->chan->producer_remove(msg->co);}// 恢复协程msg->co->resume();
}// swoole-src/coroutine/channel.cc:34
void Channel::yield(enum Opcode type) {// 获取当前协程Coroutine *co = Coroutine::get_current_safe();if (type == PRODUCER) {// 将当前协程放入到生产者队列producer_queue.push_back(co);swoole_trace_log(SW_TRACE_CHANNEL, "producer cid=%ld", co->get_cid());} else {// 将当前协程放入到消费者队列consumer_queue.push_back(co);swoole_trace_log(SW_TRACE_CHANNEL, "consumer cid=%ld", co->get_cid());}// 挂起被取消，则调用该函数Coroutine::CancelFunc cancel_fn = [this, type](Coroutine *co) {if (type == CONSUMER) {consumer_remove(co);} else {producer_remove(co);}co->resume();return true;};// 挂起当前协程co->yield(&cancel_fn);
}

总结

  1. Channel 通道需要在协程的环境中进行使用，通道是纯内存操作，没有 IO 消耗，非常高效。
  2. 底层使用 Channel::yield 函数实现了协程的自动切换和调度，如果通道处理超时则会自动调用 Channel::timer_callback 函数。
  3. Channel 通道是跨协程直接通信的一大利器，在实际的场景中使用起来十分的便利、高效。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.ulsteruni.cn/article/15154744.html

如若内容造成侵权/违法违规/事实不符，请联系编程大学网进行投诉反馈email:xxxxxxxx@qq.com，一经查实，立即删除！

相关文章

阿里云 函数计算 域名配置问题

阿里云 函数计算 域名配置问题

1 区域切到中国香港 再创建服务和函数，以省去备案域名的问题 2 云解析DNS/域名解析/解析设置 https://dns.console.aliyun.com/?spm=......#/dns/domainList 记录值设置到自己账号对应的公网地址（Endpoint）3 函数计算 FC/域名管理 关于为什么要2和3两个配置步骤 只配置2，…
阅读更多...
重装/安装Windows系统

重装/安装Windows系统

原文链接 安装前准备一个 32G 以上大小的空 U 盘 PE 系统： pe 就是一个微系统，可以在电脑没有装系统的情况下访问电脑的各种硬件。所以它也可以给磁盘分区，甚至实现破解电脑密码的功能，这个大家可以自行研究。 它还有一个优点，就是 pe 的安装只占用非常小的体积，大概 300…
阅读更多...
如何使用FastGateway自动申请免费的HTTPS证书？

如何使用FastGateway自动申请免费的HTTPS证书？

在购买域名的时候我相信很多人都遇到了对于证书的问题，之前我也是使用阿里云的免费一年的证书，那时候感觉还好，一年更换一次，但是近期阿里云对于证书的过期时间直接砍到了三个月！让我难以接受，所以我在想吧他直接集成到我的FastGateway中，让他自动申请，自动续期！下面我…
阅读更多...
Linux进程创建和管理

Linux进程创建和管理

在 Linux 中，进程创建和管理的相关函数主要是 fork()、exec()、wait() 和 exit()举个例子：#include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <sys/wait.h>int main() {pid_t pid;// 创建子进程pid = fork();if (pid < 0) {/…
阅读更多...
NU1605: 错误形式的警告: 检测到包降级的解决办法

NU1605: 错误形式的警告: 检测到包降级的解决办法

这两行的意思是需要我们升级Maui.Controls的版本在8.0.14，取高版本。同理，再次进行：最后：#####愿你一寸一寸地攻城略地，一点一点地焕然一新#####
阅读更多...
多线程（2）-线程同步条件变量

多线程（2）-线程同步条件变量

在 Linux 多线程编程中，条件变量是一种用于线程间同步的重要机制。它通常与互斥锁结合使用，用于解决多个线程竞争共享资源的问题。条件变量允许一个线程在等待某个条件变为真时阻塞，并且在另一个线程改变条件并通知时恢复执行。这个玩意跟内核等待队列差不多意思。在 Linux …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023