07 | Swoole 源码分析之 Channel 通道模块

原文首发链接: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.mzph.cn/news/798337.shtml

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

相关文章

Open3D (C++) 计算点云的特征值特征向量

目录 一、算法原理二、代码实现三、结果展示本文由CSDN点云侠原创,原文链接。如果你不是在点云侠的博客中看到该文章,那么此处便是不要脸的爬虫与GPT。 一、算法原理 针对整个点云 P = { p i } i

华为USG6000v

1、安全区域 一个及或多个接口的集合 默认的安全区域 Trust --- 优先级85&#xff0c;一般连接内网 Untrust --- 优先级5&#xff0c; 一般连接外网 Dmz --- 优先级50&#xff0c;一般连接服务器、 Local --- 优先级100&#xff0c;防火墙接口所在区的区域 2…

2024.4.7作业

//登陆界面 this->setWindowTitle("传奇霸业"); this->setWindowIcon(QIcon("C:\\Users\\l1693\\Desktop\\pictures\\1.png")); this->resize(400,300); this->setFixedSize(400,300); //登录界面修饰 //底图 QLabel *lab5 new QLabel(this);…

揭秘!接口自动化测试应该做什么?

在软件开发过程中&#xff0c;接口测试是一个至关重要的环节&#xff0c;它确保了系统或组件之间的数据交换、传递和控制管理过程以及相互逻辑依赖关系的正确性。传统的瀑布软件流程中&#xff0c;测试人员在做某个系统的手工功能测试时&#xff0c;会首先从业务人员或开发人员…

16个Python接单平台,做私活爽歪歪!(附100个爬虫源码)

一、python爬虫是可以做副业的&#xff0c;主要是爬取网站、小程序或者APP的数据&#xff0c;对数据进行分析与处理&#xff0c;或者直接向客户提供爬虫程序与技术支持。 当初学会Python那会儿&#xff0c;有朋友来介绍我去接私活&#xff0c;是为一家公司做网站&#xff0c;那…

background背景图参数边渐变CSS中创建背景图像的渐变效果

效果:可以看到灰色边边很难受,希望和背景融为一体 原理: 可以使用线性渐变&#xff08;linear-gradient&#xff09;或径向渐变&#xff08;radial-gradient&#xff09;。以下是一个使用线性渐变作为背景图像 代码: background: linear-gradient(to top, rgba(255,255,255,0)…

Collection与数据结构 Stack与Queue(二):队列与Queue

1. 队列 1.1 概念 只允许在一端进行插入数据操作&#xff0c;在另一端进行删除数据操作的特殊线性表&#xff0c;队列具有先进先出FIFO(First In First Out) 入队列&#xff1a;进行插入操作的一端称为队尾&#xff08;Tail/Rear&#xff09; 出队列&#xff1a;进行删除操作…

C语言分支语句

一、什么是语句 C语句可分为以下五类&#xff1a; 表达式语句 函数调用语句 控制语句 复合语句 空语句 本周后面介绍的是控制语句。 控制语句用于控制程序的执行流程&#xff0c;以实现程序的各种结构方式&#xff0c;它们由特定的语句定义符组成&#xff0c;C语 言有…

android 资源文件混淆

AGP7.0以上引用AndResGuard有坑 记录下 在项目的build.gradle中添加如下 buildscript {ext.kotlin_version "1.4.31"repositories {google()jcenter()maven {url "https://s01.oss.sonatype.org/content/repositories/snapshots/"}}dependencies {class…

JQuery(二)---【使用JQuery对HTML、CSS进行操作】

零.前言 JQuery(一)---【JQuery简介、安装、初步使用、各种事件】-CSDN博客 一.使用JQuery对HTML操作 1.1获取元素内容、属性 使用JQ可以操作元素的“内容” text()&#xff1a;设置或返回元素的文本内容html()&#xff1a;设置或返回元素的内容(包括HTML标记)val()&#…

Flask Python Flask-SQLAlchemy中数据库的数据类型、flask中数据可的列约束配置

Flask Python Flask-SQLAlchemy中数据库的数据类型、flask中数据可的列约束配置 SQLAlchemy官方文档地址实战的代码分享数据类型列约束配置自定义方法 SQLAlchemy官方文档地址 SQLAlchemy官方文档地址 实战的代码分享 Flask-SQLAlchemy框架为创建数据库的实例提供了一个基类…

Android Studio学习16——Activity跳转时的参数传递

传递数据——example 传递对象类型的数据——example 传递 接收 回传数据——example

PicGo + Gitee + VsCode - 搭建私人图床

文章目录 前言搭建图床VsCode 安装插件安装 PicGo准备 Gitee 图床测试 尾声 前言 本人是一个重度 vimer&#xff0c;并且喜欢客制化一些东西… Typora 固然好用&#xff0c;但不支持 vim…发现 vscode 中既可以使用 vim&#xff0c;也可以 md&#xff0c;用起来比较舒服.因此…

自动驾驶定位算法:基于多传感器融合的状态估计(muti-Sensors Fusion)

自动驾驶定位算法&#xff1a;基于多传感器融合的状态估计(muti-Sensors Fusion) 附赠自动驾驶学习资料和量产经验&#xff1a;链接 1、传感器(Sensor)选取 自动驾驶系统中用于状态估计(State Estimation)的常用传感器包括GPS/GNSS、IMU、激光雷达(Lidar)。 状态估计(State E…

Android JNI调试总结

1、确保NDK和CMake已经安装 新建能编译APK的工程&#xff0c;工程中添加相关ndk目录 2、添加C模块 添加完成后&#xff0c;工程目录自动更新&#xff0c;build.gradle导入了so编译器 修改build.gradle中添加相关gcc编译器如下 externalNativeBuild { cmake { abiFilters a…

python学习25:python中的元组(tuple)

python中的元组(tuple) 1.什么是元组&#xff1f; 元组也是容器数据类型的一种&#xff0c;同列表几乎是一样的&#xff0c;都是可以在里面封装多个&#xff0c;不同类型的元素在内&#xff1b;与列表最大的不同就是&#xff1a; 元组一旦被定义&#xff0c;就不能修改 2.元组…

头盔检测 | 基于Caffe-SSD目标检测算法实现的建筑工地头盔检测

项目应用场景 面向建筑工地头盔检测场景&#xff0c;使用深度学习 Caffe SSD 目标检测算法&#xff0c;基于 C 实现。 项目效果 项目细节 > 具体参见项目 README.md (1) 安装 Caffe SSD(2) 执行训练 sh examples/Hardhat/SSD300/train_SSD300.sh (3) 部署算法 项目获取 h…

vitepress系列-05-其他优化设置

其他优化设置 设置底部上一页和下一页 设置&#xff1a; import { defineConfig } from vitepress// https://vitepress.dev/reference/site-config export default defineConfig({lang: en-US,title: "东东爱编码的技术博客",description: "记录日常学习点点…

IO流

一、IO概述 1&#xff0e;什么是IO流? 存储和读取数据的解决方案l: inputo: output流∶像水流一样传输数据 2.IO流的作用? 用于读写数据&#xff08;本地文件&#xff0c;网络) 3.IO流按照流向可以分类哪两种流? 输出流:程序 - > 文件 输入流:文件 - > 程…

布隆过滤器详解及java实现

什么是布隆过滤器&#xff1f; 布隆过滤器&#xff08;Bloom Filter&#xff09;是一种数据结构&#xff0c;用于判断一个元素是否属于一个集合。它的特点是高效地判断一个元素是否可能存在于集合中&#xff0c;但是存在一定的误判率。 布隆过滤器的基本原理是使用一个位数组…