发表新帖

一个困扰了很久的问题，到底swooleprocess+task模式能不能做大数据通讯？

### 问题描述
我用task的方式做了一个mysql连接池
4.4.18运行一会就报错，退出了，没看明白到底是什么原因
[2020-05-23 14:05:12 #2491.3]	NOTICE	swServer_master_send (ERRNO 1005): send event$[114] failed, session#1931631660 does not exist
[2020-05-23 14:05:14 ^2520.33]	WARNING	swWorker_reactor_is_empty (ERRNO 9012): worker exit timeout, forced to terminate

### Swoole版本，PHP版本，以及操作系统版本信息
swoole

Swoole => enabled
Author => Swoole Team <team@swoole.com>
Version => 4.4.18
Built => May 21 2020 10:03:00
coroutine => enabled
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
pcre => enabled
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => On => On
swoole.unixsock_buffer_size => 67108864 => 67108864

### 相关代码
```php
请将代码粘贴至此处（请勿用截图）
public static function getServerConfig(){
		global $SYS;
// 		$workerNum = intval($SYS["serverMemory"]/3/MYSQLPOOL_MEMORY_LIMIT);
// 		if($workerNum <1){$workerNum = 1;}
		$workerNum = 20;//先手动制定,主要还是要看数据库的压力测试结果
		return array(
			"dispatch_mode"=>3,
			"worker_num" => $workerNum,//连接池的管理进程数,一般设置为PHP的(最大)进程数相当
            "reactor_num"=>swoole_cpu_num(),
			"task_worker_num"=>$workerNum,//连接池执行SQL的进程数(并发数)
			"task_max_request"=>500,//设置task进程的最大任务数。一个task进程在处理完超过此数值的任务后将自动退出。这个参数是为了防止PHP进程内存
			"open_eof_check"=>true,
			"package_max_length"=>64*1024*1024,//接收数据包的最大尺寸
			"buffer_output_size"=>64*1024*1024,//发送数据包的最大尺寸
//            "socket_buffer_size"=>64*1024*1024,//发送数据包的最大尺寸，这个参数打开好像有问题
			"package_eof"=>MysqlPoolConfig::PACK_EOF,
			"enable_reuse_port"=>true,
			"task_enable_coroutine"=>true,
			"daemonize" => true,
			"backlog" => 2048,
// 			"user"=>"www",
// 			"group"=>"www",
			"_mysql_wait_timeout"=>28000,//mysql的wait time out配置，默认是28800，通常设置比mysql的要少一点，在被动断开之前就要重连
			"_normalLogDir" => WEBROOTPATH."/logdata/mysqlPoolServer/normal/",//普通日志存放目录
			"_warningLogDir" => WEBROOTPATH."/logdata/mysqlPoolServer/warning/",//警告日志存放目录
			"_obLogDir" => WEBROOTPATH."/logdata/mysqlPoolServer/oblog/",//性能监控日志存放目录
			"_sqlErrorLogDir" => WEBROOTPATH."/logdata/mysqlPoolServer/sqlError/",//sqlError监控日志存放目录
			"log_file" => WEBROOTPATH."/logdata/mysqlPoolServer/Swoole.log",//swoole日志文件
		);
	}
```

mysqlPoolTask.php
```
<?php

class MysqlQueryTask extends WsServerTask {

const TaskName = "MysqlQueryTask";
	const TaskParam = array(
			"taskName"=>"MysqlQueryTask",
			"param"=>array(
					"subTaskId"=>0,
			),
			"queryData"=>array(),//客户端发送上来的数据
			"clientFd"=>0,//客户端Fd
			"createMicTimes"=>0,//创建时间
	);

/**
     * 数据发送到客户端最大尝试次数
     */
	private $pm_sendToClientMaxRetryTimes = 10;
	
	
	/**
	 * 执行一次数据库操作请求
	 * @param array $queryData
	 * @return array
	 */
	private function doQuery($queryData,$isRetry = false)
	{
		$queryId = $queryData["queryId"];
		$queryFw = $queryData["queryFw"];
		if($isRetry == false)
		{
		    MysqlPoolServer::logNormal("[pid:".getmypid()."][qid:{$queryId}][qf:{$queryFw}][查询]",true);
			MysqlPoolServer::logWarning("[pid:".getmypid()."][qid:{$queryId}][qf:{$queryFw}][查询],sql:".$queryData["sql"].",queryData:".var_export($queryData,true));
		}
		
		$st = microtime(true);
		$limitRecordNum = intval($queryData["limitRecordNum"]);
		try
		{
			MysqlPoolServer::$s_mgdb->resetError();
			MysqlPoolServer::$s_mgdb->resetSqlError();
			$rsarray = MysqlPoolServer::$s_mgdb->tqueryReal(
					$queryData["sql"],
					$queryData["dbNamePostFix"],
					intval($queryData["skipRecordNum"]),
					$limitRecordNum,
					intval($queryData["dbQuerySplitMode"]),
					boolval($queryData["checkoptimize"]),
					$queryData["targetDbConfigName"],
					intval($queryData["targetSubsiteid"])
			);
		}catch (Exception $e)
		{
			$toClientData = $this->toClientData(V_ERRORCODE_NORMAL_ERROR, $e->getMessage(), array(),0);
			MysqlPoolServer::logSqlError("[pid:".getmypid()."][sqlException] queryData=>".var_export($queryData,true).",error:".$e->getMessage().",realSql:".var_export(MysqlPoolServer::$s_mgdb->getLastRealQuerySql(),true));
			return $toClientData;
		}
		
// 		echo "iserror:".$queryData["sql"]."=>".MysqlPoolServer::$s_mgdb->error()."\n";
		$logContent = "[qid:{$queryId}][查询耗时:".Math::sprintf(microtime(true)-$st)."],isRetry:{$isRetry}";
		MysqlPoolServer::logWarning("[pid:".getmypid()."]".$logContent);
		
		$toClientData = null;
		if(MysqlPoolServer::$s_mgdb->errno()!=0)
		{
			if(in_array(MysqlPoolServer::$s_mgdb->errno(1),array(2006,2013)))
			{
				if($isRetry == false)
				{
					//2006 - MySQL server has gone away,自动重连
					MysqlPoolServer::reConnectMysql(true);
					MysqlPoolServer::logSqlError("[pid:".getmypid()."][mysqlError][pid:".getmypid()."] MySQL server has gone away,重新连接数据库");
					return $this->doQuery($queryData,true);
				}else
				{
					MysqlPoolServer::$s_mgdb->m_ERRORMSG = "Db server has gone away";
					$toClientData = $this->toClientData(V_ERRORCODE_NORMAL_ERROR,  MysqlPoolServer::$s_mgdb->m_ERRORMSG, $queryData,0);
				}
			}else
			{
				$toClientData = $this->toClientData(V_ERRORCODE_NORMAL_ERROR, MysqlPoolServer::$s_mgdb->getSqlError(), $queryData,0);
			}
			
			MysqlPoolServer::logSqlError("[pid:".getmypid()."][sqlQueryError] queryData=>".var_export($queryData,true).",error[".MysqlPoolServer::$s_mgdb->errno().",".MysqlPoolServer::$s_mgdb->errno(1)."]:".MysqlPoolServer::$s_mgdb->getSqlError().",realSql:".var_export(MysqlPoolServer::$s_mgdb->getLastRealQuerySql(),true));
		}elseif(MysqlPoolServer::$s_mgdb->m_ERRORCODE != V_ERRORCODE_NONE)
		{
			$toClientData = $this->toClientData(V_ERRORCODE_NORMAL_ERROR, MysqlPoolServer::$s_mgdb->m_ERRORMSG, $queryData,0);
			MysqlPoolServer::logSqlError("[pid:".getmypid()."][sqlExecError] queryData=>".var_export($queryData,true).",error:".MysqlPoolServer::$s_mgdb->m_ERRORMSG.",realSql:".var_export(MysqlPoolServer::$s_mgdb->getLastRealQuerySql(),true));
		}else
		{
			$toClientData = $this->toClientData(V_ERRORCODE_NONE, "", $rsarray["data"],MysqlPoolServer::$s_mgdb->insert_id(true));
			if($limitRecordNum >1 && strpos(strtolower($queryData["sql"]), "select") !== false && count($rsarray["data"])<$limitRecordNum)
			{
				//查询到的数据<需要查询的记录数,数据已经是最后一批了
				$toClientData["_isEnd"] = 1;
			}
			if(!empty($rsarray["sqlIndexType"]))
			{
				$toClientData["_sqlIndexTypes"] = $rsarray["sqlIndexType"];
			}
		}
		
		$useMemory = memory_get_usage();
		
		
		$logContent = "[qid:{$queryId}][查询结果]";
		$logErrorContent = "";
		if(MysqlPoolServer::$s_mgdb->errno() !=0)
		{
			$logErrorContent .= "sqlError[".MysqlPoolServer::$s_mgdb->errno().",".MysqlPoolServer::$s_mgdb->errno(1).",".MysqlPoolServer::$s_mgdb->error()."]=>".MysqlPoolServer::$s_mgdb->getSqlError().",";
		}else
		{
			$logErrorContent .= "sqlError[No],";
		}
		if(MysqlPoolServer::$s_mgdb->m_ERRORCODE != V_ERRORCODE_NONE)
		{
			$logErrorContent .= "error[".MysqlPoolServer::$s_mgdb->m_ERRORCODE."]=>".MysqlPoolServer::$s_mgdb->m_ERRORMSG.",";
		}else
		{
			$logErrorContent .= "error[No],";
		}
		MysqlPoolServer::logWarning("[pid:".getmypid()."][qid:{$queryId}]realQuerySql=>".var_export(MysqlPoolServer::$s_mgdb->getLastRealQuerySql(),true));
		
		$useMemory = intval($useMemory/1024/1024);
		$logContent .= $logErrorContent.",resultnum:".(is_array($rsarray["data"])?count($rsarray["data"]):var_export($rsarray["data"],true)).",mem:{$useMemory}Mb";
		if($useMemory > MYSQLPOOL_MEMORY_LIMIT/3)
		{
			$logContent .= ",占用内存过高=>".Math::sprintf($useMemory/MYSQLPOOL_MEMORY_LIMIT*100)."%";
		}
		MysqlPoolServer::logWarning("[pid:".getmypid()."]".$logContent);
	
		return $toClientData;
	}
	
	
	/**
	 * 返回发送给客户端的数据格式
	 * @param int $code		出错编码
	 * @param string $message	出错信息
	 * @param any $data		附加的数据
	 * @param int $sqlInsertId	如果是插入语句，这里可以附带返回插入的自增ID
	 * @return array
	 */
	public function toClientData($code,$message,$data,$sqlInsertId=0)
	{
		return array(
				"code"=>$code,
				"message"=>$message,
				"data"=>$data,
				"_maxLimitNum"=>0,//最大可以查询的返回数,根据占用内存动态计算
				"_sqlInsertId"=>$sqlInsertId,
				"_isEnd"=>0,//是否最后一批查询了
				"_sqlIndexTypes"=>array(),//查询索引类型
		);
	}
	
	
	public function runTask(Swoole\Server $server, Swoole\Server\Task $task)
	{
		$taskInfo = $task->data;
		$queryData = $taskInfo["queryData"];
		$clientFd = $taskInfo["clientFd"];
		if(empty($queryData))
		{
			MysqlPoolServer::logNormal("[pid:".getmypid()."]收到一个空请求包,from=>".var_export($task,true));
		}
		
// 		echo "处理clientFd:{$clientFd}的请求,等待时间:".(microtime(true)-$taskInfo["createMicTimes"])."\n";
		$startMemory = memory_get_usage();
		$toClientData = $this->doQuery($queryData);
		$memoryAfterQuery = memory_get_usage();
// 		echo "处理完clientFd:{$clientFd}的请求\n";
// 		echo "toClientData({$clientFd})=>".var_export($toClientData,true)."\r\n";
// 		$sendData = json_encode($toClientData,JSON_UNESCAPED_UNICODE).MysqlPoolConfig::PACK_EOF;

//保守一点,只使用指定百分比的内存
		$sendRecordNum = 0;
		$maxStrlenPerRecord = 0;
		
		/***
		 * 要获取连接池端最大返回的记录数量
		* 这里要注意,由于要把对象生成符合协议需要的字符串,所以最多只能用进程规定的50%内存,否则在字符串拼接的时候,会出现内存溢出,
		* 这受限于PHP的内存拼接要先赋值出一个新对象再销毁原来对象的操作模式
		*/
		$sendData = MysqlPoolProtocol::packDataToSendData($toClientData, MYSQLPOOL_MEMORY_LIMIT/2*0.9,$sendRecordNum,$maxStrlenPerRecord);
		
		$queryId = $queryData["queryId"];
		
		if($queryData["_getMaxLimitNum"] == "1")
		{
			$maxLimitNum = -1;//默认让客户端自己定吧
			if($maxStrlenPerRecord >0)
			{
				$memoryNow = memory_get_usage();
// 				$maxStrlenPerRecord *= 8;//8bytes
				$maxLimitNum = intval((MYSQLPOOL_MEMORY_LIMIT/2*1024*1024 - $memoryAfterQuery)*0.9/$maxStrlenPerRecord);
				if($maxLimitNum <=0){ $maxLimitNum = 1;}
			}
			$toClientData["_maxLimitNum"] = $maxLimitNum;
			unset($sendData);
			$tmp = 0;
			$sendData = MysqlPoolProtocol::packDataToSendData($toClientData, MYSQLPOOL_MEMORY_LIMIT/2*0.9,$sendRecordNum,$tmp);
		}
		$useMemory = intval(memory_get_usage()/1024/1024);
		$logContent = "[pid:".getmypid()."][qid:{$queryId}][查询],实际输出:{$sendRecordNum}条记录,mem:{$useMemory}Mb,strlen:".strlen($sendData)."|".(strlen($sendData)/1024/1024);
        $logContent .= "\r\nqueryData=>".var_export($queryData,true)."\r\n";
		if(intval($toClientData["_maxLimitNum"]) !=0)
		{
			$logContent .= ",_maxLimitNum:".$toClientData["_maxLimitNum"];
			$logContent .= ",每条记录mem:".$maxStrlenPerRecord;
		}
		if(strlen($sendData)>1024*1024*10)
        {
            MysqlPoolServer::logWarning($logContent,true);
        }

unset($toClientData);

//        $sendTimes = 0;
//        $sendResult = false;
//        while($sendTimes < $this->pm_sendToClientMaxRetryTimes)
//        {
//            $sendResult = $server->send($clientFd, $sendData);
//            if($sendResult == true)
//            {
//                break;
//            }
//            $sendTimes ++;
//            usleep(500000);//休眠500毫秒
//        }

//        if($sendResult == false)

$sendResult = MysqlPoolServer::sendDataToClient($server,$clientFd,$sendData,"cls:".__CLASS__.",line:".__LINE__,$this->pm_sendToClientMaxRetryTimes);
        if($sendResult != "")
        {
            MysqlPoolServer::logWarning($sendResult,true);
        }

// 		MysqlPoolServer::logNormal("sql_send(".$sendResult.")=>".$queryData["sql"]."=>".strlen($sendData));
	}
	
	public function registerTask(Swoole\Server $server)
	{
		return;
	}
	
}

class MysqlPoolStatusTask extends WsServerTask {

const TaskName = "MysqlPoolStatusTask";
	const TaskParam = array(
			"taskName"=>"MysqlPoolStatusTask",
			"param"=>array(
					"subTaskId"=>0,
			),
			"queryData"=>array(),//客户端发送上来的数据
			"clientFd"=>0,//客户端Fd
			"createMicTimes"=>0,//创建时间
	);

/**
     * 数据发送到客户端最大尝试次数
     */
    private $pm_sendToClientMaxRetryTimes = 10;

/**
	 * 返回发送给客户端的数据格式
	 * @param int $code		出错编码
	 * @param string $message	出错信息
	 * @param any $data		附加的数据
	 * @return array
	 */
	public function toClientData($code,$message,$data)
	{
		return array(
				"code"=>$code,
				"message"=>$message,
				"data"=>$data
		);
	}
	
	
	/**
	 * 发送一个信息给客户端
	 * @param int $errorCode
	 * @param string $message
	 * @param array $data
	 * @param Swoole\Server $server
	 * @param int $clientFd
	 */
	public function sendMessage($errorCode,$message,$data,Swoole\Server $server,$clientFd)
	{
		$toClientData = $this->toClientData($errorCode, $message, $data);
		$sendRecordNum = 0;
		$tmpNum = 0;
		$sendData = MysqlPoolProtocol::packDataToSendData($toClientData, MYSQLPOOL_MEMORY_LIMIT/2*0.9,$sendRecordNum,$tmpNum);

$sendResult = MysqlPoolServer::sendDataToClient($server,$clientFd,$sendData,"cls:".__CLASS__.",line:".__LINE__,$this->pm_sendToClientMaxRetryTimes);

//        if($sendResult == false)
        if($sendResult != "")
        {
            MysqlPoolServer::logWarning($sendResult,true);
        }
	}

public function runTask(Swoole\Server $server, Swoole\Server\Task $task)
	{
		$taskInfo = $task->data;
		$queryData = $taskInfo["queryData"];
		$clientFd = $taskInfo["clientFd"];
		if(empty($queryData))
		{
			MysqlPoolServer::logNormal("[pid:".getmypid()."]收到一个空请求包,from=>".var_export($task,true));
		}

$toClientData = $this->toClientData(V_ERRORCODE_NONE, "", array($server->stats()));
// 				echo "toClientData({$clientFd})=>".var_export($toClientData,true)."\r\n";
// 		$server->send($clientFd, json_encode($toClientData,JSON_UNESCAPED_UNICODE).MysqlPoolConfig::PACK_EOF);
		$sendRecordNum = 0;
		$tmpNum = 0;
		$sendData = MysqlPoolProtocol::packDataToSendData($toClientData, MYSQLPOOL_MEMORY_LIMIT/2*0.9,$sendRecordNum,$tmpNum);

//        $sendTimes = 0;
//        $sendResult = false;
//        while($sendTimes < $this->pm_sendToClientMaxRetryTimes) {
//            $sendResult = $server->send($clientFd, $sendData);
//            if($sendResult == true)
//            {
//                break;
//            }
//            $sendTimes ++;
//            usleep(500000);//休眠500毫秒
//        }
        $sendResult = MysqlPoolServer::sendDataToClient($server,$clientFd,$sendData,"cls:".__CLASS__.",line:".__LINE__,$this->pm_sendToClientMaxRetryTimes);

//        if($sendResult == false)
        if($sendResult != "")
        {
            MysqlPoolServer::logWarning($sendResult,true);
        }
	}

public function registerTask(Swoole\Server $server)
	{
		return;
	}

```

mysqlPoolServer.php
```
<?php
include_once("../../init.base.cli.php");
define("MYSQLPOOL_MEMORY_LIMIT",128);//mb
ini_set("memory_limit",-1);//这里必须设置为不限制,不然还是容易出现超内存的情况
// ini_set("memory_limit",MYSQLPOOL_MEMORY_LIMIT."M");
include_once("wsServerLib/wsServerTask.php");
include_once("tasks/mysqlPoolTask.php");

/***
 * 需要调整的linux系统参数
 * 1./etc/security/limits.conf,调整最大打开文件数
 * 2./etc/sysctl.conf如下参数
 * 	a.net.ipv4.tcp_tw_reuse=1
 * 3.php.ini
 * 	a.memory_limit设置为合理值
 *  b.display_errors = On
 *  c.extension[swoole.so]
 * 
 */
class MysqlPoolServer
{
	/**
	 * 
	 * @var Swoole\Server
	 */
	public $serv = null;
	
	/**
	 * 当前根目录
	 * @var string
	 */
	private $pm_currentDir = "";
	
	/**
	 * 普通的日志
	 * @var Log4PHP
	 */
	public static $s_logCls = null;
	
	/**
	 * 警告日志的日志
	 * @var Log4PHP
	 */
	public static $s_warninglogCls = null;
	
	/**
	 * 性能监控日志
	 * @var Log4PHP
	 */
	public static $s_oblogCls = null;
	
	/**
	 * SQL出错监控日志
	 * @var Log4PHP
	 */
	public static $s_sqlErrorlogCls = null;
	
	/**
	 * 数据库查询主要对象
	 * @var MGDB
	 */
	public static $s_mgdb = null;
	/**
	 * 最后一次连接数据库的时间戳
	 * @var int
	 */
	public static $s_lastConnectMysqlTimes = 0;
	
	/**
	 * 数据库自动重连时间(秒)
	 * @var int
	 */
	public static $s_mysqlReconnectTimes = 300;
	
	/**
	 * 最后一次输出性能监控的时间戳,默认每分钟输出一次内存占用等信息
	 * @var int
	 */
	private $pm_lastObInfoTimes = 0;
	
	/**
	 * 连接池的连接状态(0:已断开,-1:连接中,1:已连接)
	 * @var int
	 */
	private static $pm_dbConnectStatus = 0;
	
	
	public function __construct()
	{
		global $SYS;
		$serverConfig = MysqlPoolConfig::getServerConfig();
		
		$normalLogDir = ($serverConfig["_normalLogDir"] == "")?"./logdata/mysqlPoolServer/normal/":$serverConfig["_normalLogDir"];
		MysqlPoolServer::$s_logCls = Log4PHP::getInstance($normalLogDir);
		
		$warningLogDir = ($serverConfig["_warningLogDir"] == "")?"./logdata/mysqlPoolServer/warning/":$serverConfig["_warningLogDir"];
		MysqlPoolServer::$s_warninglogCls = Log4PHP::getInstance($warningLogDir);
		
		$obLogDir = ($serverConfig["_obLogDir"] == "")?"./logdata/mysqlPoolServer/oblog/":$serverConfig["_obLogDir"];
		MysqlPoolServer::$s_oblogCls = Log4PHP::getInstance($obLogDir);
		
		$sqlErrorLogDir = ($serverConfig["_sqlErrorLogDir"] == "")?"./logdata/mysqlPoolServer/sqlError/":$serverConfig["_sqlErrorLogDir"];
		MysqlPoolServer::$s_sqlErrorlogCls = Log4PHP::getInstance($sqlErrorLogDir);
		
		$this->pm_currentDir = pathinfo(__FILE__,PATHINFO_DIRNAME);
		$this->serv = new Swoole\Server(MysqlPoolConfig::UnixSocketFile, 0, SWOOLE_PROCESS, SWOOLE_UNIX_STREAM);
		
		#自动创建日志目录
		if(isset($serverConfig["log_file"]))
		{
			$swooleLogDir = pathinfo($serverConfig["log_file"],PATHINFO_DIRNAME);
			if(!file_exists($swooleLogDir) || !is_dir($swooleLogDir))
			{
				@mkdir($swooleLogDir,0777,true);
			}
		}

$this->serv->set($serverConfig);
		if(intval($serverConfig["_mysql_wait_timeout"]) >60)
		{
			MysqlPoolServer::$s_mysqlReconnectTimes = intval($serverConfig["_mysql_wait_timeout"]);
		}
		
		
		echo "[".date("Ymd H:i:s")."]服务器内存配置为:".$SYS["serverMemory"]."Mb,连接池启动进程数:".$serverConfig["task_worker_num"]."\n";
		
// 		$this->serv->on("connect", function ($serv, $fd){
// 		    echo "Client:Connect. {$fd}\n";
// 		});
		$this->serv->on("receive", function ($serv, $fd, $reactor_id, $data) {
// 			echo "getData taskQueueItemNum:".var_export($serv->stats(),true)."\n";
			
			if(strlen($data) < strlen(MysqlPoolConfig::PACK_EOF)){
				return;
			}
			$realData = substr($data, 0,strlen($data)-strlen(MysqlPoolConfig::PACK_EOF));
			unset($data);
			
			$queryData = MysqlPoolProtocol::unPackQueryData($realData);
			unset($realData);
			
			switch (MysqlPoolServer::$pm_dbConnectStatus)
			{
				case -1://连接中
					$poolStatusTask = new MysqlPoolStatusTask();
					$poolStatusTask->sendMessage(V_ERRORCODE_NORMAL_ERROR, "db连接池繁忙", array(), $serv, $fd);
					return;
				break;
				
				case 0://连接失败
					$poolStatusTask = new MysqlPoolStatusTask();
					$poolStatusTask->sendMessage(V_ERRORCODE_NORMAL_ERROR, "db连接失败", array(), $serv, $fd);
					return;
				break;
			} 
			
			$taskInfo = null;
			switch ($queryData["action"])
			{
				case MysqlPoolProtocol::ACTION_SQLQUERY://执行SQL查询
					MysqlPoolServer::logWarning("收到请求[qid:".$queryData["queryId"]."]");
					$taskInfo = MysqlQueryTask::TaskParam;
				break;
				
				case MysqlPoolProtocol::ACTION_GETPOOLSTATUS://获取连接池状态
					$taskInfo = MysqlPoolStatusTask::TaskParam;
				break;
			}
			
			if(time()-$this->pm_lastObInfoTimes > 60)
			{
				//输出性能监控日志,每分钟输出一次
				$this->logObInfo($serv);
				$this->pm_lastObInfoTimes = time();
			}
			
			if($taskInfo == null)
			{
				$logContent = "收到无效的action类型=>".$queryData["action"];
				echo "[".date("Ymd H:i:s")."]".$logContent."\n";
				MysqlPoolServer::logNormal($logContent);
				return ;
			}
			
			$taskInfo["queryData"] = $queryData;
			$taskInfo["clientFd"] = $fd;
			$taskInfo["createMicTimes"] = microtime(true);
		    $this->serv->task($taskInfo);
		});
		
		$this->serv->on("task", function(Swoole\Server $server, Swoole\Server\Task $task)
		{
// 			echo "on task:".var_export($task,true)."=>".getmypid()."\n";
			if(!is_array($task->data)){ return;}
            /***
             * 必须关掉RifBackTrace,否则长时间运行会导致写入大量backTrace到Redis,后面组件调用全部读入大量backTrace数据导致内存溢出
             */
			SystemEnvFW2018::setIsDebugReturnRifBackTrace(false);

$taskInfo = $task->data;
			
			switch ($taskInfo["taskName"])
			{
				case MysqlQueryTask::TaskName://数据库请求
					MysqlPoolServer::reConnectMysql();
					$sqlQueryTask = new MysqlQueryTask();
					$sqlQueryTask->runTask($server, $task);
				break;
				
				case MysqlPoolStatusTask::TaskName://数据库连接池状态
					$poolStatusTask = new MysqlPoolStatusTask();
					$poolStatusTask->runTask($server, $task);
				break;
			}
		});
		
// 		$this->serv->on("close", function ($serv, $fd) {
// 		    echo "Client: Close {$fd}.\n";
// 		});
		
			$this->serv->on("ManagerStart", function ($serv) {
				echo "[".date("Ymd H:i:s")."]MysqlPoolServer ManagerStart ,pid:".getmypid().".\n";
				//改变unixsocket的权限
				chmod(MysqlPoolConfig::UnixSocketFile, 0777);
			});
			
            $this->serv->on("WorkerStart", function ($serv) {
                echo "[".date("Ymd H:i:s")."]WorkerStart,连接池进程启动,pid:".getmypid().".\n";
                $this->reConnectMysql();
            });

$this->serv->on("WorkerError", function (Swoole\Server $server, $worker_id, $worker_pid, $exit_code, $signal) {
                echo "[".date("Ymd H:i:s")."]WorkerError,pid:{$worker_pid},wid:{$worker_id},exit_code:{$exit_code},signal:{$signal}\n";
            });

$this->serv->on("ManagerStop", function ($serv) {
				echo "[".date("Ymd H:i:s")."]MysqlPoolServer ManagerStop ,pid:".getmypid().".\n";
			});
		
		
		$this->serv->start();
	}

/**
     * 写入一个警告日志
     * @param $content
     */
	public static function logWarning($content,$focus = false)
    {
        if($focus == false)
        {
            return;
        }
        if(MysqlPoolServer::$s_warninglogCls == null){ return;}
        MysqlPoolServer::$s_warninglogCls->put($content);
    }

/**
     * 记录一个普通日志
     * @param $content
     */
    public static function logNormal($content,$focus = false)
    {
        if($focus == false) {
            return;
        }
        if(MysqlPoolServer::$s_logCls == null){ return;}
        MysqlPoolServer::$s_logCls->put($content);
    }

/**
     * 记录一个SQL出错日志
     * @param $content
     */
    public static function logSqlError($content)
    {
        if(MysqlPoolServer::$s_sqlErrorlogCls == null){ return;}
        MysqlPoolServer::$s_sqlErrorlogCls->put($content);
    }

/**
     * 写入一个性能监控日志
     * @param $content
     */
    public static function logOb($content)
    {
        if(MysqlPoolServer::$s_oblogCls == null){ return;}
        MysqlPoolServer::$s_oblogCls->put($content);
    }
	
	/**
	 * 重新连接数据库
	 * @param	$focus	是否强制重连
	 */
	public static function reConnectMysql($focus = false)
	{
		$logContent = "";
		if(MysqlPoolServer::$s_mgdb == null){
			MysqlPoolServer::$pm_dbConnectStatus = -1;
			MysqlPoolServer::$s_lastConnectMysqlTimes = time();
			MysqlPoolServer::$s_mgdb = new MGDB();
			$errMsg = MysqlPoolServer::connectAllDbHost();
			if($errMsg == "")
			{
				$logContent = "连接池进程[".getmypid()."]首次连接数据库成功";
				MysqlPoolServer::$pm_dbConnectStatus = 1;
			}else
			{
				$logContent = "连接池进程[".getmypid()."]首次连接数据库失败";
				MysqlPoolServer::$pm_dbConnectStatus = 0;
				echo "[".date("Ymd H:i:s")."]".$logContent;
			}
		}else
		{
			if($focus == true || time()-MysqlPoolServer::$s_lastConnectMysqlTimes > MysqlPoolServer::$s_mysqlReconnectTimes)
			{
				MysqlPoolServer::$pm_dbConnectStatus = -1;
				MysqlPoolServer::$s_mgdb->releaseDbConnection();
				MysqlPoolServer::$s_mgdb = null;
				MysqlPoolServer::$s_lastConnectMysqlTimes = time();
				MysqlPoolServer::$s_mgdb = new MGDB();
				$errMsg = MysqlPoolServer::connectAllDbHost();
				if($errMsg == "")
				{
					$logContent = "连接池进程[".getmypid()."]重新连接数据库成功,focus:{$focus}";
					MysqlPoolServer::$pm_dbConnectStatus = 1;
				}else
				{
					$logContent = "连接池进程[".getmypid()."]重新连接数据库失败,focus:{$focus}";
					MysqlPoolServer::$pm_dbConnectStatus = 0;
					echo "[".date("Ymd H:i:s")."]".$logContent;
				}
			}
		}
		if($logContent != "")
		{
			MysqlPoolServer::logNormal($logContent);
		}
	}

/**
     * 发送数据给客户端
     * @param \Swoole\Server $server
     * @param $clientFd
     * @param $sendData
     * @param string $traceMsg
     * @param int $maxRetryTimes
     * @return 成功返回空字符串，否则返回出错信息
     */
	public static function sendDataToClient(Swoole\Server $server,$clientFd,$sendData,$traceMsg = "",$maxRetryTimes = 3)
    {
        $sendTimes = 0;
        $sendResult = false;
        while($sendTimes < $maxRetryTimes)
        {
            if($server->exist($clientFd) == false)
            {
                return "发送数据到连接池客户端[{$clientFd}]时发现客户端已断开,次数:{$sendTimes}";
            }
            $sendResult = $server->send($clientFd, $sendData);
            if($sendResult == true)
            {
                break;
            }
            $sendTimes ++;
            usleep(500000);//休眠500毫秒
        }

if($sendResult == false)
        {
            return "发送数据到连接池客户端达到{$maxRetryTimes}次失败,traceMsg:{$traceMsg}";
        }

return "";
    }
	
	/**
	 * 重新连接数据库
	 * @return 连接成功返回空字符串,否则返回出错信息
	 */
	private static function connectAllDbHost()
	{
		if(MysqlPoolServer::$s_mgdb == null){ return;}
		
		$dbType = array(
				V_DBCONFIG_SITE,
				V_DBCONFIG_SITEEXT,
				V_DBCONFIG_STAT,
				V_DBCONFIG_LOGS,
				V_DBCONFIG_ENTERPRISE,
				V_DBCONFIG_GLOBAL,
		);
		
		foreach ($dbType as $dbConfigName)
		{
			$conn = MysqlPoolServer::$s_mgdb->getDbConnection(0, V_DB_QUERY_SPLITMODE_CAPACITY, $dbConfigName,V_SUBSITEID_SP_KA);
			if(MysqlPoolServer::$s_mgdb->m_ERRORCODE != V_ERRORCODE_NONE)
			{
				$logContent = "进程[".getmypid()."]连接数据库失败=>".$dbConfigName;
				return $logContent;
			}
		}
		return "";
	}
	
	
	public function logObInfo(Swoole\Server $serv){
		$useMemory = intval(memory_get_usage()/1024/1024);
		$peakMemory = intval(memory_get_peak_usage()/1024/1024);
		MysqlPoolServer::logOb("连接池进程[".getmypid()."]使用内存:{$useMemory}Mb,系统分配内存:{$peakMemory}Mb,当前连接数:".count($serv->connections));
	}
}
new MysqlPoolServer();
```

### 你期待的结果是什么？实际看到的错误信息又是什么？

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

minguo119

### Swoole版本，PHP版本，以及操作系统版本信息
swoole

mysqlPoolTask.php
```
<?php

class MysqlQueryTask extends WsServerTask {

unset($toClientData);

//        if($sendResult == false)

// 		MysqlPoolServer::logNormal("sql_send(".$sendResult.")=>".$queryData["sql"]."=>".strlen($sendData));
	}
	
	public function registerTask(Swoole\Server $server)
	{
		return;
	}
	
}

class MysqlPoolStatusTask extends WsServerTask {

/**
     * 数据发送到客户端最大尝试次数
     */
    private $pm_sendToClientMaxRetryTimes = 10;

$sendResult = MysqlPoolServer::sendDataToClient($server,$clientFd,$sendData,"cls:".__CLASS__.",line:".__LINE__,$this->pm_sendToClientMaxRetryTimes);

//        if($sendResult == false)
        if($sendResult != "")
        {
            MysqlPoolServer::logWarning($sendResult,true);
        }
	}

public function registerTask(Swoole\Server $server)
	{
		return;
	}

```

$this->serv->on("ManagerStop", function ($serv) {
				echo "[".date("Ymd H:i:s")."]MysqlPoolServer ManagerStop ,pid:".getmypid().".\n";
			});
		
		
		$this->serv->start();
	}

if($sendResult == false)
        {
            return "发送数据到连接池客户端达到{$maxRetryTimes}次失败,traceMsg:{$traceMsg}";
        }

### 你期待的结果是什么？实际看到的错误信息又是什么？

登录后参与评论

2020-05-26

郭新华

2020-05-28

minguo119 回复郭新华