feat(t17): 单聊同步+补拉对账+同步消息进分发;修复 get_roam_msg 60008

新增单聊(C2C)同步、补拉对账定时任务,同步消息纳入分发队列推送业务系统。
1.单聊同步④:user遍历→recentcontact/get_list收集C2C会话→会话对去重→
  get_roam_msg时间窗向前滚动全量→落库(conv_type=1/SYNC/进分发)
2.补拉对账⑤:SyncCheckTask定时(1h)+手动按钮,重跑群+单聊幂等补缺失
3.同步消息进分发:群/单聊distStatus=0入DistQueue;限速分批入队
  (next_retry_at按序号错开delay=seq/dispatchQps)
4.重构:enqueue抽到DispatchService公共方法,Pull/Sync共用

修复(get_roam_msg一直60008静默失败,C2C补拉从未工作):
-命令字 openim_admin/get_roam_msg → openim/admin_getroammsg
-字段 From/To/MaxTimeInterval → Operator_Account/Peer_Account/MaxTime
-新增 getRoamMsgRangeAs(MinTime/MaxTime+LastMsgKey续拉)+getRecentContactListAs

联调(sa 16用户):单聊链路跑通(get_list32+admin_getroammsg OK/Complete=1),
幂等正常,补拉定时工作,无死循环。enqueue实证待配callbackUrl+新消息。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
连龙刚
2026-07-10 11:21:49 +08:00
co-authored by Claude
parent 968a5f2e42
commit 7ca5bd4290
9 changed files with 536 additions and 56 deletions
@@ -634,6 +634,32 @@ public class AdminController {
ctx.redirect(basePath + "/admin/sync?msg=smdone");
}
/** 同步单聊消息:需先同步用户(群成员) */
@Post
@Mapping("/sync/c2cmsg")
public void syncC2CMsg(@Param(required = false) Long sourceAppId,
@Param(defaultValue = "") String tenantId,
Context ctx) throws Throwable {
if (sourceAppId == null && (tenantId == null || tenantId.isEmpty())) {
ctx.redirect(basePath + "/admin/sync?msg=syncopt");
return;
}
syncService.syncC2CMessages(sourceAppId, tenantId);
ctx.redirect(basePath + "/admin/sync?msg=scdone");
}
/** 补拉对账:重跑群+单聊(幂等补缺失),仅主应用 + 指定租户 */
@Post
@Mapping("/sync/check")
public void syncCheck(@Param String tenantId, Context ctx) throws Throwable {
if (tenantId == null || tenantId.isEmpty()) {
ctx.redirect(basePath + "/admin/sync?msg=syncopt");
return;
}
syncService.checkAndPull(tenantId);
ctx.redirect(basePath + "/admin/sync?msg=chkdone");
}
// ==================== 公共:构造页面模型 ====================
/**
@@ -1,7 +1,9 @@
package com.imutil.service;
import com.imutil.entity.DistQueue;
import com.imutil.entity.ImMessage;
import java.time.OffsetDateTime;
import java.util.List;
/**
@@ -32,4 +34,24 @@ public interface DispatchService {
* @return 重置条数
*/
int recoverStuck();
/**
* 写入分发队列(默认 nextRetryAt=now,立即可被消费分发)
* <p>
* 取租户 callbackUrl 作为目标;租户未配置 callbackUrl 则跳过(不报错,本地留底仍有效)。
*
* @param tenantId 归属租户
* @param msg 已落库消息(payload 取其 msgKey/convType/convId/from/to/msgTime/msgType/msgBody/source
*/
void enqueue(String tenantId, ImMessage msg);
/**
* 写入分发队列并指定下次分发时间(限速平滑分发用)
* <p>
* 同步批量入队时按序号错开 nextRetryAtdelay = seq / dispatchQps 秒),
* DispatchWorker 按时间消费,避免历史消息瞬时打爆业务系统 callbackUrl。
*
* @param nextRetryAt 下次可分发时间
*/
void enqueue(String tenantId, ImMessage msg, OffsetDateTime nextRetryAt);
}
@@ -9,8 +9,9 @@ import com.imutil.entity.MigrateTask;
* 本服务定位为<b>只读拉取</b>:把指定腾讯 IM 应用里的群组/群成员/群消息拉到本地表,
* 供管理后台按租户查看,不在腾讯侧产生任何写操作。
* <p>
* 腾讯 IM 现实约束:无"全量用户列表"接口(用户靠群成员反推);无"全量 C2C 会话列表"接口
* (单聊历史消息本轮不做全量,回调增量已由 {@link com.imutil.service.PullService} 覆盖)
* 腾讯 IM 现实约束:无"全量用户列表"接口(用户靠群成员反推);C2C 单聊会话通过
* recentcontact/get_list 按"已同步用户"反查({@link #syncC2CMessages}),漫游消息用 get_roam_msg 全量翻页
* 同步进来的消息(群/单聊)均入分发队列(dist_status=0)推送业务系统,本地同时留底。
*
* @author imutil
*/
@@ -33,9 +34,27 @@ public interface SyncService {
MigrateTask syncGroupMembers(Long sourceAppId, String tenantId);
/**
* 同步群消息:遍历该租户已同步的群,逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC)
* 同步群消息:遍历该租户已同步的群,逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_message(source=SYNC,进分发)
*
* @see #syncGroups(Long, String) 先同步群组,群消息才有目标群清单
*/
MigrateTask syncGroupMessages(Long sourceAppId, String tenantId);
/**
* 同步单聊(C2C)消息:遍历该租户已同步用户 → recentcontact/get_list 收集 C2C 会话 →
* 会话对去重 → get_roam_msg 时间窗向前滚动全量 → 写 im_message(conv_type=1, source=SYNC, 进分发)。
* <p>
* 腾讯无全量 C2C 会话列表接口,靠"已同步用户"反查其最近会话;未导入用户返回 50001 跳过。
*
* @see #syncGroupMembers(Long, String) 先同步用户(群成员),单聊才有用户清单可遍历
*/
MigrateTask syncC2CMessages(Long sourceAppId, String tenantId);
/**
* 补拉对账:重跑群消息 + 单聊同步(幂等,本地已有的跳过、缺失的补入),确保本系统数据最全。
* 定时任务 {@link com.imutil.task.SyncCheckTask} 驱动,也可手动触发。
*
* @param tenantId 归属租户(补拉对账仅对主应用归属租户有意义)
*/
MigrateTask checkAndPull(String tenantId);
}
@@ -2,9 +2,13 @@ package com.imutil.service.impl;
import com.imutil.common.Httpx;
import com.imutil.entity.DistQueue;
import com.imutil.entity.ImMessage;
import com.imutil.entity.Tenant;
import com.imutil.mapper.DistQueueMapper;
import com.imutil.service.DispatchService;
import com.imutil.service.TenantService;
import lombok.extern.slf4j.Slf4j;
import org.noear.snack4.ONode;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.noear.solon.data.annotation.Tran;
@@ -29,6 +33,9 @@ public class DispatchServiceImpl implements DispatchService {
@Inject
private DistQueueMapper distQueueMapper;
@Inject
private TenantService tenantService;
@Inject("${imutil.dispatch.fetchBatch:50}")
private int fetchBatch;
@@ -91,4 +98,51 @@ public class DispatchServiceImpl implements DispatchService {
OffsetDateTime threshold = now.minusMinutes(lockTimeoutMin);
return distQueueMapper.recoverStuck(threshold, now);
}
@Override
public void enqueue(String tenantId, ImMessage msg) {
enqueue(tenantId, msg, OffsetDateTime.now());
}
/**
* 写入分发队列(复用回调分发链路,payload 为结构化 JSON
* <p>
* 取租户 callbackUrl 作为目标;未配置则跳过(不报错,本地留底仍有效)。
* nextRetryAt 由调用方控制:同步批量入队时按序号错开以限速平滑分发。
*/
@Override
public void enqueue(String tenantId, ImMessage msg, OffsetDateTime nextRetryAt) {
Tenant t = tenantService.getById(tenantId);
if (t == null || t.getCallbackUrl() == null || t.getCallbackUrl().isEmpty()) {
return;
}
ONode payload = ONode.ofJson("{}");
payload.set("source", msg.getSource());
payload.set("msgKey", msg.getMsgKey());
payload.set("tenantId", tenantId);
payload.set("convType", msg.getConvType());
payload.set("convId", msg.getConvId());
payload.set("fromAccount", msg.getFromAccount());
payload.set("toAccount", msg.getToAccount());
payload.set("groupId", msg.getGroupId());
payload.set("msgTime", msg.getMsgTime());
payload.set("msgType", msg.getMsgType());
if (msg.getMsgBody() != null) {
try {
payload.set("msgBody", ONode.ofJson(msg.getMsgBody()));
} catch (Exception e) {
payload.set("msgBody", msg.getMsgBody());
}
}
DistQueue q = new DistQueue();
q.setMsgKey(msg.getMsgKey());
q.setTenantId(tenantId);
q.setConvId(msg.getConvId());
q.setTargetUrl(t.getCallbackUrl());
q.setPayload(payload.toJson());
q.setStatus(0);
q.setRetryCount(0);
q.setNextRetryAt(nextRetryAt);
distQueueMapper.insert(q);
}
}
@@ -10,6 +10,7 @@ import com.imutil.entity.Tenant;
import com.imutil.mapper.DistQueueMapper;
import com.imutil.mapper.ImMessageMapper;
import com.imutil.mapper.PullWatermarkMapper;
import com.imutil.service.DispatchService;
import com.imutil.service.PullService;
import com.imutil.service.TenantService;
import com.imutil.tencent.TencentImClient;
@@ -56,6 +57,9 @@ public class PullServiceImpl implements PullService {
@Inject
private TenantService tenantService;
@Inject
private DispatchService dispatchService;
@Inject("${imutil.pull.convsPerRound:20}")
private int convsPerRound;
@@ -168,50 +172,12 @@ public class PullServiceImpl implements PullService {
imMessageMapper.insert(msg);
newCount++;
// 仅新插入的消息入队分发
enqueue(wm.getTenantId(), msg);
dispatchService.enqueue(wm.getTenantId(), msg);
}
advanceWatermark(wm, maxSeq, maxTime);
return newCount;
}
/**
* 写入分发队列(复用回调分发链路,payload 用结构化 JSON
*/
private void enqueue(String tenantId, ImMessage msg) {
Tenant t = tenantService.getById(tenantId);
if (t == null || t.getCallbackUrl() == null || t.getCallbackUrl().isEmpty()) {
return;
}
ONode payload = ONode.ofJson("{}");
payload.set("source", msg.getSource());
payload.set("msgKey", msg.getMsgKey());
payload.set("tenantId", tenantId);
payload.set("convType", msg.getConvType());
payload.set("convId", msg.getConvId());
payload.set("fromAccount", msg.getFromAccount());
payload.set("toAccount", msg.getToAccount());
payload.set("groupId", msg.getGroupId());
payload.set("msgTime", msg.getMsgTime());
payload.set("msgType", msg.getMsgType());
if (msg.getMsgBody() != null) {
try {
payload.set("msgBody", ONode.ofJson(msg.getMsgBody()));
} catch (Exception e) {
payload.set("msgBody", msg.getMsgBody());
}
}
DistQueue q = new DistQueue();
q.setMsgKey(msg.getMsgKey());
q.setTenantId(tenantId);
q.setConvId(msg.getConvId());
q.setTargetUrl(t.getCallbackUrl());
q.setPayload(payload.toJson());
q.setStatus(0);
q.setRetryCount(0);
q.setNextRetryAt(OffsetDateTime.now());
distQueueMapper.insert(q);
}
/**
* 解析群消息节点为 ImMessage
*/
@@ -15,6 +15,7 @@ import com.imutil.mapper.MigrateTaskMapper;
import com.imutil.mapper.SourceAppMapper;
import com.imutil.mapper.TenantMapper;
import com.imutil.mapper.UserMappingMapper;
import com.imutil.service.DispatchService;
import com.imutil.service.SyncService;
import com.imutil.tencent.TencentImClient;
import lombok.extern.slf4j.Slf4j;
@@ -25,7 +26,10 @@ import org.noear.solon.annotation.Inject;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* 数据同步服务实现(T17
@@ -89,6 +93,21 @@ public class SyncServiceImpl implements SyncService {
@Inject("${imutil.sync.maxGroups:500}")
private int maxGroups;
/** 同步入队分发限速(每秒入队条数,保护业务系统 callbackUrl;错开 next_retry_at 平滑分发) */
@Inject("${imutil.sync.dispatchQps:10}")
private int syncDispatchQps;
@Inject
private DispatchService dispatchService;
/** 单聊同步用户上限(防海量用户 N×M 爆配额;如需全量调大) */
@Inject("${imutil.sync.maxSyncUsers:200}")
private int maxSyncUsers;
/** 单聊漫游回溯天数(腾讯 C2C 漫游存储期,默认 7 天,套餐更长可调大) */
@Inject("${imutil.sync.c2cLookbackDays:7}")
private int c2cLookbackDays;
/** 数据源密钥解析结果 */
private static class SyncCtx {
long sdkAppId;
@@ -366,6 +385,7 @@ public class SyncServiceImpl implements SyncService {
migrateTaskMapper.updateById(task);
int imported = 0;
OffsetDateTime syncBase = OffsetDateTime.now();
for (int gi = 0; gi < groups.size(); gi++) {
String groupId = groups.get(gi).getImGroupId();
try {
@@ -399,6 +419,8 @@ public class SyncServiceImpl implements SyncService {
}
imMessageMapper.insert(msg);
imported++;
// 限速入队分发:按序号错开 next_retry_atDispatchWorker 按速率消费
dispatchService.enqueue(ctx.tenantId, msg, nextRetryAt(imported, syncBase));
}
boolean finished = root.get("IsFinished").getLong() == 1;
// 终止:已拉完 / 空页 / 本页无有效 seq
@@ -432,7 +454,7 @@ public class SyncServiceImpl implements SyncService {
}
/**
* 解析腾讯群消息节点为 ImMessagesource=SYNCdist_status=1 不进分发)
* 解析腾讯群消息节点为 ImMessagesource=SYNCdist_status=0 进分发)
*/
private ImMessage parseGroupMsgForSync(ONode m, String tenantId, String groupId) {
// 过滤占位/系统消息(IsPlaceMsg=1From_Account 与 MsgBody 均空,非真实消息,不入库)
@@ -456,10 +478,229 @@ public class SyncServiceImpl implements SyncService {
msg.setMsgBody(body.toString());
msg.setSource("SYNC");
msg.setIsCrossTenant(false);
msg.setDistStatus(1);
msg.setDistStatus(0); // 进分发
return msg;
}
// ==================== 同步单聊(C2C) ====================
/**
* 同步单聊消息:遍历已同步用户 → recentcontact/get_list 收集 C2C 会话 → 会话对去重 →
* get_roam_msg 时间窗向前滚动全量 → 落库(conv_type=1/source=SYNC/进分发)。
*/
@Override
public MigrateTask syncC2CMessages(Long sourceAppId, String tenantId) {
SyncCtx ctx = resolveCtx(sourceAppId, tenantId);
MigrateTask task = startTask(ctx, "C2C_SYNC");
try {
List<UserMapping> users = userMappingMapper.selectList(Wrappers.<UserMapping>lambdaQuery()
.eq(UserMapping::getTenantId, ctx.tenantId));
task.setTotalCount((long) users.size());
migrateTaskMapper.updateById(task);
Set<String> seenPairs = new HashSet<>();
int imported = 0;
OffsetDateTime syncBase = OffsetDateTime.now();
int userCnt = 0;
for (UserMapping u : users) {
if (++userCnt > maxSyncUsers) {
log.warn("单聊同步用户达上限 {},停止(调大 imutil.sync.maxSyncUsers 可继续)", maxSyncUsers);
break;
}
String self = u.getImUserId();
if (self == null || self.isEmpty()) {
continue;
}
try {
List<String> peers = collectC2CPeers(self, ctx);
for (String peer : peers) {
if (peer == null || peer.isEmpty() || peer.equals(self)) {
continue;
}
// 会话对去重:A↔B 只拉一次
if (!seenPairs.add(pairKey(self, peer))) {
continue;
}
imported += pullRoamMsgs(self, peer, ctx, syncBase, imported);
}
} catch (Exception ue) {
// 50001(用户未导入)等按用户级跳过
log.warn("单聊同步用户失败 {} : {}", self, ue.toString());
}
task.setProcessedCount((long) imported);
task.setPosCursor("user:" + userCnt);
migrateTaskMapper.updateById(task);
}
task.setStatus(3);
task.setFinishedAt(OffsetDateTime.now());
task.setVerifyResult(String.format("{\"users\":%d,\"pairs\":%d,\"imported\":%d}",
Math.min(userCnt, maxSyncUsers), seenPairs.size(), imported));
migrateTaskMapper.updateById(task);
log.info("单聊同步完成 tenant={} users={} pairs={} imported={}",
ctx.tenantId, Math.min(userCnt, maxSyncUsers), seenPairs.size(), imported);
return task;
} catch (Exception e) {
failTask(task, e);
log.error("单聊同步失败 tenant={}", ctx.tenantId, e);
throw e;
}
}
/**
* 分页拉取用户最近会话,收集 C2C(Type=1) 对端账号
*/
private List<String> collectC2CPeers(String self, SyncCtx ctx) {
List<String> peers = new ArrayList<>();
long timeStamp = 0;
long startIndex = 0;
int guard = 0;
while (guard++ < 1000) {
String resp = tencentImClient.getRecentContactListAs(self, timeStamp, startIndex,
ctx.sdkAppId, ctx.secretKey);
ONode root = ONode.ofJson(resp);
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
break;
}
ONode items = root.get("SessionItem");
int n = items.size();
for (int i = 0; i < n; i++) {
ONode item = items.get(i);
if (item.get("Type").getLong() == 1) { // 1=C2C
String to = item.get("To_Account").getString();
if (to != null && !to.isEmpty()) {
peers.add(to);
}
}
}
long complete = root.get("CompleteFlag").getLong();
if (complete == 1 || n == 0) {
break;
}
long ts = root.get("TimeStamp").getLong();
long si = root.get("StartIndex").getLong();
if (ts == timeStamp && si == startIndex) {
break; // 游标未推进防死循环
}
timeStamp = ts;
startIndex = si;
}
return peers;
}
/**
* 全量拉取两人漫游消息(向旧翻页),落库 + 限速入队分发,返回新增条数
*/
private int pullRoamMsgs(String self, String peer, SyncCtx ctx, OffsetDateTime syncBase, int seqBase) {
long minTime = Instant.now().getEpochSecond() - c2cLookbackDays * 86400L;
long maxTime = Instant.now().getEpochSecond();
String lastMsgKey = null;
int imported = 0;
int pages = 0;
while (pages++ < groupMsgMaxPages) {
String resp = tencentImClient.getRoamMsgRangeAs(self, peer, groupMsgPerPage, minTime, maxTime,
lastMsgKey, ctx.sdkAppId, ctx.secretKey);
ONode root = ONode.ofJson(resp);
if (!"OK".equalsIgnoreCase(root.get("ActionStatus").getString())) {
break;
}
ONode list = root.get("MsgList");
int n = list.size();
for (int i = 0; i < n; i++) {
ONode m = list.get(i);
ImMessage msg = parseC2CMsgForSync(m, ctx.tenantId, self, peer);
if (msg == null) {
continue;
}
long cnt = imMessageMapper.selectCount(Wrappers.<ImMessage>lambdaQuery()
.eq(ImMessage::getMsgKey, msg.getMsgKey()));
if (cnt > 0) {
continue;
}
imMessageMapper.insert(msg);
imported++;
dispatchService.enqueue(ctx.tenantId, msg, nextRetryAt(seqBase + imported, syncBase));
}
long complete = root.get("Complete").getLong();
long lastMsgTime = root.get("LastMsgTime").getLong();
String nextLastMsgKey = root.get("LastMsgKey").getString();
// 终止:已拉完 / 空页 / 无 LastMsgTime
if (complete == 1 || n == 0 || lastMsgTime <= 0) {
break;
}
// 游标未推进防死循环(LastMsgTime 未向旧推进且 LastMsgKey 未变)
if (lastMsgTime >= maxTime && (nextLastMsgKey == null || nextLastMsgKey.equals(lastMsgKey))) {
break;
}
maxTime = lastMsgTime;
lastMsgKey = nextLastMsgKey;
}
return imported;
}
/**
* 解析腾讯单聊消息节点为 ImMessagesource=SYNC,进分发)
*/
private ImMessage parseC2CMsgForSync(ONode m, String tenantId, String from, String to) {
long msgSeq = m.get("MsgSeq").getLong();
long msgRandom = m.get("MsgRandom").getLong();
long msgTs = m.get("MsgTimeStamp").getLong();
String fromAcc = m.get("From_Account").getString();
if (fromAcc == null || fromAcc.isEmpty()) {
fromAcc = from;
}
ONode body = m.get("MsgBody");
ImMessage msg = new ImMessage();
msg.setMsgKey(MsgKeys.build(fromAcc, to, msgSeq, msgRandom));
msg.setTenantId(tenantId);
msg.setMsgTime(toTime(msgTs));
msg.setConvType(1);
msg.setConvId(to);
msg.setFromAccount(fromAcc);
msg.setToAccount(to);
msg.setMsgType(firstBodyType(body));
msg.setMsgBody(body.toString());
msg.setSource("SYNC");
msg.setIsCrossTenant(false);
msg.setDistStatus(0); // 进分发
return msg;
}
/** 会话对去重键:min#max,使 A↔B 只拉一次 */
private String pairKey(String a, String b) {
return (a.compareTo(b) <= 0) ? a + "#" + b : b + "#" + a;
}
// ==================== 补拉对账 ====================
/**
* 补拉对账:顺序重跑群消息 + 单聊同步(幂等,本地已有的跳过、缺失的补入)。
* 供定时任务 SyncCheckTask 与后台手动按钮调用,确保本系统数据最全。
*/
@Override
public MigrateTask checkAndPull(String tenantId) {
SyncCtx ctx = resolveCtx(null, tenantId);
MigrateTask task = startTask(ctx, "CHECK_PULL");
try {
task.setPosCursor("start");
migrateTaskMapper.updateById(task);
log.info("补拉对账开始 tenant={}", tenantId);
MigrateTask g = syncGroupMessages(null, tenantId);
MigrateTask c = syncC2CMessages(null, tenantId);
long gImp = g.getProcessedCount() == null ? 0 : g.getProcessedCount();
long cImp = c.getProcessedCount() == null ? 0 : c.getProcessedCount();
task.setStatus(3);
task.setFinishedAt(OffsetDateTime.now());
task.setVerifyResult(String.format("{\"groupImported\":%d,\"c2cImported\":%d}", gImp, cImp));
migrateTaskMapper.updateById(task);
log.info("补拉对账完成 tenant={} groupImported={} c2cImported={}", tenantId, gImp, cImp);
return task;
} catch (Exception e) {
failTask(task, e);
log.error("补拉对账失败 tenant={}", tenantId, e);
throw e;
}
}
// ==================== 工具方法 ====================
private void failTask(MigrateTask task, Exception e) {
@@ -491,4 +732,16 @@ public class SyncServiceImpl implements SyncService {
}
return OffsetDateTime.ofInstant(Instant.ofEpochSecond(epochSecond), ZoneId.systemDefault());
}
/**
* 计算限速入队的 next_retry_at:按序号错开(delay = seq / dispatchQps 秒),
* 使同步进来的历史消息按速率平滑分发,避免瞬时打爆业务系统 callbackUrl。
* dispatchQps&lt;=0 时不限速,立即 next_retry_at=now。
*/
private OffsetDateTime nextRetryAt(long seq, OffsetDateTime base) {
if (syncDispatchQps <= 0) {
return OffsetDateTime.now();
}
return base.plusSeconds(seq / syncDispatchQps);
}
}
@@ -0,0 +1,59 @@
package com.imutil.task;
import com.imutil.entity.Tenant;
import com.imutil.mapper.TenantMapper;
import com.imutil.service.SyncService;
import lombok.extern.slf4j.Slf4j;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.noear.solon.scheduling.annotation.Scheduled;
import java.util.List;
/**
* 数据同步补拉对账定时任务
* <p>
* 定期重跑群消息 + 单聊同步(幂等,本地已有的跳过、缺失的补入),
* 兜底腾讯侧有而本系统暂无的消息,确保本系统数据最全、业务系统不漏收。
* 对应 app.yml 的 solon.scheduling.job.syncCheckJob。
* <p>
* 遍历所有租户,对每个用主应用密钥执行 checkAndPull(主应用数据为主)。
*
* @author imutil
*/
@Slf4j
@Component
public class SyncCheckTask {
@Inject
private SyncService syncService;
@Inject
private TenantMapper tenantMapper;
/**
* 由 app.yml syncCheckJob 驱动(默认 fixedDelay=1 小时)
*/
@Scheduled(name = "syncCheckJob")
public void run() {
List<Tenant> tenants = tenantMapper.selectList(null);
if (tenants == null || tenants.isEmpty()) {
return;
}
int ok = 0;
for (Tenant t : tenants) {
String tid = t.getTenantId();
if (tid == null || tid.isEmpty()) {
continue;
}
try {
syncService.checkAndPull(tid);
ok++;
} catch (Throwable e) {
// 单租户异常不影响其他租户
log.error("补拉对账定时任务异常 tenant={}", tid, e);
}
}
log.info("补拉对账定时任务完成 租户数={} 成功={}", tenants.size(), ok);
}
}
@@ -256,7 +256,7 @@ public class TencentImClient {
/**
* 查询单聊历史漫游消息(C2C)
* <p>
* 命令字 openim_admin/get_roam_msg,按时间窗返回指定两人之间最近的 MaxCnt 条消息。
* 命令字 openim/admin_getroammsg,按时间窗返回指定两人之间最近的 MaxCnt 条消息。
* 补拉服务据此增量补全回调丢失的单聊消息。
*
* @param fromAccount 发送方 IM 账号
@@ -268,12 +268,13 @@ public class TencentImClient {
*/
public String getRoamMsg(String fromAccount, String toAccount, int maxCnt, long minTime, long maxInterval) {
Map<String, Object> body = new HashMap<>();
body.put("From_Account", fromAccount);
body.put("To_Account", toAccount);
// 腾讯 get_roam_msg 请求字段为 Operator_Account/Peer_Account + MinTime/MaxTime(无 MaxTimeInterval
body.put("Operator_Account", fromAccount);
body.put("Peer_Account", toAccount);
body.put("MaxCnt", maxCnt);
body.put("MinTime", minTime);
body.put("MaxTimeInterval", maxInterval);
return callApi("openim_admin/get_roam_msg", Jsons.stringify(body));
body.put("MaxTime", minTime + maxInterval);
return callApi("openim/admin_getroammsg", Jsons.stringify(body));
}
/**
@@ -301,12 +302,12 @@ public class TencentImClient {
public String getRoamMsgAs(String fromAccount, String toAccount, int maxCnt, long minTime, long maxInterval,
long srcSdkAppId, String srcSecretKey) {
Map<String, Object> body = new HashMap<>();
body.put("From_Account", fromAccount);
body.put("To_Account", toAccount);
body.put("Operator_Account", fromAccount);
body.put("Peer_Account", toAccount);
body.put("MaxCnt", maxCnt);
body.put("MinTime", minTime);
body.put("MaxTimeInterval", maxInterval);
return callApiAs("openim_admin/get_roam_msg", Jsons.stringify(body), srcSdkAppId, srcSecretKey);
body.put("MaxTime", minTime + maxInterval);
return callApiAs("openim/admin_getroammsg", Jsons.stringify(body), srcSdkAppId, srcSecretKey);
}
/**
@@ -325,6 +326,58 @@ public class TencentImClient {
return callApiAs("group_open_http_svc/group_msg_get_simple", Jsons.stringify(body), srcSdkAppId, srcSecretKey);
}
/**
* 拉取用户最近会话列表(含 C2C 单聊会话)— 单聊同步用
* <p>
* 命令字 recentcontact/get_list,分页返回 From_Account 的会话 SessionItem
* Type=1 为 C2C(带 To_Account),Type=2 为群(带 GroupId)。
* 单聊同步据此发现用户的 C2C 会话对,再配 getRoamMsgRangeAs 全量拉取漫游消息。
*
* @param fromAccount 本端账号
* @param timeStamp 分页时间戳(首轮 0;后续取上轮返回的 TimeStamp
* @param startIndex 分页起始索引(首轮 0;后续取上轮返回的 StartIndex
* @return 腾讯响应原始 JSON(含 SessionItem[]/CompleteFlag/TimeStamp/StartIndex
*/
public String getRecentContactListAs(String fromAccount, long timeStamp, long startIndex,
long srcSdkAppId, String srcSecretKey) {
Map<String, Object> body = new HashMap<>();
body.put("From_Account", fromAccount);
body.put("TimeStamp", timeStamp);
body.put("StartIndex", startIndex);
body.put("TopTimeStamp", 0);
body.put("AssistFlags", 15);
return callApiAs("recentcontact/get_list", Jsons.stringify(body), srcSdkAppId, srcSecretKey);
}
/**
* 查询单聊漫游消息(按时间区间,全量续拉用)— 单聊同步用
* <p>
* 腾讯 openim/admin_getroammsg 请求字段为 Operator_Account/Peer_Account + MinTime/MaxTime
* 续拉:MaxTime 取上轮返回 LastMsgTime,并带 LastMsgKey,直到 Complete=1。
*
* @param operatorAccount 会话一方(本端)
* @param peerAccount 会话另一方
* @param maxCnt 单次条数上限
* @param minTime 区间下界(秒级 epoch)
* @param maxTime 区间上界(秒级 epoch;首轮 now,续拉取上轮 LastMsgTime
* @param lastMsgKey 上轮返回 LastMsgKey(首轮 null
* @return 腾讯响应原始 JSON(含 MsgList/Complete/LastMsgTime/LastMsgKey
*/
public String getRoamMsgRangeAs(String operatorAccount, String peerAccount, int maxCnt,
long minTime, long maxTime, String lastMsgKey,
long srcSdkAppId, String srcSecretKey) {
Map<String, Object> body = new HashMap<>();
body.put("Operator_Account", operatorAccount);
body.put("Peer_Account", peerAccount);
body.put("MaxCnt", maxCnt);
body.put("MinTime", minTime);
body.put("MaxTime", maxTime);
if (lastMsgKey != null && !lastMsgKey.isEmpty()) {
body.put("LastMsgKey", lastMsgKey);
}
return callApiAs("openim/admin_getroammsg", Jsons.stringify(body), srcSdkAppId, srcSecretKey);
}
// ==================== 群迁移相关(T14 S3 ====================
/**
+31 -3
View File
@@ -5,6 +5,8 @@
<#if msg=='sgdone'>✅ 群组同步完成(详见任务记录)
<#elseif msg=='sudone'>✅ 群成员/用户同步完成(详见任务记录)
<#elseif msg=='smdone'>✅ 群消息同步完成(详见任务记录)
<#elseif msg=='scdone'>✅ 单聊消息同步完成(详见任务记录)
<#elseif msg=='chkdone'>✅ 补拉对账完成(详见任务记录)
<#elseif msg=='syncopt'>❌ 请选择数据源(主应用时须同时选归属租户)
<#else>${msg}
</#if>
@@ -14,8 +16,8 @@
<div class="stat-card" style="margin-bottom:14px;padding:12px 18px">
<b>说明</b>
<span class="muted">:本页为<b>只读拉取</b>,把腾讯 IM 数据同步到本地表用于查看,不在腾讯侧建群/建号。
<b>群组</b>=get_appid_group_list 全量分页;<b>群成员(用户)</b>=遍历群取成员(腾讯无全量用户接口,靠群成员反推);<b>群消息</b>=逐群全量拉取。
<b>C2C 单聊消息无全量会话列表接口</b>,历史单聊请用「数据迁移」页或依赖回调增量。</span>
<b>群组</b>=get_appid_group_list 全量分页;<b>群成员(用户)</b>=遍历群取成员(腾讯无全量用户接口,靠群成员反推);<b>群消息</b>=逐群全量拉取<b>单聊</b>=recentcontact/get_list 按已同步用户反查会话 + get_roam_msg 全量翻页
<b>所有同步消息均入分发队列推送业务系统</b>(限速分批入队),本地同时留底。</span>
</div>
<#-- 公共:数据源 + 归属租户选择(三个表单各自携带) -->
@@ -60,13 +62,39 @@
<div class="stat-card" style="margin-bottom:20px">
<div class="modal-title">③ 同步群消息</div>
<p class="muted">遍历该租户已同步的群,逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_messagesource=SYNC进分发队列)。<b>请先执行「同步群组」</b>。</p>
<p class="muted">遍历该租户已同步的群,逐群 getGroupMsg + IsFinished 滚动全量 → 写 im_messagesource=SYNC<b>进分发队列</b>,限速分批)。<b>请先执行「同步群组」</b>。</p>
<form method="post" action="${basePath}/admin/sync/groupmsg" style="margin-top:14px">
<div class="modal-body"><@srcSelect/></div>
<button type="submit" class="btn btn-primary">同步群消息</button>
</form>
</div>
<div class="stat-card" style="margin-bottom:20px">
<div class="modal-title">④ 同步单聊消息</div>
<p class="muted">遍历该租户已同步用户 → recentcontact/get_list 收集 C2C 会话 → 会话对去重 → get_roam_msg 全量翻页 → 写 im_messageconv_type=1source=SYNC,进分发)。<b>请先执行「同步群成员」</b>。</p>
<form method="post" action="${basePath}/admin/sync/c2cmsg" style="margin-top:14px">
<div class="modal-body"><@srcSelect/></div>
<button type="submit" class="btn btn-primary">同步单聊消息</button>
</form>
</div>
<div class="stat-card" style="margin-bottom:20px">
<div class="modal-title">⑤ 补拉对账</div>
<p class="muted">重跑群消息 + 单聊同步(幂等:本地已有的跳过、缺失的补入),确保本系统数据最全。仅主应用 + 指定租户,可定时自动或手动触发。</p>
<form method="post" action="${basePath}/admin/sync/check" style="margin-top:14px">
<div class="form-group">
<label>归属租户 <span class="muted">(主应用,对账该租户)</span></label>
<select name="tenantId">
<option value="">请选择租户</option>
<#list tenants as t>
<option value="${t.tenantId!}">${t.tenantName!}${t.tenantId!}</option>
</#list>
</select>
</div>
<button type="submit" class="btn btn-primary">补拉对账</button>
</form>
</div>
<h3 style="margin:20px 0 10px;font-size:15px;color:#2c3e50">同步/迁移任务记录</h3>
<#if tasks?has_content>
<table class="data-table">