周海龙 / mail-serve · 提交

转到一个项目

作者 邓超
提交 d44057d99faf013284d7913bde605ba5fc892b12 1 个父辈 60a9aaa0

es

内嵌 并排对比
正在显示 1 个修改的文件 包含 130 行增加 和 22 行删除
... ... @@ -83,17 +83,20 @@ class SyncToEsCmd {
return $this->isStop;
}
/**
* es链接
* @var \Lib\Es\Es
*/
public $es;
public function handler(){
// $es = es(); // 第一个库，即将丢弃
$es2 = es('email_lists_copy'); // 第二个库 新
$this->es = es('email_lists_copy'); // 第二个库 新
while (!$this->isStop()){
$id = redis()->lPop('sync_to_es');
$origin_id = $id;
$code = 500;
if($id){
$is_check_body = false;
... ... @@ -102,17 +105,40 @@ class SyncToEsCmd {
$is_check_body = true;
}
$data = $this->getDataByEs($id,$is_check_body);
if($data){
list($doc_id,$data) = $data;
// 主库
$this->toDataEs('email_lists_copy',$id,$doc_id,$data);
// 分库
$this->toDataEs('email_lists_branch_'.$data['postid'],$id,$doc_id,$data);
}
}else{
// 没有数据时暂停1秒
sleep(1);
}
}
}
$doc_id = '';
/**
* @param $id
* @param $is_check_body
* @return array|false
* @author：dc
* @time 2025/8/5 10:21
*/
public function getDataByEs($id,$is_check_body) {
try {
$data = $this->db->throw()->first(\Model\listsSql::first('`id` = '.$id));
if(!$data){
$data = $this->db->throw()->first(\Model\listsSql::firstHot('`id` = '.$id));
}
}catch (Throwable $e){
redis()->rPush('sync_to_es',$origin_id);
_echo('sync to es '.$e->getMessage());
break;
$this->log($id);
// redis()->rPush('sync_to_es',$origin_id);
_echo('sync to es '.$id.":".$e->getMessage());
return false;
}
if($data){
... ... @@ -144,35 +170,117 @@ class SyncToEsCmd {
list($data['postid'],$data['source']) = $this->getPostid($data['email_id'],$data['udate']);
}catch (Throwable $e){
redis()->rPush('sync_to_es',$origin_id);
_echo('sync to es '.$e->getMessage());
break;
$this->log($id);
// redis()->rPush('sync_to_es',$origin_id);
_echo('sync to es '.$id.":".$e->getMessage());
return false;
}
$data = $this->getEsData($data);
$doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
// 新
$code = $es2->save($doc_id,$data);
$doc_id = $data['email_id'].'_'.$data['folder_id'].'_'.$data['uid'];
// 这个验证数据没问题后会丢弃
// $code = $es->save($doc_id,$data);
return [$doc_id,$data];
}
if($code!==200){
@file_put_contents(LOG_PATH.'/sync_es_fail.log',$id."\n",FILE_APPEND);
_echo('同步es: '.$doc_id.'===>'.$code);
return false;
}
}else{
sleep(1);
public $checkEsIndex = [];
/**
* 同步数据到es
* @param string $index 索引名称
* @param int $data_id list表的id
* @param string $doc_id es文档id
* @param array $data es数据
* @author：dc
* @time 2025/8/5 10:09
*/
public function toDataEs($index,$data_id,$doc_id,$data){
// echo $index;
// echo ' ==== ';
$this->es->setIndex($index);
// 检查数据库是否存在
if(empty($this->checkEsIndex[$index])){
if(!$this->es->getMapping()){
$map = $this->es->putMapping([
'properties' => [
'subject' => ['type' => 'text'],
'from' => [
'type' => 'object', // 定义 from 字段为对象
'properties' => [
'name' => [
'type' => 'keyword' // 或者 'keyword'，根据需求选择
],
'email' => [
'type' => 'text' // email 通常使用 keyword 类型
]
]
],
'to' => ['type' => 'text'],
'cc' => ['type' => 'keyword'],
'bcc' => ['type' => 'keyword'],
'uid' => ['type' => 'integer'],
'udate' => ['type' => 'integer'],
'folder_id' => ['type' => 'integer'],
'email_id' => ['type' => 'integer'],
'size' => ['type' => 'integer'],
'recent' => ['type' => 'integer'],
'flagged' => ['type' => 'integer'],
'deleted' => ['type' => 'integer'],
'seen' => ['type' => 'integer'],
'draft' => ['type' => 'integer'],
'is_file' => ['type' => 'integer'],
'is_hots' => ['type' => 'integer'],
'is_auto' => ['type' => 'integer'],
'folder_as_int' => ['type' => 'integer'],
'postid' => ['type' => 'integer'],
'source' => ['type' => 'integer'],
'created_at' => ['type' => 'date'],
'updated_at' => ['type' => 'date'],
'description' => ['type' => 'keyword'],
'references' => ['type' => 'keyword']
]
],$index == 'email_lists_copy' ? [
'number_of_shards' => 21, // 设置分片数
'number_of_replicas' => 1, // 设置副本数 暂用内存 主片+副片*
]:[
'number_of_shards' => 1, // 设置分片数
'number_of_replicas' => 0, // 设置副本数 暂用内存 主片+副片*
]
);
_echo("{$index} 创建索引 ".$map);
}
}
// 下次不在检查
$this->checkEsIndex[$index] = 1;
$code = $this->es->save($doc_id,$data);
if($code!==200){
$this->log($data_id,$index);
@file_put_contents(LOG_PATH.'/sync_es_fail.log',$index.":".$data_id."\n",FILE_APPEND);
_echo($index.': '.$data_id.'===>'.$code);
}
}
/**
* 记录日志
* @param $id
* @param string $index
* @author：dc
* @time 2025/8/5 10:17
*/
public function log($id,$index=''){
@file_put_contents(LOG_PATH.'/sync_es_fail.log',$index.":".$id."\n",FILE_APPEND);
}
/**
* 项目id
* @author：dc
... ...