From 5aed9ce29eb1d3f2b769dce490d74fef2364e65d Mon Sep 17 00:00:00 2001 From: kuaifan Date: Fri, 18 Apr 2025 12:56:21 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E4=BC=98=E5=8C=96=E5=85=A8=E6=96=87?= =?UTF-8?q?=E6=90=9C=E7=B4=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...Search.php => SyncUserMsgToZincSearch.php} | 47 +++-- app/Http/Controllers/Api/DialogController.php | 6 +- ...logUserMsg.php => ZincSearchDialogMsg.php} | 184 +++++++++++++++--- app/Observers/WebSocketDialogMsgObserver.php | 8 +- app/Observers/WebSocketDialogUserObserver.php | 8 +- 5 files changed, 195 insertions(+), 58 deletions(-) rename app/Console/Commands/{SyncUserMsgToSearch.php => SyncUserMsgToZincSearch.php} (61%) rename app/Module/ZincSearch/{ZincSearchDialogUserMsg.php => ZincSearchDialogMsg.php} (70%) diff --git a/app/Console/Commands/SyncUserMsgToSearch.php b/app/Console/Commands/SyncUserMsgToZincSearch.php similarity index 61% rename from app/Console/Commands/SyncUserMsgToSearch.php rename to app/Console/Commands/SyncUserMsgToZincSearch.php index ea859478f..b5662eebc 100644 --- a/app/Console/Commands/SyncUserMsgToSearch.php +++ b/app/Console/Commands/SyncUserMsgToZincSearch.php @@ -4,10 +4,11 @@ namespace App\Console\Commands; use App\Models\WebSocketDialogMsg; use App\Module\ZincSearch\ZincSearchKeyValue; -use App\Module\ZincSearch\ZincSearchDialogUserMsg; +use App\Module\ZincSearch\ZincSearchDialogMsg; +use Cache; use Illuminate\Console\Command; -class SyncUserMsgToSearch extends Command +class SyncUserMsgToZincSearch extends Command { /** * 更新数据 @@ -18,7 +19,7 @@ class SyncUserMsgToSearch extends Command * --c: 清除索引 */ - protected $signature = 'search:sync-user-msg {--f} {--i} {--c} {--batch=1000}'; + protected $signature = 'zinc:sync-user-msg {--f} {--i} {--c} {--batch=1000}'; protected $description = '同步聊天会话用户和消息到 ZincSearch'; /** @@ -26,23 +27,35 @@ class SyncUserMsgToSearch extends Command */ public function handle(): int { - // 清除索引 - if ($this->option('c')) { - $this->info('清除索引...'); - ZincSearchKeyValue::clear(); - ZincSearchDialogUserMsg::clear(); - $this->info("索引删除成功"); - return 0; + // 使用缓存锁确保一次只能运行一个实例 + $lock = Cache::lock('zinc:sync-user-msg', 3600 * 6); // 锁定6小时 + if (!$lock->get()) { + $this->error('命令已在运行中，请等待当前实例完成'); + return 1; } - $this->info('开始同步聊天数据...'); + try { + // 清除索引 + if ($this->option('c')) { + $this->info('清除索引...'); + ZincSearchKeyValue::clear(); + ZincSearchDialogMsg::clear(); + $this->info("索引删除成功"); + return 0; + } - // 同步消息数据 - $this->syncDialogMsgs(); + $this->info('开始同步聊天数据...'); - // 完成 - $this->info("\n同步完成"); - return 0; + // 同步消息数据 + $this->syncDialogMsgs(); + + // 完成 + $this->info("\n同步完成"); + return 0; + } finally { + // 确保无论如何都会释放锁 + $lock->release(); + } } /** @@ -78,7 +91,7 @@ class SyncUserMsgToSearch extends Command $this->info("{$num}/{$count} ({$progress}%) 正在同步消息ID {$lastId} ~ {$dialogMsgs->last()->id}"); // 同步数据 - ZincSearchDialogUserMsg::batchSync($dialogMsgs); + ZincSearchDialogMsg::batchSync($dialogMsgs); // 更新最后ID $lastId = $dialogMsgs->last()->id; diff --git a/app/Http/Controllers/Api/DialogController.php b/app/Http/Controllers/Api/DialogController.php index d506346d2..f41f9a084 100755 --- a/app/Http/Controllers/Api/DialogController.php +++ b/app/Http/Controllers/Api/DialogController.php @@ -28,7 +28,7 @@ use App\Models\WebSocketDialogMsgTodo; use App\Models\WebSocketDialogMsgTranslate; use App\Models\WebSocketDialogSession; use App\Module\Table\OnlineData; -use App\Module\ZincSearch\ZincSearchDialogUserMsg; +use App\Module\ZincSearch\ZincSearchDialogMsg; use Hhxsv5\LaravelS\Swoole\Task\Task; /** @@ -174,7 +174,7 @@ class DialogController extends AbstractController } // 搜索消息会话 if (count($list) < 20) { - $searchResults = ZincSearchDialogUserMsg::searchByKeyword($user->userid, $key, 0, 20 - count($list)); + $searchResults = ZincSearchDialogMsg::search($user->userid, $key, 0, 20 - count($list)); if ($searchResults) { foreach ($searchResults as $item) { if ($dialog = WebSocketDialog::find($item['id'])) { @@ -728,7 +728,7 @@ class DialogController extends AbstractController $key = trim(Request::input('key')); $list = []; // - $searchResults = ZincSearchDialogUserMsg::searchByKeyword($user->userid, $key, 0, Base::getPaginate(50, 20)); + $searchResults = ZincSearchDialogMsg::search($user->userid, $key, 0, Base::getPaginate(50, 20)); if ($searchResults) { foreach ($searchResults as $item) { if ($dialog = WebSocketDialog::find($item['id'])) { diff --git a/app/Module/ZincSearch/ZincSearchDialogUserMsg.php b/app/Module/ZincSearch/ZincSearchDialogMsg.php similarity index 70% rename from app/Module/ZincSearch/ZincSearchDialogUserMsg.php rename to app/Module/ZincSearch/ZincSearchDialogMsg.php index acf70aab3..8b286687b 100644 --- a/app/Module/ZincSearch/ZincSearchDialogUserMsg.php +++ b/app/Module/ZincSearch/ZincSearchDialogMsg.php @@ -4,7 +4,9 @@ namespace App\Module\ZincSearch; use App\Models\WebSocketDialogMsg; use App\Models\WebSocketDialogUser; +use Carbon\Carbon; use Illuminate\Support\Facades\Log; +use Swoole\Coroutine; /** * ZincSearch 会话消息类 @@ -21,9 +23,9 @@ use Illuminate\Support\Facades\Log; * - 单个同步: sync(WebSocketDialogMsg $dialogMsg); * - 批量同步: batchSync(WebSocketDialogMsg[] $dialogMsgs); * - 用户同步: userSync(WebSocketDialogUser $dialogUser); - * - 删除消息: delete(WebSocketDialogMsg|WebSocketDialogUser $data); + * - 删除消息: delete(WebSocketDialogMsg|WebSocketDialogUser|int $data); */ -class ZincSearchDialogUserMsg +class ZincSearchDialogMsg { /** * 索引名称 @@ -45,6 +47,7 @@ class ZincSearchDialogUserMsg 'properties' => [ // 拓展数据 'dialog_userid' => ['type' => 'keyword', 'index' => true], // 对话ID+用户ID + 'to_userid' => ['type' => 'numeric', 'index' => true], // 此消息发给的用户ID // 消息数据 'id' => ['type' => 'numeric', 'index' => true], @@ -130,8 +133,7 @@ class ZincSearchDialogUserMsg 'query' => [ 'bool' => [ 'must' => [ - ['term' => ['userid' => $userid]], - ['term' => ['bot' => 0]], + ['term' => ['to_userid' => $userid]], ['match_phrase' => ['key' => $keyword]] ] ] @@ -145,18 +147,84 @@ class ZincSearchDialogUserMsg try { $result = ZincSearchBase::elasticSearch(self::$indexNameMsg, $searchParams); - return array_map(function ($hit) { - // todo 格式化消息 - return $hit['_source']; - }, $result['data']['hits']['hits'] ?? []); + $hits = $result['data']['hits']['hits'] ?? []; + + // 收集所有的用户信息 + $dialogUserids = []; + foreach ($hits as $hit) { + $source = $hit['_source']; + $dialogUserids[] = $source['dialog_userid']; + } + $userInfos = self::searchUser(array_unique($dialogUserids)); + + // 组合返回结果，将用户信息合并到消息中 + $msgs = []; + foreach ($hits as $hit) { + $msgInfo = $hit['_source']; + $userInfo = $userInfos[$msgInfo['dialog_userid']] ?? []; + if ($userInfo) { + $msgs[] = [ + 'id' => $msgInfo['dialog_id'], + 'search_msg_id' => $msgInfo['id'], + 'user_at' => Carbon::parse($msgInfo['updated_at'])->format('Y-m-d H:i:s'), + + 'mark_unread' => $userInfo['mark_unread'], + 'silence' => $userInfo['silence'], + 'hide' => $userInfo['hide'], + 'color' => $userInfo['color'], + 'top_at' => Carbon::parse($userInfo['top_at'])->format('Y-m-d H:i:s'), + 'last_at' => Carbon::parse($userInfo['last_at'])->format('Y-m-d H:i:s'), + ]; + } + } + return $msgs; } catch (\Exception $e) { - Log::error('搜索对话消息失败: ' . $e->getMessage()); - return [ - 'success' => false, - 'error' => $e->getMessage(), - 'hits' => ['total' => ['value' => 0], 'hits' => []] + Log::error('search: ' . $e->getMessage()); + return []; + } + } + + /** + * 根据对话用户ID搜索用户信息 + * @param array $dialogUserids + * @return array + */ + private static function searchUser(array $dialogUserids): array + { + if (empty($dialogUserids)) { + return []; + } + + $userInfos = []; + + // 构建用户查询条件 + $userSearchParams = [ + 'query' => [ + 'bool' => [ + 'should' => [] + ] + ], + 'size' => count($dialogUserids) // 确保取到所有符合条件的记录 + ]; + + // 添加所有 dialog_userid 到查询条件 + foreach ($dialogUserids as $dialogUserid) { + $userSearchParams['query']['bool']['should'][] = [ + 'term' => ['dialog_userid' => $dialogUserid] ]; } + + // 查询用户信息 + $userResult = ZincSearchBase::elasticSearch(self::$indexNameUser, $userSearchParams); + $userHits = $userResult['data']['hits']['hits'] ?? []; + + // 以 dialog_userid 为键保存用户信息 + foreach ($userHits as $userHit) { + $userSource = $userHit['_source']; + $userInfos[$userSource['dialog_userid']] = $userSource; + } + + return $userInfos; } // ============================== @@ -164,16 +232,11 @@ class ZincSearchDialogUserMsg // ============================== /** - * 生成文档ID + * 生成 dialog_userid * - * @param WebSocketDialogMsg $dialogMsg - * @param int $userid + * @param WebSocketDialogUser $dialogUser * @return string */ - private static function generateDocId(WebSocketDialogMsg $dialogMsg, int $userid): string - { - return "{$dialogMsg->id}_{$userid}"; - } private static function generateDialogUserid(WebSocketDialogUser $dialogUser): string { return "{$dialogUser->dialog_id}_{$dialogUser->userid}"; @@ -189,8 +252,9 @@ class ZincSearchDialogUserMsg private static function generateMsgData(WebSocketDialogMsg $dialogMsg, WebSocketDialogUser $dialogUser): array { return [ - '_id' => self::generateDocId($dialogMsg, $dialogUser->userid), + '_id' => self::$indexNameMsg . "_" . $dialogMsg->id . "_" . $dialogUser->userid, 'dialog_userid' => self::generateDialogUserid($dialogUser), + 'to_userid' => $dialogUser->userid, 'id' => $dialogMsg->id, 'dialog_id' => $dialogMsg->dialog_id, @@ -206,7 +270,7 @@ class ZincSearchDialogUserMsg private static function generateUserData(WebSocketDialogUser $dialogUser): array { return [ - '_id' => $dialogUser->id, + '_id' => self::$indexNameUser . "_" . $dialogUser->id, 'dialog_userid' => self::generateDialogUserid($dialogUser), 'id' => $dialogUser->id, @@ -275,9 +339,9 @@ class ZincSearchDialogUserMsg return true; } catch (\Exception $e) { - Log::error('syncMsg: ' . $e->getMessage()); + Log::error('sync: ' . $e->getMessage()); + return false; } - return false; } /** @@ -352,7 +416,7 @@ class ZincSearchDialogUserMsg } } catch (\Exception $e) { - Log::error('batchSyncMsgs: ' . $e->getMessage()); + Log::error('batchSync: ' . $e->getMessage()); } return $count; @@ -369,14 +433,71 @@ class ZincSearchDialogUserMsg return false; } $data = self::generateUserData($dialogUser); - $result = ZincSearchBase::addDoc(self::$indexNameUser, $data); - return $result['success'] ?? false; + + // 生成查询用户条件 + $searchParams = [ + 'query' => [ + 'bool' => [ + 'must' => [ + ['term' => ['dialog_userid' => $data['dialog_userid']]] + ] + ] + ], + 'size' => 1 + ]; + + try { + // 查询用户是否存在 + $result = ZincSearchBase::elasticSearch(self::$indexNameUser, $searchParams); + $hits = $result['data']['hits']['hits'] ?? []; + + // 同步用户（存在更新、不存在添加） + $result = ZincSearchBase::addDoc(self::$indexNameUser, $data); + if (!isset($result['success'])) { + return false; + } + + // 用户不存在，同步消息 + if (empty($hits)) { + go(function () use ($dialogUser) { + Coroutine::sleep(0.1); + + $lastId = 0; // 上次同步的最后ID + $batchSize = 500; // 每批处理的消息数量 + + // 分批同步消息 + do { + // 获取一批 + $dialogMsgs = WebSocketDialogMsg::whereDialogId($dialogUser->dialog_id) + ->where('id', '>', $lastId) + ->orderBy('id') + ->limit($batchSize) + ->get(); + + if ($dialogMsgs->isEmpty()) { + break; + } + + // 同步数据 + ZincSearchDialogMsg::batchSync($dialogMsgs); + + // 更新最后ID + $lastId = $dialogMsgs->last()->id; + } while (count($dialogMsgs) == $batchSize); + }); + } + + return true; + } catch (\Exception $e) { + Log::error('userSync: ' . $e->getMessage()); + return false; + } } /** * 删除 * - * @param WebSocketDialogMsg|WebSocketDialogUser $data + * @param WebSocketDialogMsg|WebSocketDialogUser|int $data * @return int */ public static function delete(mixed $data): int @@ -397,7 +518,10 @@ class ZincSearchDialogUserMsg 'term' => self::generateDialogUserid($data), ]; } else { - return 0; + $query = [ + 'field' => 'id', + 'term' => (string) $data + ]; } try { @@ -433,7 +557,7 @@ class ZincSearchDialogUserMsg $from += $batchSize; } } catch (\Exception $e) { - Log::error('deleteMsg: ' . $e->getMessage()); + Log::error('delete: ' . $e->getMessage()); } return $totalDeleted; diff --git a/app/Observers/WebSocketDialogMsgObserver.php b/app/Observers/WebSocketDialogMsgObserver.php index 89046f83b..2d813d028 100644 --- a/app/Observers/WebSocketDialogMsgObserver.php +++ b/app/Observers/WebSocketDialogMsgObserver.php @@ -3,7 +3,7 @@ namespace App\Observers; use App\Models\WebSocketDialogMsg; -use App\Module\ZincSearch\ZincSearchDialogUserMsg; +use App\Module\ZincSearch\ZincSearchDialogMsg; class WebSocketDialogMsgObserver { @@ -15,7 +15,7 @@ class WebSocketDialogMsgObserver */ public function created(WebSocketDialogMsg $webSocketDialogMsg) { - ZincSearchDialogUserMsg::syncMsg($webSocketDialogMsg); + ZincSearchDialogMsg::sync($webSocketDialogMsg); } /** @@ -26,7 +26,7 @@ class WebSocketDialogMsgObserver */ public function updated(WebSocketDialogMsg $webSocketDialogMsg) { - ZincSearchDialogUserMsg::syncMsg($webSocketDialogMsg); + ZincSearchDialogMsg::sync($webSocketDialogMsg); } /** @@ -37,7 +37,7 @@ class WebSocketDialogMsgObserver */ public function deleted(WebSocketDialogMsg $webSocketDialogMsg) { - ZincSearchDialogUserMsg::deleteMsg($webSocketDialogMsg); + ZincSearchDialogMsg::delete($webSocketDialogMsg); } /** diff --git a/app/Observers/WebSocketDialogUserObserver.php b/app/Observers/WebSocketDialogUserObserver.php index c8329f706..46d72a0ff 100644 --- a/app/Observers/WebSocketDialogUserObserver.php +++ b/app/Observers/WebSocketDialogUserObserver.php @@ -4,7 +4,7 @@ namespace App\Observers; use App\Models\Deleted; use App\Models\WebSocketDialogUser; -use App\Module\ZincSearch\ZincSearchDialogUserMsg; +use App\Module\ZincSearch\ZincSearchDialogMsg; use Carbon\Carbon; class WebSocketDialogUserObserver @@ -30,7 +30,7 @@ class WebSocketDialogUserObserver } } Deleted::forget('dialog', $webSocketDialogUser->dialog_id, $webSocketDialogUser->userid); - ZincSearchDialogUserMsg::syncUser($webSocketDialogUser); + ZincSearchDialogMsg::userSync($webSocketDialogUser); } /** @@ -41,7 +41,7 @@ class WebSocketDialogUserObserver */ public function updated(WebSocketDialogUser $webSocketDialogUser) { - ZincSearchDialogUserMsg::syncUser($webSocketDialogUser); + ZincSearchDialogMsg::userSync($webSocketDialogUser); } /** @@ -53,7 +53,7 @@ class WebSocketDialogUserObserver public function deleted(WebSocketDialogUser $webSocketDialogUser) { Deleted::record('dialog', $webSocketDialogUser->dialog_id, $webSocketDialogUser->userid); - ZincSearchDialogUserMsg::deleteUser($webSocketDialogUser); + ZincSearchDialogMsg::delete($webSocketDialogUser); } /**