| 
...
 | 
...
 | 
@@ -3,76 +3,49 @@ | 
| 
 | 
 | 
 //error_reporting();
 | 
| 
 | 
 | 
 include_once __DIR__."/../vendor/autoload.php";
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 @file_put_contents('sync_run_my.log',"进程启动 ".getmypid(),FILE_APPEND);
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 swoole_set_process_name('php-email-sync-list-my');
 | 
| 
 | 
 | 
 //TODO:: 不知道为什么,隔断时间mysql会连不上
 | 
| 
 | 
 | 
 \Co\run(function (){
 | 
| 
 | 
 | 
     $start_time = time();
 | 
| 
 | 
 | 
     $goNum = 0;
 | 
| 
 | 
 | 
 // 循环阻塞
 | 
| 
 | 
 | 
     while (true){
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
         if (time() - $start_time > 43200){
 | 
| 
 | 
 | 
             @file_put_contents('sync_run_my.log',"进程停止 ".system("kill ".getmypid()),FILE_APPEND);
 | 
| 
 | 
 | 
             break;
 | 
| 
 | 
 | 
         }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
         if($goNum > 50){
 | 
| 
 | 
 | 
             co::sleep(0.5);
 | 
| 
 | 
 | 
             continue;
 | 
| 
 | 
 | 
         }
 | 
| 
 | 
 | 
 $pm = new \Swoole\Process\Manager();
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
         // 需要同步的id
 | 
| 
 | 
 | 
 $pm->addBatch(30,function ($work_id){
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
     $number = 0;
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
     while (1){
 | 
| 
 | 
 | 
         if($number>500){
 | 
| 
 | 
 | 
             break;
 | 
| 
 | 
 | 
         }
 | 
| 
 | 
 | 
         $id = redis()->lPop('sync_email_lists_my');
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
         if($id && is_numeric($id)){
 | 
| 
 | 
 | 
             // 占用当前的id,占用2小时
 | 
| 
 | 
 | 
             if(redis()->add('just_sync_'.$id,time(),600)){
 | 
| 
 | 
 | 
                 // 启动一个协程
 | 
| 
 | 
 | 
                 go(function () use ($id,&$goNum){
 | 
| 
 | 
 | 
                     $goNum++;
 | 
| 
 | 
 | 
                 $number++;
 | 
| 
 | 
 | 
                 try{
 | 
| 
 | 
 | 
                     // 开始同步
 | 
| 
 | 
 | 
                     $email = db()->cache(3600)->first(\Model\emailSql::first($id));
 | 
| 
 | 
 | 
                     if($email){
 | 
| 
 | 
 | 
                         (new \Service\SyncMail($email))->sync();
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
                     }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
                 }catch (Throwable $e){
 | 
| 
 | 
 | 
                     logs('sync : '.$e->getMessage());
 | 
| 
 | 
 | 
                 }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
                     // 协程完成后执行的函数
 | 
| 
 | 
 | 
                     co::defer(function () use ($id,&$goNum){
 | 
| 
 | 
 | 
                         $goNum--;
 | 
| 
 | 
 | 
                 // 30秒后 消除占用
 | 
| 
 | 
 | 
                 redis()->expire('just_sync_'.$id,120);
 | 
| 
 | 
 | 
                         // 写入日志
 | 
| 
 | 
 | 
                         \Lib\Log::getInstance()->write();
 | 
| 
 | 
 | 
                     });
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
                 });
 | 
| 
 | 
 | 
             }
 | 
| 
 | 
 | 
         }else{
 | 
| 
 | 
 | 
             sleep(1);
 | 
| 
 | 
 | 
         }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
         //每次都暂停1秒,防止同一时间启动太多的任务
 | 
| 
 | 
 | 
         co::sleep(0.1);
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
     }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 //    while ($goNum > 0){
 | 
| 
 | 
 | 
 //        if (time() - $start_time > 24060){ break; }
 | 
| 
 | 
 | 
 //        co::sleep(1);
 | 
| 
 | 
 | 
 //    }
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
     _echo('子进程即将推出');
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 });
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 $pm->start();
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
| 
 | 
 | 
 
 | 
...
 | 
...
 | 
 |