php 使用swoole多进程执行任务，避免mysql has gone away

曾经终败给现在 2023-06-10 05:27 2阅读 0赞

swoole多进程处理数据速度很快，但是会遇到mysql has gone away的情况需要小心：

Mysql.php:

<?php
 
 class Mysql
 {
 
 private $host;
 
 private $user;
 
 private $password;
 
 private $charset;
 
 private $database;
 
 /**
 * 新建数据库连接对象，测试数据库连接
 *
 * @param string $host
 * @param string $user
 * @param string $password
 * @param string $charset
 * @param string $database
 */
 function __construct($host, $user, $password, $charset, $database)
 {
 $link = mysqli_connect($host, $user, $password) or die('数据库连接失败 ERROR ' . mysqli_connect_errno() . ':' . mysqli_connect_error());
 $char = mysqli_set_charset($link, $charset) or die('charset设置错误，请输入正确的字符集名称 ERROR ' . mysqli_errno($link) . ':' . mysqli_error($link));
 $db = mysqli_select_db($link, $database) or die('未找到数据库，请输入正确的数据库名称 ERROR ' . mysqli_errno($link) . ':' . mysqli_error($link));
 $this->host = $host;
 $this->user = $user;
 $this->password = $password;
 $this->charset = $charset;
 $this->database = $database;
 mysqli_close($link);
 }
 
 /**
 * 连接数据库
 *
 * @param string $host
 * @param string $user
 * @param string $password
 * @param string $charset
 * @param string $database
 * @return object 连接标识符
 */
 public function connect($host = '', $user = '', $password = '', $charset = '', $database = '')
 {
 $link = mysqli_connect($this->host, $this->user, $this->password);
 mysqli_set_charset($link, $this->charset);
 mysqli_select_db($link, $this->database);
 return $link;
 }
 
 /**
 * 增加数据
 *
 * @param array $data
 * @param string $table
 * @return boolean
 */
 public function insert($data, $table,$link = '')
 {
 if(empty($link)){
 
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 }
 $keys = join(',', array_keys($data));
 $vals = "'" . join("','", array_values($data)) . "'";
 $query = "INSERT INTO {$table}({$keys}) VALUES({$vals})";
 $result = mysqli_query($link, $query) or die('插入数据出错，请检查！ ERROR ' . mysqli_errno($link) . ':' . mysqli_error($link));
 if ($result) {
 $id = mysqli_insert_id($link);
 } else {
 $id = false;
 }
 mysqli_close($link);
 return $id;
 }
 
 /**
 * 删除数据
 *
 * @param string $table
 * @param string $where
 * @return boolean
 */
 public function delete($table, $where = null,$link = '')
 {
 if(empty($link)){
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 }
 $where = $where ? ' WHERE ' . $where : '';
 $query = "DELETE FROM {$table}{$where}";
 $result = mysqli_query($link, $query) or die('删除数据出错，请检查！ ERROR ' . mysqli_errno($link) . ':' . mysqli_error($link));
 if ($result) {
 $row = mysqli_affected_rows($link);
 } else {
 $row = false;
 }
 mysqli_close($link);
 return $row;
 }
 
 /**
 * 修改数据
 *
 * @param array $data
 * @param string $table
 * @param string $where
 * @return boolean
 */
 public function update($data, $table, $where = null,$link = '')
 {
 if(empty($link)) {
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 }
 $set = '';
 foreach ($data as $key => $val) {
 $set .= "{$key}='{$val}',";
 }
 $set = trim($set, ',');
 $where = $where ? ' WHERE ' . $where : '';
 $query = "UPDATE {$table} SET {$set}{$where}";
 $result = mysqli_query($link, $query);
 if ($result) {
 $row = mysqli_affected_rows($link);
 } else {
 $row = false;
 }
 mysqli_close($link);
 return $row;
 }
 
 /**
 * 查询指定记录
 *
 * @param string $query
 * @param string $result_type
 * @return array|boolean
 */
 public function select_one($query, $result_type = MYSQLI_ASSOC,$link = '')
 {
 if(empty($link)) {
 
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 }
 $result = mysqli_query($link, $query);
 if ($result && mysqli_num_rows($result) > 0) {
 $row = mysqli_fetch_array($result, $result_type);
 } else {
 $row = false;
 }
 mysqli_free_result($result);
 mysqli_close($link);
 return $row;
 }
 
 /**
 * 查询所有记录
 *
 * @param string $query
 * @param string $result_type
 * @return array|boolean
 */
 public function select_all($query, $result_type = MYSQLI_ASSOC,$link = '')
 {
 if(empty($link)){
 
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 }
 // $query = "SELECT * FROM {$table}";
 $result = mysqli_query($link, $query);
 if ($result && mysqli_num_rows($result) > 0) {
 while ($row = mysqli_fetch_array($result, $result_type)) {
 $rows[] = $row;
 }
 } else {
 $rows = false;
 }
 mysqli_free_result($result);
 mysqli_close($link);
 return $rows;
 }
 
 /**
 * 得到表中记录数
 *
 * @param string $table
 * @return number|boolean
 */
 public function get_total_rows($table)
 {
 $link = $this->connect($this->host, $this->user, $this->password, $this->charset, $this->database);
 $query = "SELECT COUNT(*) AS totalRows FROM {$table}";
 $result = mysqli_query($link, $query);
 if ($result && mysqli_num_rows($result) == 1) {
 $row = mysqli_fetch_assoc($result);
 } else {
 $row['totalRows'] = false;
 }
 mysqli_close($link);
 return $row['totalRows'];
 }
 }

PageUtils.php

class PageUtils {
    
        public static $countpage;
        /**
         * 数组分页函数  核心函数  array_slice
         * 用此函数之前要先将数据库里面的所有数据按一定的顺序查询出来存入数组中
         * $count   每页多少条数据
         * $page   当前第几页
         * $array   查询出来的所有数组
         * order 0 - 不变     1- 反序
         */
    
        public static function page_array($count,$page,$array,$order){
            global $countpage; #定全局变量
            $page=(empty($page))?'1':$page; #判断当前页面是否为空 如果为空就表示为第一页面
            $start=($page-1)*$count; #计算每次分页的开始位置
            if($order==1){
                $array=array_reverse($array);
            }
            $totals=count($array);
            self::$countpage=ceil($totals/$count); #计算总页面数
            $pagedata=array();
            $pagedata=array_slice($array,$start,$count);
            return $pagedata;  #返回查询数据
        }
    
    
    }

swoole多进程循环执行任务demo代码：

<?php
 /**
 *
 * swoole 多进程循环处理数据
 * 适用cli脚本
 *
 */
 
 include 'Mysql.php';
 include 'PageUtils.php';
 
 class cli_run{
 
 public function process_demo()
 {
 
 echo '当前占用内存大小：' . memory_get_usage()/1024 . "kb\n";
 //数据库
 $mysql = new Mysql('数据库链接', '帐号','密码','utf8','数据库名称');
 $min_id = 0;
 $flag = false;
 $redirectStdout = false;
 do{
 
 $workers = array();
 $flag = false;
 $all_PicList = $mysql->select_all("SELECT * FROM table WHERE id > {$min_id} AND status = 1 limit 1000");
 
 if(empty($all_PicList)){
 
 die('finish');
 }
 
 $limit = 100;
 $all_page = ceil(count($all_PicList)/$limit);
 //分10 个进程，每个进程执行100条数据
 for ($i = 1;$i<=$all_page;$i++){
 
 $picList = PageUtils::page_array($limit,$i,$all_PicList,0);
 $process = new swoole_process(function (swoole_process $worker) use ($picList,$mysql) {
 
 foreach ($picList as $v){
 
 //逻辑处理
 
 
 $mysql->select_one('');
 $mysql->update('');
 }
 
 $worker->exit(0);
 
 },$redirectStdout);
 
 //执行fork系统调用，启动进程
 $pid = $process->start();
 // 进程重命名
 $process->name('cli_run_'.$pid);
 // 将每一个进程的句柄存起来
 $workers[$pid] = $process;
 $process = null;
 
 }
 
 if (count($picList) > 0) {
 // 监控/回收子进程
 while (1) {
 $ret = \swoole_process::wait();
 if ($ret) { // $ret 是个数组 code是进程退出状态码，
 $pid = $ret['pid'];
 echo "Worker Exit, PID=" . $pid . PHP_EOL;
 } else {
 $last = array_slice($all_PicList,-1,1);
 //获取最后一个最大的id，继续循环直到数据没有
 $min_id = $last[0]['id'];
 echo '当前占用内存大小：' . memory_get_usage()/1024 . "kb\n";
 $flag = true;
 break;
 }
 }
 }
 
 }while($flag);
 
 }
 
 }

把大量数据先进行limit定量获取，再把取出的数据分配到各个进程去处理任务，缩短了数据处理的时间，比起while(1)单进程任务速度快百倍。