[PHP知识库]什么是 swoole channel?

channel?可以理解为消息队列，只不过是协程间的消息队列，多个协程通过?push?和?pop?操作生产消息和消费消息，用来协程之间的通讯。需要注意的是?channel?是没法跨进程的，只能一个?Swoole?进程里的协程间通讯，最典型的应用是连接池和并发调用。

<?php


/***
 *
 *
 *
 * 此功能请求对方接口，使用协程方法，比如说发送微信模板消息。php-fpm 模式下， 只能一个一个发送。遇到阻塞 就不能全部发送。
 *
 *
 * 使用swoole 就可以直接全部发送。节省时间
 */
use Swoole\Coroutine;
use Swoole\Coroutine\WaitGroup;
use Swoole\Coroutine\Http\Client;
use function Swoole\Coroutine\run;

echo 'start_at: '. date('Y-m-d H:i:s') ."\r\n";
run(function () {
    $wg = new WaitGroup();
    $result = [];
    //启动第一个协程
    $openids = [1,2,3,4,5,6,7,8,9,10];
    $params= [
        'openids'       => json_encode($openids),
        'push_content'  => 111,
        'url'           => 'https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=',
        'func'          => 'wx_push',
    ];
    $wg = new \Swoole\Coroutine\WaitGroup();
    $result = []; // 结果信息
    $result['start_at'] = microtime(true); // 记录开始时间
    $result['num'] = 0; // 记录推送数量
    switch ($params['func']) {
        case 'wx_push':
            // 数据验证
            if (!isset($params['openids']) || !$params['openids'] ||
                !isset($params['push_content']) || !$params['push_content'] ||
                !isset($params['url']) || !$params['url']
            ) {
                $result['params'] = $params;
                $result['err_msg'] = '参数异常！';
                break;
            }
            $url_arr = parse_url($params['url']); // 解析url
            $params['openids'] = is_array($params['openids']) ? $params['openids'] : json_decode($params['openids'], 1); // 推送的用户ID 列表
            $content = [];  // 推送的内容
            foreach ($params['openids'] as $k => $openid) {
                // 一次推送3000条，推送完成再推下一轮3000条
                if ($k > 0 && $k % 3 == 0) {
                    //挂起当前协程，等待所有任务完成后恢复
                    $wg->wait();
                }
                $wg->add();
                $result['num']++; // 记录推送数量
                //启动第一个协程
                go(function () use ($wg, &$result, $openid, $content, $url_arr,$k) {
                    $content['touser'] = $openid;
                    if (is_array($content)) {
                        $content = json_encode($content,JSON_UNESCAPED_UNICODE);
                    }
                    //启动一个协程客户端client，请求淘宝首页
                    $cli = new \Swoole\Coroutine\Http\Client('cy.huazhanhao.top', 82);
                    $cli->setHeaders([
                        //'Host' => 'api.weixin.qq.com',
                        'Host' => 'cy.huazhanhao.top',
                        'Content-Length' => strlen($content),
                        'Content-type' => 'application/x-www-form-urlencoded', //'application/x-www-form-urlencoded',
                    ]);
                    $cli->set(['timeout' => 6]); // 设置请求超时时长
                    // 拼接请求地址和参数
                    $cli->post('/redis_mq/curl.php', $content);
                    echo $cli->body . "\r\n";
                    echo "第".$k.'轮';
                    $result['body'][$openid] = $cli->body; // 获取返回结果信息
                    $cli->close();
                    $wg->done();
                });
            }
            break;
        case 'ali':
            break;
        default:
            $result['default'] = 'no function';
            break;
    }

    // 挂起当前协程，等待所有任务完成后恢复
    $wg->wait();
    $result['end_at'] = microtime(true); // 记录结束时间
    $result['use_time'] = $result['end_at'] - $result['start_at']; // 记录结束时间
    //这里 $result 包含了 2 个任务执行结果
    var_dump($result);
});

开启N个协程，每次执行N个数据包请求

• add?方法增加计数
• done?表示任务已完成
• wait?等待所有任务完成恢复当前协程的执行
• WaitGroup?对象可以复用，add、done、wait?之后可以再次使用