|
|
<?php
|
|
|
|
|
|
//error_reporting();
|
|
|
include_once __DIR__."/../vendor/autoload.php";
|
|
|
|
|
|
use Swoole\Process;
|
|
|
|
|
|
swoole_set_process_name('php-email-sync-list');
|
|
|
|
|
|
\Co\run(function (){
|
|
|
|
|
|
function start(){
|
|
|
$goNum = 0;
|
|
|
// 循环阻塞
|
|
|
while (true){
|
|
|
|
|
|
// 删除停止运行的值
|
|
|
// redis()->delete(SYNC_RUNNING_REDIS_KEY,'email_sync_stop_num');
|
|
|
|
|
|
// 进程管理器
|
|
|
$pm = new Process\Manager();
|
|
|
|
|
|
// 启动业务进程
|
|
|
$pm->addBatch(10,function (Process\Pool $pool, int $worker_id){
|
|
|
|
|
|
swoole_set_process_name('php-email-sync-list-'.$worker_id);
|
|
|
|
|
|
include_once __DIR__."/../vendor/autoload.php";
|
|
|
_echo("业务进程({$worker_id})启动成功");
|
|
|
|
|
|
$goNum = 0;
|
|
|
$start_time = time();
|
|
|
// 循环阻塞
|
|
|
while (true){
|
|
|
// 每执行 12小时 就重启任务
|
|
|
if((time()-$start_time) > 43200){
|
|
|
if($goNum < 1 || (time()-$start_time) > 43300) break;
|
|
|
co::sleep(1);
|
|
|
continue;
|
|
|
}
|
|
|
if($goNum > 50){
|
|
|
co::sleep(0.5);
|
|
|
continue;
|
|
|
// break;
|
|
|
}
|
|
|
if($goNum > 50){
|
|
|
co::sleep(0.5);
|
|
|
continue;
|
|
|
}
|
|
|
|
|
|
// 需要同步的id
|
|
|
$id = redis()->lPop('sync_email_lists');
|
|
|
|
|
|
if($id && is_numeric($id)){
|
|
|
// 占用当前的id,占用2小时
|
|
|
if(redis()->add('just_sync_'.$id,time(),600)){
|
|
|
// 启动一个协程
|
|
|
go(function () use ($id,&$goNum){
|
|
|
$goNum++;
|
|
|
try{
|
|
|
// 开始同步
|
|
|
$email = db()->cache(3600)->first(\Model\emailSql::first($id));
|
|
|
if($email){
|
|
|
$sync = new \Service\SyncMail($email);
|
|
|
$search = new \Lib\Imap\ImapSearch();
|
|
|
// 第一次同步 只同步当天的
|
|
|
if(!db()->cache(600)->count(\Model\listsSql::first('`id` > 0'))){
|
|
|
$sync->search($search->dateGt($email['created_at']));
|
|
|
}else{
|
|
|
// 是否是ai邮箱
|
|
|
if(db()->count("select count(*) from `hot_mail` where `email` = '{$email['email']}'")){
|
|
|
// ai邮件只同步2天内的
|
|
|
if(strtotime("-2 day") > strtotime($email['created_at'])){
|
|
|
$sync->search(
|
|
|
$search->dateGt(
|
|
|
date('Y-m-d',
|
|
|
strtotime("-2 day")
|
|
|
)
|
|
|
// 需要同步的id
|
|
|
$id = redis()->lPop('sync_email_lists');
|
|
|
|
|
|
if($id && is_numeric($id)){
|
|
|
// 占用当前的id,占用2小时
|
|
|
if(redis()->add('just_sync_'.$id,time(),600)){
|
|
|
// 启动一个协程
|
|
|
go(function () use ($id,&$goNum){
|
|
|
$goNum++;
|
|
|
try{
|
|
|
// 开始同步
|
|
|
$email = db()->cache(3600)->first(\Model\emailSql::first($id));
|
|
|
if($email){
|
|
|
$sync = new \Service\SyncMail($email);
|
|
|
$search = new \Lib\Imap\ImapSearch();
|
|
|
// 第一次同步 只同步当天的
|
|
|
if(!db()->cache(600)->count(\Model\listsSql::first('`id` > 0'))){
|
|
|
$sync->search($search->dateGt($email['created_at']));
|
|
|
}else{
|
|
|
// 是否是ai邮箱
|
|
|
if(db()->count("select count(*) from `hot_mail` where `email` = '{$email['email']}'")){
|
|
|
// ai邮件只同步2天内的
|
|
|
if(strtotime("-2 day") > strtotime($email['created_at'])){
|
|
|
$sync->search(
|
|
|
$search->dateGt(
|
|
|
date('Y-m-d',
|
|
|
strtotime("-2 day")
|
|
|
)
|
|
|
);
|
|
|
}else{
|
|
|
$sync->search(
|
|
|
$search->dateGt($email['created_at'])
|
|
|
);
|
|
|
}
|
|
|
|
|
|
)
|
|
|
);
|
|
|
}else{
|
|
|
$sync->search(
|
|
|
$search->dateGt($email['created_at'])
|
|
|
);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
$sync->sync();
|
|
|
|
|
|
$sync = null;
|
|
|
unset($sync);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
}catch (Throwable $e){
|
|
|
logs('sync : '.$e->getMessage());
|
|
|
$sync->sync();
|
|
|
|
|
|
$sync = null;
|
|
|
unset($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();
|
|
|
});
|
|
|
|
|
|
// 协程完成后执行的函数
|
|
|
co::defer(function () use ($id,&$goNum){
|
|
|
$goNum--;
|
|
|
// 30秒后 消除占用
|
|
|
redis()->expire('just_sync_'.$id,120);
|
|
|
// 写入日志
|
|
|
\Lib\Log::getInstance()->write();
|
|
|
});
|
|
|
}
|
|
|
}
|
|
|
|
|
|
//每次都暂停1秒,防止同一时间启动太多的任务
|
|
|
co::sleep(0.1);
|
|
|
|
|
|
|
|
|
});
|
|
|
}
|
|
|
}
|
|
|
|
|
|
},true);
|
|
|
//每次都暂停1秒,防止同一时间启动太多的任务
|
|
|
co::sleep(0.1);
|
|
|
|
|
|
|
|
|
// 启动管理器
|
|
|
$pm->start();
|
|
|
}
|
|
|
|
|
|
}
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
|
|
start();
|
|
|
|
|
|
|
|
|
|
...
|
...
|
|