原文首发链接：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_OPTIONAL
Z_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_OPTIONAL
Z_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_co
Coroutine *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);
// 设置超时变量为-1
double timeout = -1;
// 解析一个超时参数
ZEND_PARSE_PARAMETERS_START_EX(ZEND_PARSE_PARAMS_THROW, 0, 1)
Z_PARAM_OPTIONAL
Z_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_co
Coroutine *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);
}
总结
Channel通道需要在协程的环境中进行使用，通道是纯内存操作，没有 IO 消耗，非常高效。- 底层使用
Channel::yield函数实现了协程的自动切换和调度，如果通道处理超时则会自动调用Channel::timer_callback函数。 Channel通道是跨协程直接通信的一大利器，在实际的场景中使用起来十分的便利、高效。
