发表新帖

请大神来帮我看下我使用 swoole 群发数据的代码是否正确

### 问题描述
就是我想用swoole 做个批量群发 的服务器。
功能就是 我向微信的API 发送模板消息 给所以粉丝。

### Swoole版本，PHP版本，以及操作系统版本信息
swoole4.x版本
php 7.4
linux
4核 8G内存 10M带宽
### 相关代码
```php
<?php
require './communication.php'; //这个是封装的CURL方法
 
$serv	= new Swoole\Server('0.0.0.0', 9555);
 
//设置异步任务的工作进程数量
$serv->set([
    //配置此参数后将会启用 task 功能。所以 Server 务必要注册 onTask、onFinish 2 个事件回调函数。如果没有注册，服务器程序将无法启动。
   	'worker_num'=>4,
    'task_worker_num' => 10,
	'task_enable_coroutine'=>true,
	'enable_coroutine'=>true,
]);
$serv->PDO = new PDO('mysql:host=60.0.0.1;dbname=name;charset=utf8', 'username', 'ttNxMyj4scFneyES');
 
//此回调函数在worker进程中执行
$serv->on('Receive', function($serv, $fd, $reactor_id, $data) {
    //投递异步任务
	$account	= json_decode($data,true); 
	$page   	= $account['page'];
	$pdo		= $serv->PDO;
    $pageindex 	= max(intval($page), 1);
    $pagesize 	= $account['pagesize']; 
    $sql 		= 'SELECT openid FROM `ims_mc_mapping_fans` WHERE `uniacid`=:uniacid and `follow`=:follow ORDER BY fanid desc LIMIT '.(($pageindex -1) * $pagesize).','. $pagesize;
	$statement  = $pdo->prepare($sql);
    $statement->execute(array(':uniacid'=>$account['uniacid'],':follow'=>'1'));
    $swoole     = $statement->fetchall();
	$statement 	= null;
	$pdo		= null;
	foreach($swoole as $key=>$val){
		unset($swoole[$key][0]);
	}
	$account['fans'] = $swoole;
	$task_id = $serv->task($account);
    $serv->send($fd,'SUCCESS');
    echo "已接收worker:id={$task_id}\n";
});

//处理异步任务(此回调函数在task进程中执行)
$serv->on('Task', function ($serv,Swoole\Server\Task $task) {
	
	$account 	= $task->data;
	$page 		= $account['page'];
	$swoolecount= count($account['fans']);
	$result 			= array();
	$return_arr['no']	= array();
	$return_arr['ok']	= array();
	if($swoolecount>0){
		$chan 		= new chan($swoolecount);
		$chan->datar= array(
			'getAccessToken'=>$account['getAccessToken'],
			'template_id'=>$account['template_id'],
			'url'=>$account['url'],
			'postdata'=>$account['postdata']
		);
		foreach($account['fans'] as $key=>$val){
			$chan->val	=	$val;
			go(function () use ($chan) {
				$valdata = $chan->val;
				$obj_data= $chan->datar;
				$token 					= $obj_data['getAccessToken'];
				$datar 					= array();
				$datar['touser']        = $valdata['openid'];
				$datar['template_id']   = trim($obj_data['template_id']);
				$datar['url']           = trim($obj_data['url']);
				$datar['topcolor']      = trim('#FF683F');
				$datar['data']          = $obj_data['postdata'];
				$datar                  = json_encode($datar);
				$post_url               = "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={$token}";
				$response               = ihttp_request($post_url, $datar);
				$result                 = @json_decode($response['content'], true);
				if($result['errmsg']=='ok'){
					$as['ok'] = $valdata['openid'];
					$chan->push($as);
				}else{
					$as['no'] = $valdata['openid'];
					$chan->push($as);	
				}
			});
			$chan->val 	= null; 
		}; 
		for ($i = 0; $i < $swoolecount; $i++){
			$result[]= $chan->pop(); 
		}
		 
		foreach($result as $k=>$v){
			if($v['ok']!=''){
				$return_arr['ok'][] = $v['ok'];
			}
			if($v['no']!=''){
				$return_arr['no'][] = $v['no'];
			}
		}
		
	} 
	$result	= count($return_arr['ok']);
	$error	= count($return_arr['no']);
	echo "处理异步任务[id={$task_id}]-数据数量：{$swoolecount}-当前处理第{$page}页-成功发送{$result}条-失败{$error}条".PHP_EOL;

//返回任务执行的结果
	$query['result']		=	'10000';
	$query['messages']		=	'执行完成'; 
	$query['pagesize']		=	$account['pagesize'];
	$query['tid']			=	$account['tid']; 
	$query['page']			=	$account['page']; 
	$query['success']		=	$swoolecount;//查询数据
	$query['result']		=	$result; //执行数据
	$query['error']			=	$error; //执行数据
    $task->finish($query);
});

//处理异步任务的结果(此回调函数在worker进程中执行)
$serv->on('Finish', function ($serv, $task_id, $data) {
 	$pdo		=  $serv->PDO;
	if($data['success']<=0 && $data['result']<=0){
		$data['state'] = '2';
		$tmplmsg_s 	= "update ims_gicai_asgr_tmplmsg set click=click+{$data['result']},error=error+{$data['error']},page=page-1,now_worker=now_worker-1,state={$data['state']} where id=:id";
	}else{
		$tmplmsg_s 	= "update ims_gicai_asgr_tmplmsg set click=click+{$data['result']},error=error+{$data['error']},now_worker=now_worker-1 where id=:id";
	}
	$statement  = $pdo->prepare($tmplmsg_s);
	$statement->execute(array(':id'=>$data['tid'])); 
	
	if($statement){
		$affect_row = $statement->rowCount();
		echo "AsyncTask[{$task_id}] Finish: ok".PHP_EOL;
	}else{
		 
		echo "AsyncTask[{$task_id}] Finish: no".PHP_EOL;
	}
    $statement 	= null;
	$pdo		= null;
});

$serv->on('Close', function ($serv, $fd) {
    echo "Client: Close.\n";
});

$serv->start();
```

### 你期待的结果是什么？实际看到的错误信息又是什么？
我现在的业务是 通过PHP TCP 向这个服务器 发送数据 ，然后服务器 接收到参数后 开始查询数据库的粉丝 拿到 第一页 100条数据 开始 布置Task任务 ，然后 通过foreach 循环 创建GO协程 每个GO里面是 curl 向服务器提交模板消息，然后回调 统计出成功的 返回 Finish 里 然后修改下 SQL 指定的表 。这个过程 都跑通了，但是CUP 大概在 60-90%徘徊 内存一点没消耗，大概 1个小时能发 不到 6万条。 我想肯定有其他好办法，我请教各位大神 能帮我优化下吗 。

发布于2年前 · 19 次浏览 · 来自提问