1. 云栖社区>
  2. PHP教程>
  3. 正文

CURL多线程请求

作者:用户 来源:互联网 时间:2017-11-30 10:59:18

curl线程请求

CURL多线程请求 - 摘要: 本文讲的是CURL多线程请求,CURL_MULTI基本资料 使用步骤调用这些函数实现目的的步骤一般如下: 第一步:调用curl_multi_init初始化一个curl批处理句柄资源 第二步:循环调用curl_multi_add_handle向curl批处理会话中添加单独

CURL_MULTI基本资料

CURL多线程请求-


使用步骤

调用这些函数实现目的的步骤一般如下:


第一步:调用curl_multi_init初始化一个curl批处理句柄资源


第二步:循环调用curl_multi_add_handle向curl批处理会话中添加单独的curl句柄资源(这一步需要注意的是,curl_multi_add_handle的第二个参数是由curl_init而来的子handle)


第三步:持续调用curl_multi_exec解析curl批处理句柄


第四步:根据需要循环调用curl_multi_getcontent返回获取的输出的文本流以获取结果


第五步:调用curl_multi_remove_handle移除curl批处理句柄资源中的某个句柄资源,并为每个handle调用curl_close


第六步:调用curl_multi_close关闭批处理句柄资源



另外需要注意的是PHP 5 版本才可以使用这个函数,必须开启 curl 扩展(打开 php.ini 把;extension=php_curl.dll前面的分号去掉 , 重启apache就可以使用了)。



CURL_MULTI封装的高并发请求类
/core/curl.php
<?php
/**
* ChromeMozilla/5.0 (Windows NT 6.1) AppleWebKit/536.11 (KHTML, like Gecko) Chrome/20.0.1132.47 Safari/536.11
* IE6 Mozilla/5.0 (Windows NT 6.1; rv:9.0.1) Gecko/20100101 Firefox/9.0.1
* FFMozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1; CIBA; .NET CLR 2.0.50727)
*/
class CURL{
const ITEM_URL=0;
const ITEM_P=1;
const ITEM_F=2;
const ITEM_TRYED=3;
const ITEM_FP=4;
const ITEM_P_OPT=5;
//线程的限制
public $limit=30;
//try time(s) before curl failed
public $maxTry=2;
//用户定义的选择
public $opt=array();
//缓存选项
public $cache=array('on'=>false,'dir'=>null,'expire'=>86400);
//任务回调,if taskpool is empty,this callback will be called,you can call CUrl::add() in callback
public $task=null;
//the real multi-thread num 真正的多线程数
private $activeNum=0;
//队列中已完成的任务
private $queueNum=0;
//finished task number,include failed task and cache
private $finishNum=0;
//The number of cache hit
private $cacheNum=0;
//completely failed task number
private $failedNum=0;
//已经添加任务数
private $taskNum=0;
//保存所有添加的任务
private $taskPool=array();
//running task(s)
private $taskRunning=array();
//failed task need to retry
private $taskFailed=array();
//total downloaded size,byte
private $traffic=0;
//handle of multi-thread curl
private $mh=null;
//time multi-thread start
private $startTime=null;
/**
* running infomation
* 运行信息
*/
function status($debug=false){
if($debug==='reset'){
$this->taskNum=0;
$this->finishNum=0;
$this->cacheNum=0;
$this->activeNum=0;
$this->taskRunning=array();
$this->queueNum=0;
$this->failedNum=0;
$this->traffic=0;
$this->taskPool=array();
$last=0;
$strlen=0;
return;
}
if($debug){
$s="finish:".($this->finishNum).'('.$this->cacheNum.')';
$s.="task:".$this->taskNum;
$s.="active:".$this->activeNum;
$s.="running:".count($this->taskRunning);
$s.="queue:".$this->queueNum;
$s.="failed:".$this->failedNum;
$s.="taskPool:".count($this->taskPool);
echo $s."/n";
}else{
static $last=0;
static $strlen=0;
$now=time();
//update status every 1 minute or all task finished
$msg='';
if($now>$last or ($this->finishNum==$this->taskNum)){
$last=$now;
$timeSpent=$now-$this->startTime;
if($timeSpent==0)
$timeSpent=1;
//percent
$s=sprintf('%-.2f%%',round($this->finishNum/$this->taskNum,4)*100);
//num
$s.=sprintf('%'.strlen($this->finishNum).'d/%-'.strlen($this->taskNum).'d(%-'.strlen($this->cacheNum).'d)',$this->finishNum,$this->taskNum,$this->cacheNum);
//speed
$speed=($this->finishNum-$this->cacheNum)/$timeSpent;
$s.=sprintf('%-d',$speed).'/s';
//net speed
$suffix='KB';
$netSpeed=$this->traffic/1024/$timeSpent;
if($netSpeed>1024){
$suffix='MB';
$netSpeed/=1024;
}
$s.=sprintf('%-.2f'.$suffix.'/s',$netSpeed);
//total size
$suffix='KB';
$size=$this->traffic/1024;
if($size>1024){
$suffix='MB';
$size/=1024;
if($size>1024){
$suffix='GB';
$size/=1024;
}
}
$s.=sprintf('%-.2f'.$suffix,$size);
//estimated time of arrival
if($speed==0){
$str='--';
}else{
$eta=($this->taskNum-$this->finishNum)/$speed;
$str=ceil($eta).'s';
if($eta>3600){
$str=ceil($eta/3600).'h'.ceil(($eta%3600)/60).'m';
}elseif($eta>60){
$str=ceil($eta/60).'m'.($eta%60).'s';
}
}
$s.='ETA '.$str;
$msg=$s;
}
return $msg;
}
}
/**
* read interface
*/
function __get($name){
return $this->$name;
}
/**
* single thread
*
* @param mixed $url
* @return mixed curl_exec() result
*/
function read($url){
$r=array();
$ch=$this->init($url);
$content=curl_exec($ch);
if(curl_errno($ch)===0){
$r['info']=curl_getinfo($ch);
$r['content']=$content;
}else{
debug_print('error: code '.curl_errno($ch).", ".curl_error($ch),E_USER_WARNING);
}
return $r;
}
/**
* add a task to taskPool
*
* @param array $url $url[0] is url,$url[1] is file path if isset,$url[2] is curl option
* @param array $p 成功时回调,$p[0]是回调的函数,$p[1]是回调函数的参数
* @param array $f 失败时回调,$f[0]是回调的函数,$f[1]是回调函数的参数
*/
function add($url=array(),$p=array(),$f=array()){
//check
if(!is_array($url) or empty($url[0])){
var_dump($url);
debug_print('url is invalid',E_USER_ERROR);
}
if(!is_array($p) or !is_array($f))
debug_print('callback is not array',E_USER_ERROR);
if(!isset($p[0]))
debug_print('process callback is not set',E_USER_ERROR);
if((isset($p[1]) and !is_array($p[1])) or (isset($f[1]) and !is_array($f[1]))){
debug_print('callback function parameter must be an array',E_USER_ERROR);
}
//fix
if(empty($url[1]))
$url[1]=null;
if(empty($url[2]))
$url[2]=null;
if(!isset($p[1]))
$p[1]=array();
if(isset($f[0]) and !isset($f[1]))
$f[1]=array();
$task=array();
$task[self::ITEM_URL]=$url;
$task[self::ITEM_P]=$p;
$task[self::ITEM_P_OPT]=$url[2];
$task[self::ITEM_F]=$f;
$task[self::ITEM_TRYED]=0; //try times befroe complete failure
$task[self::ITEM_FP]=null; //file handler for file download
$this->taskPool[]=$task;
$this->taskNum++;
}
/**
* Perform the actual task(s).
* 执行实际任务
*/
function go(){
static $running=false;
if($running)
debug_print('CURL can only run one instance',E_USER_ERROR);
$this->mh=curl_multi_init();
for($i=0;$i<$this->limit;$i++)
$this->addTask();
$this->startTime=time();
$running=true;
do{
$this->exec();
//主要用于阻断curl_multi_select
curl_multi_select($this->mh);
while($curlInfo = curl_multi_info_read($this->mh,$this->queueNum)){
$ch=$curlInfo['handle'];
$info=curl_getinfo($ch);
$this->traffic+=$info['size_download'];
$k=(int)$ch;
$task=$this->taskRunning[$k];
if(empty($task)){
debug_print("can't get running task",E_USER_WARNING);
}
$callFail=false;
if($curlInfo['result']==CURLE_OK){
if(isset($task[self::ITEM_P])){
array_unshift($task[self::ITEM_P][1],array(
'info'=>$info,
'content'=>curl_multi_getcontent($ch),
));
}
}else{
if($task[self::ITEM_TRYED] >= $this->maxTry){
$msg='curl error '.$curlInfo['result'].', '.curl_error($ch).', '.$info['url'];
if(isset($task[self::ITEM_F][0])){
array_unshift($task[self::ITEM_F][1],$msg);
$callFail=true;
}else{
echo $msg."/n";
}
$this->failedNum++;
}else{
$task[self::ITEM_TRYED]++;
$this->taskFailed[]=$task;
$this->taskNum++;
}
}
curl_multi_remove_handle($this->mh,$ch);
curl_close($ch);
unset($this->taskRunning[$k]);
$this->finishNum++;
if($curlInfo['result']==CURLE_OK){
call_user_func_array($task[self::ITEM_P][0],$task[self::ITEM_P][1]);
}elseif($callFail){
call_user_func_array($task[self::ITEM_F][0],$task[self::ITEM_F][1]);
}
$this->addTask();
//so skilful,if $this->queueNum grow very fast there will be no efficiency lost,because outer $this->exec() won't be executed.
$this->exec();
}
}while($this->activeNum || $this->queueNum || !empty($this->taskFailed) || !empty($this->taskRunning) || !empty($this->taskPool));
unset($this->startTime);
curl_multi_close($this->mh);
$running=false;
}
/**
* curl_multi_exec()
* 运行当前 cURL 句柄的子连接
*/
private function exec(){
while(curl_multi_exec($this->mh, $this->activeNum)===CURLM_CALL_MULTI_PERFORM){}
}
/**
* add a task to curl
* 添加一个任务到curl
*/
private function addTask(){
$c=$this->limit-count($this->taskRunning);
while($c>0){
$task=array();
//search failed first
if(!empty($this->taskFailed)){
$task=array_pop($this->taskFailed);
}else{
if(!empty($this->taskPool))
$task=array_pop($this->taskPool);
}
if(!empty($task)){
$ch = '';
$ch=$this->init($task[self::ITEM_URL][0]);
if(is_resource($ch)){
//单curl任务选项
if(isset($task[self::ITEM_P_OPT])){
foreach($task[self::ITEM_P_OPT] as $k=>$v)
curl_setopt($ch,$k,$v);
curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false); // 跳过证书检查
curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, true);// 从证书中检查SSL加密算法是否存在
}
curl_multi_add_handle($this->mh,$ch);
$this->taskRunning[(int)$ch]=$task;
}else{
debug_print('$ch is not resource,curl_init failed.',E_USER_WARNING);
}
}
$c--;
}
}
/**
* set or get file cache
* 设置或获取文件缓存
* @param mixed $key
* @param mixed $content
* @return return content or false if read,true or false if write
*/
private function cache($url,$content=null){
$key=md5($url);
if(!isset($this->cache['dir']))
debug_print('Cache dir is not defined',E_USER_ERROR);
$dir=$this->cache['dir'].DIRECTORY_SEPARATOR.substr($key,0,3);
$file=$dir.DIRECTORY_SEPARATOR.substr($key,3);
if(!isset($content)){
if(file_exists($file)){
if((time()-filemtime($file)) < $this->cache['expire']){
return unserialize(file_get_contents($file));
}else{
unlink($file);
}
}
}else{
$r=false;
//检查主目录是否存在
if(!is_dir($this->cache['dir'])){
debug_print("Cache dir doesn't exists",E_USER_ERROR);
}else{
$dir=dirname($file);
if(!file_exists($dir) and !mkdir($dir,0777))
debug_print("Create dir failed",E_USER_WARNING);
$content=serialize($content);
if(file_put_contents($file,$content,LOCK_EX))
$r=true;
else
debug_print('Write cache file failed',E_USER_WARNING);
}
return $r;
}
}
private function init($url){
$ch=curl_init();
$opt=array();
$opt[CURLOPT_URL]=$url;
$opt[CURLOPT_HEADER]=false;
$opt[CURLOPT_CONNECTTIMEOUT]=15;
$opt[CURLOPT_TIMEOUT]=300;
// $opt[CURLOPT_AUTOREFERER]=true;
// $opt[CURLOPT_USERAGENT]='Mozilla/5.0 (Windows NT 6.1) AppleWebKit/536.11 (KHTML, like Gecko) Chrome/20.0.1132.47 Safari/536.11';
$opt[CURLOPT_RETURNTRANSFER]=true;
// $opt[CURLOPT_FOLLOWLOCATION]=true;
$opt[CURLOPT_MAXREDIRS]=10;
//user defined opt
if(!empty($this->opt))
foreach($this->opt as $k=>$v)
$opt[$k]=$v;
curl_setopt_array($ch,$opt);
return $ch;
}
}
使用方法
index.php
//使用要求,需要php5.3版本以上
//$access_token和$openid_array的获取省略
define('ROOT_PATH', dirname(__FILE__));
require_once ROOT_PATH.'/core/curl.php';
$curl = new CURL();
$curl_url = "https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token=".$access_token;
foreach($openid_array as $openid){
$curl_post = charsetToUTF8('{"touser":"'.$openid.'","msgtype":"text","text":{"content":"你好"}}');
$url = array($curl_url,null,array(
CURLOPT_POST=>1,
CURLOPT_POSTFIELDS=>$curl_post,
CURLOPT_HTTPHEADER=>array("application/json;charset=utf-8",'Content-Length: '.strlen($curl_post)),
));
$curl->add(
$url,
array('cb',array($openid)),//请求成功的回调函数
array('cb',array($openid))//请求失败的回调函数
);
}
$curl->go();//开始执行
}
//$curl回调函数
function cb($res,$val){
//$res 是指请求相关信息 $res['content']是指请求返回的结果
//$val 是调用时传入的参数 (ex:foreach中传入的$openid)
}
目录结构
| pro 项目目录
| -- | index.php
| -- | core | curl.php

以上是云栖社区小编为您精心准备的的内容,在云栖社区的博客、问答、公众号、人物、课程等栏目也有的相关内容,欢迎继续使用右上角搜索按钮进行搜索curl , 线程 请求 ,以便于您获取更多的相关知识。

弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率

40+云计算产品,6个月免费体验

稳定可靠、可弹性伸缩的在线数据库服务,全球最受欢迎的开源数据库之一

云服务器9.9元/月,大学必备