作者 邓超

sync es

... ... @@ -4,57 +4,101 @@
require_once "../vendor/autoload.php";
function getEsData($data){
if(!empty($data['id'])){
$data['uuid'] = $data['id'];
unset($data['id']);
}
if(!empty($data['from'])){
$data['from'] = [
'email' => $data['from'],
'name' => $data['from_name']??''
];
unset($data['from_name']);
}
unset($data['date']);
if(!empty($data['created_at'])){
$data['created_at'] = date('Y-m-d\TH:i:s',strtotime($data['created_at']));
}
if(!empty($data['updated_at'])){
$data['updated_at'] = date('Y-m-d\TH:i:s',strtotime($data['updated_at']));
}
$data['references'] = empty($data['references']) ? '' : $data['references'];
return $data;
}
/**
* 把mysql的数据同步到es
* @author:dc
* @time 2025/3/4 10:19
* Class SyncToEsCmd
*/
class SyncToEsCmd {
public $isStop = false;
public function handler(){
$handler = function ($signal){
_echo('收到进程信号 '. $signal);
// 可以处理其他程序
$this->isStop = true;
};
pcntl_signal(SIGTERM, $handler); // 这个是kill
pcntl_signal(SIGINT, $handler); // 这个是 ctrl+c
// pcntl_signal(SIGHUP, $handler);
$es = (new Lib\Es\Es('hg_ai_emails'));
$db = db();
while (1){
// 检查是否接收到信号
pcntl_signal_dispatch();
if($this->isStop) {
_echo('已退出进程');
break;
}
$id = redis()->lPop('sync_to_es');
if($id){
$data = $db->first(\Model\listsSql::first('`id` = '.$id));
if($data){
// 设置 进程 是否在运行
redis()->set('sync_my_pid:'.getmypid(),time(),86400);
$db = db();
$es = (new Lib\Es\Es(''));
while (1){
$id = redis()->lPop('sync_to_es');
if($id){
$data = $db->first(\Model\listsSql::first('`id` = '.$id));
if($data){
$data['is_auto']=$db->count('select count(*) from `lists_auto` where `list_id` = '.$data['id']) ? 1 : 0;
$data['is_auto']=$db->count('select count(*) from `lists_auto` where `list_id` = '.$data['id']) ? 1 : 0;
$data = $this->getEsData($data);
$doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
$data = getEsData($data);
$doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
$code = $es->save($doc_id,$data);
if($code!==200){
sleep(1);
$code = $es->save($doc_id,$data); // 重试一次
}
$code = $es->save($doc_id,$data);
if($code!==200){
sleep(1);
$code = $es->save($doc_id,$data); // 重试一次
_echo('同步es: '.$doc_id.'===>'.$code);
continue;
}
}
_echo('同步es: '.$doc_id.'===>'.$code);
continue;
echo '没有找到数据'.PHP_EOL;
sleep(1);
}
}
echo '没有找到数据'.PHP_EOL;
sleep(1);
public function getEsData($data){
if(!empty($data['id'])){
$data['uuid'] = $data['id'];
unset($data['id']);
}
if(!empty($data['from'])){
$data['from'] = [
'email' => $data['from'],
'name' => $data['from_name']??''
];
unset($data['from_name']);
}
unset($data['date']);
if(!empty($data['created_at'])){
$data['created_at'] = date('Y-m-d\TH:i:s',strtotime($data['created_at']));
}
if(!empty($data['updated_at'])){
$data['updated_at'] = date('Y-m-d\TH:i:s',strtotime($data['updated_at']));
}
$data['references'] = empty($data['references']) ? '' : $data['references'];
return $data;
}
}
(new SyncToEsCmd())->handler();
return 1;
... ...