正在显示
1 个修改的文件
包含
204 行增加
和
147 行删除
| @@ -3,185 +3,242 @@ | @@ -3,185 +3,242 @@ | ||
| 3 | include_once "../vendor/autoload.php"; | 3 | include_once "../vendor/autoload.php"; |
| 4 | 4 | ||
| 5 | // 这里试试不用多进程模式，用多协程模式 | 5 | // 这里试试不用多进程模式，用多协程模式 |
| 6 | -function start(){ | ||
| 7 | - _echo('启动邮件群发任务'); | ||
| 8 | - // 删除key | ||
| 9 | - redis()->delete('send_job_is_stop'); | ||
| 10 | - // 开启协程 | ||
| 11 | - \Co\run(function (){ | ||
| 12 | - | ||
| 13 | - $cNum = 0;//协程运行的数量 | ||
| 14 | - $maxRunNum = 500; | ||
| 15 | - while ($maxRunNum){ | ||
| 16 | - $maxRunNum--; | ||
| 17 | - if(!$maxRunNum){ | ||
| 18 | - break; | ||
| 19 | - } | ||
| 20 | - try { | ||
| 21 | - // 是否要停止 | ||
| 22 | - if(redis()->get('send_job_is_stop')=='stop'){ | ||
| 23 | - break; | ||
| 24 | - } | ||
| 25 | 6 | ||
| 26 | - $lists = db()->all(\Model\sendJobsSql::sendList(500)); | ||
| 27 | - $lists = $lists?$lists:[]; | ||
| 28 | - // 循环 | ||
| 29 | - foreach ($lists as $list){ | ||
| 30 | - | ||
| 31 | - // 占用 id | ||
| 32 | - if(redis()->add('send_job_run_id_'.$list['id'],$list['id'],3600)){ | ||
| 33 | - go(function ($data) use (&$cNum){ | ||
| 34 | - $cNum++; // 协程数+1 | ||
| 35 | - // 表单数据 | ||
| 36 | - $data['maildata'] = json_decode($data['maildata'],true); | ||
| 37 | - // 查询邮箱 | ||
| 38 | - $email = db()->first(\Model\emailSql::first($data['email_id'])); | ||
| 39 | - // 更新状态 | ||
| 40 | - \Model\sendJobsSql::upStatus($data['id'],1,db()); | ||
| 41 | - // 是否是单发送 | ||
| 42 | - if($data['maildata']['massSuit']??0){ | ||
| 43 | - $tos = $data['maildata']['tos']; | ||
| 44 | - foreach ($tos as $to){ | ||
| 45 | - | ||
| 46 | - // 是否暂停 | ||
| 47 | - $dst = db()->first(\Model\sendJobsSql::isStatus($data['id'])); | ||
| 48 | - if($dst && $dst['status'] === 3){ | ||
| 49 | - break; | ||
| 50 | - } | ||
| 51 | 7 | ||
| 52 | - // 是否已发送过了 | ||
| 53 | - if(db()->count(\Model\sendJobStatusSql::count($data['id'],$to['email']))){ | ||
| 54 | - continue; | ||
| 55 | - } | ||
| 56 | 8 | ||
| 57 | - // 每个收件人单独发送 | ||
| 58 | - $data['maildata']['tos'] = [$to]; | ||
| 59 | - //替换邮件内容中的指定字段为客户名字 | ||
| 60 | - $data['maildata']['body'] = str_replace('{customer_name}', $to['name'], $data['maildata']['body']); | ||
| 61 | - $result = \Lib\Mail\MailFun::sendEmail($data['maildata'],$email); | ||
| 62 | - | ||
| 63 | - // 插入紫薯精 | ||
| 64 | - db()->insert(\Model\sendJobStatusSql::$table,[ | ||
| 65 | - 'job_id' => $data['id'], | ||
| 66 | - 'to_email' => $to['email'], | ||
| 67 | - 'status' => $result[0] ? 1 : 0, | ||
| 68 | - 'error' => $result[0] ? $result[1] : '' | ||
| 69 | - ]); | ||
| 70 | - | ||
| 71 | - | ||
| 72 | - // 时间距离下次的时间 | ||
| 73 | - if($data['maildata']['masssuit_interval_send']??[]){ | ||
| 74 | - $time = rand($data['maildata']['masssuit_interval_send']['start'],$data['maildata']['masssuit_interval_send']['end']); | ||
| 75 | - if($time){ | ||
| 76 | - $block = false; | ||
| 77 | - while (true){ | ||
| 78 | - // 没5秒循环一次 | ||
| 79 | - if(redis()->get('send_job_is_stop')=='stop'){ | ||
| 80 | - $block = true; | ||
| 81 | - break; | ||
| 82 | - } | ||
| 83 | - $time-=5; | ||
| 84 | - co::sleep(5); | ||
| 85 | - // 执行下一次了 | ||
| 86 | - if (!$time){ | ||
| 87 | - $block = true; | ||
| 88 | - break; | ||
| 89 | - } | ||
| 90 | - } | ||
| 91 | - | ||
| 92 | - if($block){ | ||
| 93 | - break; | ||
| 94 | - } | ||
| 95 | - } | ||
| 96 | - } | 9 | +class SendJob { |
| 97 | 10 | ||
| 98 | - } | 11 | + public $cnum = 0; |
| 99 | 12 | ||
| 100 | - }else{ | ||
| 101 | - $result = \Lib\Mail\MailFun::sendEmail($data['maildata'],$email); | ||
| 102 | - // 更新状态 | ||
| 103 | - db()->update(\Model\sendJobsSql::$table,[ | ||
| 104 | - 'status' => 2, | ||
| 105 | - 'success' => $result[0] ? $data['total'] : 0, | ||
| 106 | - 'error' => $result[0] ? 0 : $data['total'], | ||
| 107 | - ],dbWhere(['id'=>$data['id']])); | ||
| 108 | - // 插入紫薯精 | ||
| 109 | - db()->insert(\Model\sendJobStatusSql::$table,[ | ||
| 110 | - 'job_id' => $data['id'], | ||
| 111 | - 'to_email' => 'all', | ||
| 112 | - 'status' => $result[0] ? 1 : 0, | ||
| 113 | - 'error' => $result[0] ? $result[1] : '' | ||
| 114 | - ]); | ||
| 115 | - } | 13 | + /** |
| 14 | + * 是否停止 | ||
| 15 | + * @return bool | ||
| 16 | + * @author：dc | ||
| 17 | + * @time 2024/4/10 9:12 | ||
| 18 | + */ | ||
| 19 | + private function isStop(){ | ||
| 20 | + return redis()->get('send_job_is_stop') == 'stop'; | ||
| 21 | + } | ||
| 116 | 22 | ||
| 117 | - // 协程结束后 | ||
| 118 | - co::defer(function ($id) use(&$cNum,$data){ | ||
| 119 | - $cNum--; | ||
| 120 | - // 验证是否完成 | ||
| 121 | - if($data['maildata']['massSuit']??0){ | ||
| 122 | - $total = db()->first(\Model\sendJobStatusSql::countSum($data['id'])); | ||
| 123 | - if($total){ | ||
| 124 | - // 更新状态 | ||
| 125 | - db()->update(\Model\sendJobsSql::$table,[ | ||
| 126 | - 'status' => $total['t'] == $data['total'] ? 2 : 0, | ||
| 127 | - 'success' => $total['s'], | ||
| 128 | - 'error' => $total['e'], | ||
| 129 | - ],dbWhere(['id'=>$data['id']])); | ||
| 130 | - } | 23 | + /** |
| 24 | + * 休眠 | ||
| 25 | + * @param float $sleep | ||
| 26 | + * @return bool | ||
| 27 | + * @author：dc | ||
| 28 | + * @time 2024/4/10 9:12 | ||
| 29 | + */ | ||
| 30 | + private function s_sleep(float $sleep):bool { | ||
| 31 | + if($sleep > 0){ | ||
| 32 | + $t = microtime(1); | ||
| 33 | + | ||
| 34 | + while (!$this->isStop()){ | ||
| 35 | + co::sleep(0.1); | ||
| 36 | + if($sleep - (microtime(1)-$t) <= 0){ | ||
| 37 | + break; | ||
| 38 | + } | ||
| 39 | + } | ||
| 40 | + } | ||
| 41 | + return true; | ||
| 42 | + } | ||
| 131 | 43 | ||
| 132 | - } | ||
| 133 | 44 | ||
| 134 | - // 写入日志 | ||
| 135 | - \Lib\Log::getInstance()->write(); | 45 | + public function start(){ |
| 46 | + _echo('启动邮件群发任务 '.getmypid()); | ||
| 47 | + // 删除key | ||
| 48 | + redis()->delete('send_job_is_stop'); | ||
| 136 | 49 | ||
| 137 | - // 结束后要关闭数据库链接，不然链接一直暂用 | ||
| 138 | - db()->close(); | ||
| 139 | - // 删除占用 | ||
| 140 | - redis()->delete('send_job_run_id_'.$data['id']); | ||
| 141 | 50 | ||
| 142 | - redis()->close(); | ||
| 143 | - }); | 51 | + while (1){ |
| 52 | + // 是否要停止 | ||
| 53 | + if($this->isStop()){ | ||
| 54 | + break; | ||
| 55 | + } | ||
| 144 | 56 | ||
| 145 | - },$list); | ||
| 146 | - } | 57 | + $lists = db()->all(\Model\sendJobsSql::sendList(500)); |
| 58 | + $lists = $lists?$lists:[]; | ||
| 147 | 59 | ||
| 60 | + if($lists){ | ||
| 61 | + foreach ($lists as $list){ | ||
| 62 | + $this->go_($list); | ||
| 148 | } | 63 | } |
| 149 | - | ||
| 150 | - }catch (Throwable $e){ | ||
| 151 | - logs($e->getMessage().$e->getTraceAsString()); | 64 | + }else{ |
| 65 | + // 休眠30秒 | ||
| 66 | + $this->s_sleep(30); | ||
| 152 | } | 67 | } |
| 153 | - | ||
| 154 | - \Lib\Log::getInstance()->write(); | ||
| 155 | - // 暂停5秒 | ||
| 156 | - co::sleep(5); | ||
| 157 | } | 68 | } |
| 158 | 69 | ||
| 70 | + | ||
| 159 | // 这个是等待所有协程退出 | 71 | // 这个是等待所有协程退出 |
| 160 | while (true){ | 72 | while (true){ |
| 161 | - if(!$cNum){ | 73 | + _echo('等待协程退出...'); |
| 74 | + if(!$this->cnum){ | ||
| 162 | break; | 75 | break; |
| 163 | } | 76 | } |
| 164 | - co::sleep(0.5); | 77 | + co::sleep(1); |
| 165 | } | 78 | } |
| 166 | 79 | ||
| 80 | + } | ||
| 167 | 81 | ||
| 168 | - }); | ||
| 169 | -} | ||
| 170 | 82 | ||
| 83 | + /** | ||
| 84 | + * @param $list | ||
| 85 | + * @throws \Lib\Err | ||
| 86 | + * @throws \PHPMailer\PHPMailer\Exception | ||
| 87 | + * @author：dc | ||
| 88 | + * @time 2024/4/10 9:25 | ||
| 89 | + */ | ||
| 90 | + public function go_($list){ | ||
| 91 | + // 占用 id | ||
| 92 | + if(redis()->add('send_job_run_id_'.$list['id'],$list['id'],600)){ | ||
| 93 | + go(function ($data) { | ||
| 94 | + _echo('正在执行任务 '.$data['id']); | ||
| 95 | + $this->cnum++; // 协程数+1 | ||
| 96 | + // 表单数据 | ||
| 97 | + $data['maildata'] = json_decode($data['maildata'],true); | ||
| 98 | + // 查询邮箱 | ||
| 99 | + $email = db()->first(\Model\emailSql::first($data['email_id'])); | ||
| 100 | + // 更新状态 | ||
| 101 | + \Model\sendJobsSql::upStatus($data['id'],1,db()); | ||
| 102 | + // 是否是单发送 | ||
| 103 | + if($data['maildata']['massSuit']??0){ | ||
| 104 | + $tos = $data['maildata']['tos']; | ||
| 105 | + foreach ($tos as $to){ | ||
| 106 | + _echo('正在执行任务 发送邮件 '.$to['email']); | ||
| 107 | + // 续时间 | ||
| 108 | + redis()->set('send_job_run_id_'.$data['id'],$data['id'],600); | ||
| 109 | + | ||
| 110 | + // 是否暂停 | ||
| 111 | + $dst = db()->first(\Model\sendJobsSql::isStatus($data['id'])); | ||
| 112 | + if($dst && $dst['status'] === 3){ | ||
| 113 | + break; | ||
| 114 | + } | ||
| 115 | + | ||
| 116 | + // 是否已发送过了 | ||
| 117 | + if(db()->count(\Model\sendJobStatusSql::count($data['id'],$to['email']))){ | ||
| 118 | + continue; | ||
| 119 | + } | ||
| 120 | + | ||
| 121 | + // 每个收件人单独发送 | ||
| 122 | + $data['maildata']['tos'] = [$to]; | ||
| 123 | + //替换邮件内容中的指定字段为客户名字 | ||
| 124 | + $data['maildata']['body'] = str_replace('{customer_name}', $to['name'], $data['maildata']['body']); | ||
| 125 | + $result = \Lib\Mail\MailFun::sendEmail($data['maildata'],$email); | ||
| 126 | + | ||
| 127 | + // 插入紫薯精 | ||
| 128 | + db()->insert(\Model\sendJobStatusSql::$table,[ | ||
| 129 | + 'job_id' => $data['id'], | ||
| 130 | + 'to_email' => $to['email'], | ||
| 131 | + 'status' => $result[0] ? 1 : 0, | ||
| 132 | + 'error' => $result[0] ? $result[1] : '' | ||
| 133 | + ]); | ||
| 134 | + | ||
| 135 | + | ||
| 136 | + // 时间距离下次的时间 | ||
| 137 | + if($data['maildata']['masssuit_interval_send']??[]){ | ||
| 138 | + $time = rand($data['maildata']['masssuit_interval_send']['start'],$data['maildata']['masssuit_interval_send']['end']); | ||
| 139 | + if($time){ | ||
| 140 | + $block = false; | ||
| 141 | + while (true){ | ||
| 142 | + // 没5秒循环一次 | ||
| 143 | + if($this->isStop()){ | ||
| 144 | + $block = true; | ||
| 145 | + break; | ||
| 146 | + } | ||
| 147 | + $time-=5; | ||
| 148 | + $this->s_sleep(5); | ||
| 149 | + // 执行下一次了 | ||
| 150 | + if (!$time){ | ||
| 151 | + $block = true; | ||
| 152 | + break; | ||
| 153 | + } | ||
| 154 | + } | ||
| 155 | + | ||
| 156 | + if($block){ | ||
| 157 | + break; | ||
| 158 | + } | ||
| 159 | + } | ||
| 160 | + } | ||
| 161 | + | ||
| 162 | + } | ||
| 163 | + | ||
| 164 | + } | ||
| 165 | + else{ | ||
| 166 | + $result = \Lib\Mail\MailFun::sendEmail($data['maildata'],$email); | ||
| 167 | + // 更新状态 | ||
| 168 | + db()->update(\Model\sendJobsSql::$table,[ | ||
| 169 | + 'status' => 2, | ||
| 170 | + 'success' => $result[0] ? $data['total'] : 0, | ||
| 171 | + 'error' => $result[0] ? 0 : $data['total'], | ||
| 172 | + ],dbWhere(['id'=>$data['id']])); | ||
| 173 | + // 插入紫薯精 | ||
| 174 | + db()->insert(\Model\sendJobStatusSql::$table,[ | ||
| 175 | + 'job_id' => $data['id'], | ||
| 176 | + 'to_email' => 'all', | ||
| 177 | + 'status' => $result[0] ? 1 : 0, | ||
| 178 | + 'error' => $result[0] ? $result[1] : '' | ||
| 179 | + ]); | ||
| 180 | + } | ||
| 181 | + | ||
| 182 | + // 协程结束后 | ||
| 183 | + co::defer(function ($id) use($data){ | ||
| 184 | + $this->cnum--; | ||
| 185 | + // 验证是否完成 | ||
| 186 | + if($data['maildata']['massSuit']??0){ | ||
| 187 | + $total = db()->first(\Model\sendJobStatusSql::countSum($data['id'])); | ||
| 188 | + if($total){ | ||
| 189 | + // 更新状态 | ||
| 190 | + db()->update(\Model\sendJobsSql::$table,[ | ||
| 191 | + 'status' => $total['t'] == $data['total'] ? 2 : 0, | ||
| 192 | + 'success' => $total['s'], | ||
| 193 | + 'error' => $total['e'], | ||
| 194 | + ],dbWhere(['id'=>$data['id']])); | ||
| 195 | + } | ||
| 196 | + | ||
| 197 | + } | ||
| 171 | 198 | ||
| 199 | + // 写入日志 | ||
| 200 | + \Lib\Log::getInstance()->write(); | ||
| 172 | 201 | ||
| 202 | + // 结束后要关闭数据库链接，不然链接一直暂用 | ||
| 203 | + db()->close(); | ||
| 204 | + // 删除占用 | ||
| 205 | + redis()->delete('send_job_run_id_'.$data['id']); | ||
| 206 | + | ||
| 207 | + redis()->close(); | ||
| 208 | + }); | ||
| 209 | + | ||
| 210 | + },$list); | ||
| 211 | + } | ||
| 212 | + | ||
| 213 | + } | ||
| 214 | + | ||
| 215 | +} | ||
| 173 | 216 | ||
| 174 | 217 | ||
| 175 | $ps = "ps -ef | grep \"send_job.php start\" | grep -v grep | wc -l"; | 218 | $ps = "ps -ef | grep \"send_job.php start\" | grep -v grep | wc -l"; |
| 176 | 219 | ||
| 177 | switch ($argv[1]??0){ | 220 | switch ($argv[1]??0){ |
| 178 | case 'start':{ | 221 | case 'start':{ |
| 179 | -// $num = exec($ps); | ||
| 180 | -// if($num){ | ||
| 181 | -// echo '正则运行，请勿重复运行'; | ||
| 182 | -// }else{ | ||
| 183 | - start(); | ||
| 184 | -// } | 222 | + |
| 223 | + // 开启协程 | ||
| 224 | + \Co\run(function (){ | ||
| 225 | + | ||
| 226 | + $handler = function ($signal){ | ||
| 227 | + // 可以处理其他程序 | ||
| 228 | + redis()->set('send_job_is_stop','stop'); | ||
| 229 | + | ||
| 230 | + _echo('收到退出信号 '.$signal); | ||
| 231 | + }; | ||
| 232 | + | ||
| 233 | + \Swoole\Process::signal(SIGTERM,$handler); | ||
| 234 | + \Swoole\Process::signal(SIGINT,$handler); | ||
| 235 | + | ||
| 236 | + (new SendJob)->start(); | ||
| 237 | + | ||
| 238 | + _echo('进程已退出'); | ||
| 239 | + | ||
| 240 | + }); | ||
| 241 | + | ||
| 185 | break; | 242 | break; |
| 186 | } | 243 | } |
| 187 | case 'stop':{ | 244 | case 'stop':{ |
-
请 注册 或 登录 后发表评论