|
|
@@ -0,0 +1,343 @@
|
|
|
+package com.fs.company.service.easycall;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.fs.common.utils.StringUtils;
|
|
|
+import com.fs.company.domain.CompanySiptaskInfo;
|
|
|
+import com.fs.company.mapper.CompanySiptaskInfoMapper;
|
|
|
+import com.fs.company.vo.easycall.EasyCallCommonAddCallListParam;
|
|
|
+import com.fs.company.vo.easycall.EasyCallPhoneItemVO;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+
|
|
|
+import java.util.Date;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.UUID;
|
|
|
+import java.util.concurrent.ThreadLocalRandom;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 外呼线路限流重试队列:去重入队、抖动分散、原子出队、失败退避重入队。
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Component
|
|
|
+public class OutboundLimitRetrySupport {
|
|
|
+
|
|
|
+ /** 与 EasyCallServiceImpl 保持一致 */
|
|
|
+ public static final String OUTBOUND_LIMIT_REDIS_PREFIX = "outbound:limit:";
|
|
|
+
|
|
|
+ public static final String RETRY_ZSET_KEY = OUTBOUND_LIMIT_REDIS_PREFIX + "retry:zset";
|
|
|
+
|
|
|
+ public static final String RETRY_PAYLOAD_HASH_KEY = OUTBOUND_LIMIT_REDIS_PREFIX + "retry:payload";
|
|
|
+
|
|
|
+ private static final String RETRY_PROCESSING_PREFIX = OUTBOUND_LIMIT_REDIS_PREFIX + "retry:processing:";
|
|
|
+
|
|
|
+ private static final String RETRY_DISPATCHED_PREFIX = OUTBOUND_LIMIT_REDIS_PREFIX + "retry:dispatched:";
|
|
|
+
|
|
|
+ /** 已成功 append+startTask 的 callBackUuid 标记 TTL(防 pop 后崩溃导致重复外呼) */
|
|
|
+ private static final int DISPATCHED_TTL_HOURS = 24;
|
|
|
+
|
|
|
+ private static final String RETRY_RATE_PREFIX = OUTBOUND_LIMIT_REDIS_PREFIX + "retry:rate:";
|
|
|
+
|
|
|
+ /** 到期后随机延迟上限(毫秒),平滑窗口边界流量 */
|
|
|
+ private static final int RETRY_JITTER_MAX_MS = 30_000;
|
|
|
+
|
|
|
+ /** 孤儿 payload 单次最多恢复条数 */
|
|
|
+ private static final int MAX_ORPHAN_RECOVER_BATCH = 100;
|
|
|
+
|
|
|
+ /** 失败退避基础延迟(毫秒) */
|
|
|
+ private static final long RETRY_BACKOFF_MS = 60_000L;
|
|
|
+
|
|
|
+ /** 单网关每秒最多重试外呼次数(多实例共享) */
|
|
|
+ private static final int MAX_RETRY_PER_GATEWAY_PER_SEC = 30;
|
|
|
+
|
|
|
+ /** 处理锁 TTL,防止实例崩溃后长期占锁 */
|
|
|
+ private static final int PROCESSING_LOCK_SECONDS = 600;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private OutboundLimitRetryRedisHelper retryRedisHelper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private CompanySiptaskInfoMapper companySiptaskInfoMapper;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 限流后入队(member=callBackUuid,同 UUID 不重复入队,score 带抖动)。
|
|
|
+ */
|
|
|
+ public void enqueueRetry(Long companyId, Long gatewayId, EasyCallCommonAddCallListParam param, Date nextAvailableTime) {
|
|
|
+ if (param == null || nextAvailableTime == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ String callBackUuid = extractCallBackUuid(param);
|
|
|
+ String member = StringUtils.isNotBlank(callBackUuid) ? callBackUuid : UUID.randomUUID().toString();
|
|
|
+
|
|
|
+ JSONObject retryData = new JSONObject();
|
|
|
+ retryData.put("companyId", companyId);
|
|
|
+ retryData.put("gatewayId", gatewayId);
|
|
|
+ retryData.put("param", param);
|
|
|
+ retryData.put("callBackUuid", member);
|
|
|
+ retryData.put("nextAvailableTime", nextAvailableTime.getTime());
|
|
|
+ retryData.put("createTime", System.currentTimeMillis());
|
|
|
+ Long roboticId = resolveRoboticId(param);
|
|
|
+ if (roboticId != null) {
|
|
|
+ retryData.put("roboticId", roboticId);
|
|
|
+ }
|
|
|
+
|
|
|
+ String existingPayload = retryRedisHelper.hashGet(RETRY_PAYLOAD_HASH_KEY, member);
|
|
|
+ if (existingPayload != null) {
|
|
|
+ JSONObject oldData = JSON.parseObject(existingPayload);
|
|
|
+ if (oldData != null && oldData.containsKey("failCount")) {
|
|
|
+ retryData.put("failCount", oldData.getIntValue("failCount"));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ double score = computeScoreWithJitter(nextAvailableTime.getTime());
|
|
|
+
|
|
|
+ Double existingScore = retryRedisHelper.zScore(RETRY_ZSET_KEY, member);
|
|
|
+ if (existingScore != null && existingScore <= score) {
|
|
|
+ retryRedisHelper.hashPut(RETRY_PAYLOAD_HASH_KEY, member, retryData.toJSONString());
|
|
|
+ log.info("enqueueRetry: 已存在更早或相同调度,仅刷新 payload - member={}", member);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ retryRedisHelper.zSetAdd(RETRY_ZSET_KEY, member, score);
|
|
|
+ retryRedisHelper.hashPut(RETRY_PAYLOAD_HASH_KEY, member, retryData.toJSONString());
|
|
|
+ log.info("enqueueRetry: member={}, score={}, companyId={}, gatewayId={}", member, (long) score, companyId, gatewayId);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("enqueueRetry 异常 companyId={}, gatewayId={}", companyId, gatewayId, e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 原子弹出已到期的重试 member(ZRANGE+ZREM 脚本,防多实例重复消费)。
|
|
|
+ */
|
|
|
+ public List<String> popDueMembers(int batchSize) {
|
|
|
+ long now = System.currentTimeMillis();
|
|
|
+ return retryRedisHelper.popDueMembersByScore(RETRY_ZSET_KEY, now, batchSize);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 解析任务 payload;兼容旧版 member 为整段 JSON 的写法。
|
|
|
+ */
|
|
|
+ public JSONObject resolvePayload(String member) {
|
|
|
+ if (StringUtils.isBlank(member)) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ if (member.trim().startsWith("{")) {
|
|
|
+ return JSON.parseObject(member);
|
|
|
+ }
|
|
|
+ String raw = retryRedisHelper.hashGet(RETRY_PAYLOAD_HASH_KEY, member);
|
|
|
+ if (raw == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return JSON.parseObject(raw);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 尝试获取处理锁,避免极端情况下重复外呼。
|
|
|
+ */
|
|
|
+ public boolean tryAcquireProcessingLock(String callBackUuid) {
|
|
|
+ if (StringUtils.isBlank(callBackUuid)) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ String lockKey = RETRY_PROCESSING_PREFIX + callBackUuid;
|
|
|
+ return retryRedisHelper.setIfAbsent(lockKey, "1", PROCESSING_LOCK_SECONDS, TimeUnit.SECONDS);
|
|
|
+ }
|
|
|
+
|
|
|
+ public void releaseProcessingLock(String callBackUuid) {
|
|
|
+ if (StringUtils.isBlank(callBackUuid)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ retryRedisHelper.deleteKey(RETRY_PROCESSING_PREFIX + callBackUuid);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 是否已成功完成 append+startTask(防 pop 后进程崩溃再次重试导致重复外呼)。
|
|
|
+ */
|
|
|
+ public boolean isDispatched(String callBackUuid) {
|
|
|
+ if (StringUtils.isBlank(callBackUuid)) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ return retryRedisHelper.getString(RETRY_DISPATCHED_PREFIX + callBackUuid) != null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 标记 callBackUuid 已完成派发(append 成功且 startTask 已触发)。
|
|
|
+ */
|
|
|
+ public void markDispatched(String callBackUuid) {
|
|
|
+ if (StringUtils.isBlank(callBackUuid)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ retryRedisHelper.setString(RETRY_DISPATCHED_PREFIX + callBackUuid, "1", DISPATCHED_TTL_HOURS, TimeUnit.HOURS);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 恢复 ZSET 已弹出但 payload 仍滞留 Hash 的孤儿任务(进程崩溃场景,不重试丢单)。
|
|
|
+ *
|
|
|
+ * @return 本次恢复入队条数
|
|
|
+ */
|
|
|
+ public int recoverOrphanedPayloads() {
|
|
|
+ Map<String, String> all = retryRedisHelper.hashGetAll(RETRY_PAYLOAD_HASH_KEY);
|
|
|
+ if (all == null || all.isEmpty()) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ int recovered = 0;
|
|
|
+ for (Map.Entry<String, String> entry : all.entrySet()) {
|
|
|
+ if (recovered >= MAX_ORPHAN_RECOVER_BATCH) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ String member = entry.getKey();
|
|
|
+ if (retryRedisHelper.zScore(RETRY_ZSET_KEY, member) != null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ JSONObject data = JSON.parseObject(entry.getValue());
|
|
|
+ if (data == null) {
|
|
|
+ ackSuccess(member);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String callBackUuid = data.getString("callBackUuid");
|
|
|
+ EasyCallCommonAddCallListParam param = data.getObject("param", EasyCallCommonAddCallListParam.class);
|
|
|
+ if (StringUtils.isBlank(callBackUuid) && param != null) {
|
|
|
+ callBackUuid = extractCallBackUuid(param);
|
|
|
+ }
|
|
|
+ if (isDispatched(callBackUuid)) {
|
|
|
+ ackSuccess(member);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ Long companyId = data.getLong("companyId");
|
|
|
+ Long gatewayId = data.getLong("gatewayId");
|
|
|
+ if (companyId == null || param == null) {
|
|
|
+ ackSuccess(member);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ Date nextTime = new Date(System.currentTimeMillis() + RETRY_BACKOFF_MS);
|
|
|
+ enqueueRetry(companyId, gatewayId, param, nextTime);
|
|
|
+ recovered++;
|
|
|
+ log.warn("recoverOrphanedPayloads: 孤儿任务重新入队 member={}, callBackUuid={}", member, callBackUuid);
|
|
|
+ }
|
|
|
+ return recovered;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 按网关令牌桶限速,超限则延迟重入队。
|
|
|
+ *
|
|
|
+ * @return true 表示当前秒额度已用尽,调用方应跳过本次外呼
|
|
|
+ */
|
|
|
+ public boolean isGatewayRateLimited(Long companyId, Long gatewayId) {
|
|
|
+ if (companyId == null || gatewayId == null) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ String bucketKey = RETRY_RATE_PREFIX + companyId + ":" + gatewayId;
|
|
|
+ Long count = retryRedisHelper.incr(bucketKey, 1L);
|
|
|
+ if (count == null) {
|
|
|
+ count = 1L;
|
|
|
+ }
|
|
|
+ if (count == 1L) {
|
|
|
+ retryRedisHelper.expire(bucketKey, 1, TimeUnit.SECONDS);
|
|
|
+ }
|
|
|
+ return count > MAX_RETRY_PER_GATEWAY_PER_SEC;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 成功后清理 payload;member 已在 pop 时从 ZSET 移除。
|
|
|
+ */
|
|
|
+ public void ackSuccess(String member) {
|
|
|
+ if (StringUtils.isBlank(member) || member.trim().startsWith("{")) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ retryRedisHelper.hashDelete(RETRY_PAYLOAD_HASH_KEY, member);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 失败或异常:退避后重新入队,不丢单。
|
|
|
+ */
|
|
|
+ public void requeueWithBackoff(JSONObject retryData, String member) {
|
|
|
+ if (retryData == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ Long companyId = retryData.getLong("companyId");
|
|
|
+ Long gatewayId = retryData.getLong("gatewayId");
|
|
|
+ EasyCallCommonAddCallListParam param = retryData.getObject("param", EasyCallCommonAddCallListParam.class);
|
|
|
+ if (param == null) {
|
|
|
+ log.warn("requeueWithBackoff: param 为空,无法重入队 member={}", member);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ int failCount = retryData.getIntValue("failCount") + 1;
|
|
|
+ retryData.put("failCount", failCount);
|
|
|
+ if (StringUtils.isNotBlank(member) && !member.trim().startsWith("{")) {
|
|
|
+ retryRedisHelper.hashPut(RETRY_PAYLOAD_HASH_KEY, member, retryData.toJSONString());
|
|
|
+ }
|
|
|
+ long delay = RETRY_BACKOFF_MS * Math.min(failCount, 5);
|
|
|
+ Date nextTime = new Date(System.currentTimeMillis() + delay);
|
|
|
+ enqueueRetry(companyId, gatewayId, param, nextTime);
|
|
|
+ log.warn("requeueWithBackoff: member={}, failCount={}, nextDelayMs={}", member, failCount, delay);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static String extractCallBackUuid(EasyCallCommonAddCallListParam param) {
|
|
|
+ if (param == null || param.getPhoneList() == null || param.getPhoneList().isEmpty()) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ EasyCallPhoneItemVO item = param.getPhoneList().get(0);
|
|
|
+ if (item == null || item.getBizJson() == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ JSONObject bizJson;
|
|
|
+ if (item.getBizJson() instanceof JSONObject) {
|
|
|
+ bizJson = (JSONObject) item.getBizJson();
|
|
|
+ } else {
|
|
|
+ bizJson = JSON.parseObject(String.valueOf(item.getBizJson()));
|
|
|
+ }
|
|
|
+ return bizJson != null ? bizJson.getString("callBackUuid") : null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 任务暂停/删除时,取消该任务尚未派发的限流重试,避免继续 append 或 startTask 把 EasyCall 任务拉回运行。
|
|
|
+ */
|
|
|
+ public int cancelPendingRetriesForRobotic(Long roboticId) {
|
|
|
+ if (roboticId == null) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ Map<String, String> all = retryRedisHelper.hashGetAll(RETRY_PAYLOAD_HASH_KEY);
|
|
|
+ if (all == null || all.isEmpty()) {
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ int cancelled = 0;
|
|
|
+ for (Map.Entry<String, String> entry : all.entrySet()) {
|
|
|
+ JSONObject data = JSON.parseObject(entry.getValue());
|
|
|
+ if (data == null) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ Long rid = data.getLong("roboticId");
|
|
|
+ if (rid == null) {
|
|
|
+ EasyCallCommonAddCallListParam param = data.getObject("param", EasyCallCommonAddCallListParam.class);
|
|
|
+ rid = resolveRoboticId(param);
|
|
|
+ }
|
|
|
+ if (!roboticId.equals(rid)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ String member = entry.getKey();
|
|
|
+ retryRedisHelper.zSetRemove(RETRY_ZSET_KEY, member);
|
|
|
+ ackSuccess(member);
|
|
|
+ cancelled++;
|
|
|
+ }
|
|
|
+ if (cancelled > 0) {
|
|
|
+ log.info("cancelPendingRetriesForRobotic: roboticId={}, cancelled={}", roboticId, cancelled);
|
|
|
+ }
|
|
|
+ return cancelled;
|
|
|
+ }
|
|
|
+
|
|
|
+ public Long resolveRoboticId(EasyCallCommonAddCallListParam param) {
|
|
|
+ if (param == null || param.getBatchId() == null || companySiptaskInfoMapper == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ CompanySiptaskInfo sipTaskInfo = companySiptaskInfoMapper.selectSipTaskInfoByBatchId(param.getBatchId());
|
|
|
+ return sipTaskInfo != null ? sipTaskInfo.getTaskId() : null;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static double computeScoreWithJitter(long baseTimeMs) {
|
|
|
+ int jitter = ThreadLocalRandom.current().nextInt(RETRY_JITTER_MAX_MS + 1);
|
|
|
+ return baseTimeMs + jitter;
|
|
|
+ }
|
|
|
+}
|