From 968a5f2e422e94a4506e6cacd07ed70ce32d529f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=BF=9E=E9=BE=99=E5=88=9A?= Date: Fri, 10 Jul 2026 09:55:48 +0800 Subject: [PATCH] =?UTF-8?q?fix(t17):=20=E6=95=B0=E6=8D=AE=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E8=81=94=E8=B0=83=E4=BF=AE=E5=A4=8D=E2=80=94=E2=80=94?= =?UTF-8?q?=E7=BE=A4=E6=B6=88=E6=81=AF=E6=AD=BB=E5=BE=AA=E7=8E=AF=20+=20?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=90=8D=20+=20schema=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E5=8D=87=E7=BA=A7=20+=20maxGroups=E7=B2=BE=E7=A1=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 联调(新疆信创测试应用 1600134460,租户sa 67群)发现并修复4类问题: 1.群消息死循环:group_msg_get_simple 的 IsFinished=2 表"还有更旧可拉",占位消息群恒为2; 原 oldestSeq=max(seq) 游标方向反,反复拉同一批致2217次/200s超时。 改 min(seq) 向旧翻页 + 游标未推进即停兜底 + 过滤 IsPlaceMsg=1 占位消息(119次/15.7s) 2.get_group_info 请求体字段名 GroupId→GroupIdList(腾讯要求 GroupIdList:["id"] 字符串数组) 3.运行库缺列致 upserted=0:init.sql 仅新建库执行,新增 SchemaUpgradeBootstrap 启动幂等ALTER(6条DDL) 4.maxGroups 防护:break 由 for 外移入 for 内,精确限制避免单页海量群绕过上限 联调验证(租户sa 67群):群组67/67、群成员241/13、群消息119次/imported=3,死循环消除 Co-Authored-By: Claude --- .../imutil/service/impl/SyncServiceImpl.java | 40 +++++++++++++--- .../imutil/task/SchemaUpgradeBootstrap.java | 47 +++++++++++++++++++ .../com/imutil/tencent/TencentImClient.java | 3 +- 3 files changed, 82 insertions(+), 8 deletions(-) create mode 100644 src/main/java/com/imutil/task/SchemaUpgradeBootstrap.java diff --git a/src/main/java/com/imutil/service/impl/SyncServiceImpl.java b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java index 82d2f11..1797a9a 100644 --- a/src/main/java/com/imutil/service/impl/SyncServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/SyncServiceImpl.java @@ -85,6 +85,10 @@ public class SyncServiceImpl implements SyncService { @Inject("${imutil.sync.groupMsgMaxPages:500}") private int groupMsgMaxPages; + /** 单次同步群组数量上限（同步阻塞 HTTP，避免海量群卡死；如需全量调大） */ + @Inject("${imutil.sync.maxGroups:500}") + private int maxGroups; + /** 数据源密钥解析结果 */ private static class SyncCtx { long sdkAppId; @@ -158,7 +162,9 @@ public class SyncServiceImpl implements SyncService { ONode groupIds = root.get("GroupIdList"); int n = groupIds.size(); for (int i = 0; i < n; i++) { - String groupId = groupIds.get(i).getString(); + // 实测 GroupIdList 元素为对象 {GroupId:"..."}，兼容字符串形态 + ONode gidNode = groupIds.get(i); + String groupId = gidNode.isObject() ? gidNode.get("GroupId").getString() : gidNode.getString(); if (groupId == null || groupId.isEmpty()) { continue; } @@ -168,7 +174,12 @@ public class SyncServiceImpl implements SyncService { upserted++; } } catch (Exception ge) { - log.warn("同步单群失败 {} : {}", groupId, ge.getMessage()); + log.warn("同步单群失败 {} : {}", groupId, ge.toString()); + } + // 达上限停止（for 内精确限制，避免单页海量群卡死同步 HTTP） + if (total >= maxGroups) { + log.warn("群组同步达上限 {}，停止（调大 imutil.sync.maxGroups 可继续）", maxGroups); + break; } } task.setTotalCount((long) total); @@ -176,10 +187,16 @@ public class SyncServiceImpl implements SyncService { task.setPosCursor("page:" + guard); migrateTaskMapper.updateById(task); - next = root.get("Next").getString(); - if (next == null || next.isEmpty()) { + // 达上限停止（由 for 内触发并告警，此处仅跳出 while） + if (total >= maxGroups) { break; } + // Next 游标推进；为空或未变化则结束（防死循环） + String respNext = root.get("Next").getString(); + if (respNext == null || respNext.isEmpty() || respNext.equals(next)) { + break; + } + next = respNext; } task.setStatus(3); task.setFinishedAt(OffsetDateTime.now()); @@ -363,11 +380,11 @@ public class SyncServiceImpl implements SyncService { } ONode rspList = root.get("RspMsgList"); int n = rspList.size(); - long oldestSeq = 0; + long oldestSeq = Long.MAX_VALUE; // 本页最小 seq（最旧） for (int i = 0; i < n; i++) { ONode m = rspList.get(i); long seq = m.get("MsgSeq").getLong(); - if (seq > oldestSeq) { + if (seq > 0 && seq < oldestSeq) { oldestSeq = seq; } ImMessage msg = parseGroupMsgForSync(m, ctx.tenantId, groupId); @@ -384,7 +401,12 @@ public class SyncServiceImpl implements SyncService { imported++; } boolean finished = root.get("IsFinished").getLong() == 1; - if (finished || n == 0 || oldestSeq <= 0) { + // 终止：已拉完 / 空页 / 本页无有效 seq + if (finished || n == 0 || oldestSeq == Long.MAX_VALUE) { + break; + } + // 游标未向更旧方向推进则停（占位消息群 IsFinished 恒为 2，靠此兜底防死循环） + if (reqSeq > 0 && oldestSeq >= reqSeq) { break; } reqSeq = oldestSeq; @@ -413,6 +435,10 @@ public class SyncServiceImpl implements SyncService { * 解析腾讯群消息节点为 ImMessage（source=SYNC，dist_status=1 不进分发） */ private ImMessage parseGroupMsgForSync(ONode m, String tenantId, String groupId) { + // 过滤占位/系统消息（IsPlaceMsg=1：From_Account 与 MsgBody 均空，非真实消息，不入库） + if (m.get("IsPlaceMsg").getLong() == 1) { + return null; + } long msgSeq = m.get("MsgSeq").getLong(); long msgRandom = m.get("MsgRandom").getLong(); long msgTs = m.get("MsgTimeStamp").getLong(); diff --git a/src/main/java/com/imutil/task/SchemaUpgradeBootstrap.java b/src/main/java/com/imutil/task/SchemaUpgradeBootstrap.java new file mode 100644 index 0000000..048ddfc --- /dev/null +++ b/src/main/java/com/imutil/task/SchemaUpgradeBootstrap.java @@ -0,0 +1,47 @@ +package com.imutil.task; + +import com.imutil.mapper.PartitionMapper; +import lombok.extern.slf4j.Slf4j; +import org.noear.solon.annotation.Component; +import org.noear.solon.annotation.Init; +import org.noear.solon.annotation.Inject; + +/** + * 数据库 Schema 启动自动升级（幂等） + *

+ * 解决"实体加字段但运行库未 ALTER"问题：init.sql 仅新建库时执行，老库不会自动加列， + * 导致 MyBatis insert/update 到新列时报 column 不存在。启动时对增量列执行 + * ADD COLUMN IF NOT EXISTS，老库自动补列，新库重复执行也无副作用。 + * + * @author imutil + */ +@Slf4j +@Component +public class SchemaUpgradeBootstrap { + + @Inject + private PartitionMapper partitionMapper; + + @Init + public void init() { + // T17: group_mapping/user_mapping 同步资料列 + String[] ddls = { + "ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS name varchar(256)", + "ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS owner_account varchar(128)", + "ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS member_count int", + "ALTER TABLE group_mapping ADD COLUMN IF NOT EXISTS last_synced_at timestamptz", + "ALTER TABLE user_mapping ADD COLUMN IF NOT EXISTS nick varchar(128)", + "ALTER TABLE user_mapping ADD COLUMN IF NOT EXISTS last_synced_at timestamptz", + }; + int ok = 0; + for (String sql : ddls) { + try { + partitionMapper.execute(sql); + ok++; + } catch (Exception e) { + log.warn("schema升级跳过 [{}] : {}", sql, e.getMessage()); + } + } + log.info("启动 schema 升级完成（{}/{} 条 DDL）", ok, ddls.length); + } +} diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index d389bcd..1cd11e0 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -379,8 +379,9 @@ public class TencentImClient { * @return 腾讯响应原始 JSON，由调用方解析 */ public String getGroupInfo(String groupId, long srcSdkAppId, String srcSecretKey) { + // 腾讯 get_group_info 要求 GroupIdList: ["groupId"]（字符串数组，元素须为 string） Map body = new HashMap<>(); - body.put("GroupId", List.of(groupId)); + body.put("GroupIdList", List.of(groupId)); return callApiAs("group_open_http_svc/get_group_info", Jsons.stringify(body), srcSdkAppId, srcSecretKey); }