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
之后可以再次使用