package carrierqueue import "github.com/redis/go-redis/v9" // 卡到期 Hash 字段前缀(见 constants.RedisCardDueKey)。 const ( fieldDue = "due:" // 各检查类型下次到期(毫秒) fieldOK = "ok:" // 最近一次成功(毫秒),用于冷却判断 fieldFail = "fail:" // 连续失败次数 fieldFirstFail = "ffail:" // 首次失败(毫秒) fieldError = "err:" // 最近错误(类别|信息) fieldConfirm = "cfm:" // 待执行的延迟确认序列(秒,逗号分隔) fieldExpected = "exp:" // 延迟确认的期望结果(如 offline / online),满足时取消后续确认 fieldPriority = "prio" // 优先标记:1 表示应在优先队列(执行中收到请求检查时由收尾移入优先队列) fieldCarrier = "carrier" ) // claimLua 账户领取(分发器抢到账户分发锁后执行)。 // // KEYS[1] 优先队列;KEYS[2] 普通队列;KEYS[3] 执行中计数。 // ARGV:now_ms、lease_ms、执行上限、优先占比(0 表示不限)、token 前缀、 // 限流组本秒剩余("g=n;g=n",空串表示不按 QPS 限制)、检查类型所属限流组("type=g;...")、是否计入实名(1/0)。 // 返回 {本轮清理的过期执行数, card_id, token, 是否来自优先(1/0), ...}。 const claimLua = ` local now = tonumber(ARGV[1]) local lease = tonumber(ARGV[2]) local limit = tonumber(ARGV[3]) local ratio = tonumber(ARGV[4]) local prefix = ARGV[5] local unlimited = ARGV[6] == '' local countRealname = ARGV[8] == '1' local expired = redis.call('ZREMRANGEBYSCORE', KEYS[3], '-inf', now) local result = {expired} local capacity = limit - redis.call('ZCARD', KEYS[3]) if capacity <= 0 then return result end local budget = {} for g, n in string.gmatch(ARGV[6], '([^=;]+)=(%-?%d+)') do budget[g] = tonumber(n) end local typeGroup = {} for t, g in string.gmatch(ARGV[7], '([^=;]+)=([^;]+)') do typeGroup[t] = g end local gatewayTypes = {'card_status', 'carddata', 'realname'} local used = {} local taken = 0 local stop = false local function tryClaim(cardId, fromPriority, qkey) local dueKey = 'polling:card:' .. cardId .. ':due' local need = {} if not unlimited then for _, t in ipairs(gatewayTypes) do if t ~= 'realname' or countRealname then local d = tonumber(redis.call('HGET', dueKey, 'due:' .. t) or '-1') if d >= 0 and d <= now then local g = typeGroup[t] if g then need[g] = (need[g] or 0) + 1 end end end end for g, n in pairs(need) do -- 本轮该组尚未占用且预算大于 0 时至少放行一张卡:否则一张卡的调用数大于预算(如账户 1 QPS、卡同时到期 3 项)时永远领不到; -- 超出部分由网关限流等待模式在调用时排队。 local u = used[g] or 0 if budget[g] ~= nil and (budget[g] <= 0 or (u > 0 and u + n > budget[g])) then stop = true return false end end end local claimKey = 'polling:card:' .. cardId .. ':claim' local token = prefix .. ':' .. cardId local ok = redis.call('SET', claimKey, token, 'NX', 'PX', lease) if not ok then local ttl = redis.call('PTTL', claimKey) if ttl < 0 then ttl = lease end redis.call('ZADD', qkey, 'XX', now + ttl, cardId) return false end for g, n in pairs(need) do used[g] = (used[g] or 0) + n end redis.call('ZADD', qkey, 'XX', now + lease, cardId) redis.call('ZADD', KEYS[3], now + lease, token) if fromPriority then redis.call('HDEL', dueKey, 'prio') end taken = taken + 1 table.insert(result, cardId) table.insert(result, token) table.insert(result, fromPriority and '1' or '0') return true end local function drain(qkey, fromPriority, quota) if quota <= 0 or stop then return end local ids = redis.call('ZRANGEBYSCORE', qkey, '-inf', now, 'LIMIT', 0, quota * 2 + 10) for _, id in ipairs(ids) do if stop or quota <= 0 or taken >= capacity then return end if tryClaim(id, fromPriority, qkey) then quota = quota - 1 end end end local priorityQuota = capacity if ratio > 0 then priorityQuota = math.floor(capacity * ratio) if priorityQuota < 1 then priorityQuota = 1 end end drain(KEYS[1], true, priorityQuota) drain(KEYS[2], false, capacity - taken) drain(KEYS[1], true, capacity - taken) return result ` // renewLua 续期:token 与执行权一致时延长租约与执行权,并推后 score 与执行中计数到期时间。 // // KEYS[1] claim;KEYS[2] exec;KEYS[3] 执行中计数;KEYS[4] 优先队列;KEYS[5] 普通队列。 // ARGV:token、执行权值、lease_ms、exec_ttl_ms、now_ms、card_id。返回 1 成功,0 token 或执行权已不一致。 const renewLua = ` if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end if redis.call('GET', KEYS[2]) ~= ARGV[2] then return 0 end local lease = tonumber(ARGV[3]) local now = tonumber(ARGV[5]) redis.call('PEXPIRE', KEYS[1], lease) redis.call('PEXPIRE', KEYS[2], tonumber(ARGV[4])) redis.call('ZADD', KEYS[3], 'XX', now + lease, ARGV[1]) redis.call('ZADD', KEYS[4], 'XX', now + lease, ARGV[6]) redis.call('ZADD', KEYS[5], 'XX', now + lease, ARGV[6]) return 1 ` // acquireExecLua 抢执行权:claim 等于 token 时 SET exec NX PX。返回 1 抢到,0 claim 不一致,-1 已有执行者。 // // KEYS[1] claim;KEYS[2] exec。ARGV:token、执行权值、exec_ttl_ms。 const acquireExecLua = ` if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end if redis.call('SET', KEYS[2], ARGV[2], 'NX', 'PX', tonumber(ARGV[3])) then return 1 end return -1 ` // finishLua 收尾:写回业务结果;token 一致时删除处理中标记与执行权、移出执行中计数, // 按优先标记把卡放入优先或普通队列,score = 各检查类型最早到期。卡已不在任一队列(已移出轮询)时不重新加入。 // // KEYS[1] due;KEYS[2] claim;KEYS[3] exec;KEYS[4] 执行中计数;KEYS[5] 优先队列;KEYS[6] 普通队列。 // ARGV:token、card_id、now_ms、set 字段对数 n、字段1、值1、…、删除字段…。 // 返回 1 正常收尾,0 token 不一致(只写了业务结果)。 const finishLua = ` local n = tonumber(ARGV[4]) local idx = 5 for i = 1, n do redis.call('HSET', KEYS[1], ARGV[idx], ARGV[idx + 1]) idx = idx + 2 end while idx <= #ARGV do redis.call('HDEL', KEYS[1], ARGV[idx]) idx = idx + 1 end if redis.call('GET', KEYS[2]) ~= ARGV[1] then return 0 end redis.call('DEL', KEYS[2]) redis.call('DEL', KEYS[3]) redis.call('ZREM', KEYS[4], ARGV[1]) local cardId = ARGV[2] local inPriority = redis.call('ZSCORE', KEYS[5], cardId) local inQueue = redis.call('ZSCORE', KEYS[6], cardId) if not inPriority and not inQueue then return 1 end local fields = redis.call('HGETALL', KEYS[1]) local nextDue = nil for i = 1, #fields, 2 do if string.sub(fields[i], 1, 4) == 'due:' then local v = tonumber(fields[i + 1]) if v and (nextDue == nil or v < nextDue) then nextDue = v end end end if nextDue == nil then nextDue = tonumber(ARGV[3]) + 3600000 end if redis.call('HGET', KEYS[1], 'prio') == '1' then redis.call('ZREM', KEYS[6], cardId) redis.call('ZADD', KEYS[5], nextDue, cardId) else redis.call('ZREM', KEYS[5], cardId) redis.call('ZADD', KEYS[6], nextDue, cardId) end return 1 ` // requestCheckLua 请求检查:把指定检查类型设为立即到期并放入优先队列;卡执行中时只打优先标记,收尾后立即再执行。 // 只作用于仍在轮询中的卡(在优先或普通队列中)。冷却期内(最近成功距今未超过冷却)的类型不请求。 // // KEYS[1] due;KEYS[2] 优先队列;KEYS[3] 普通队列;KEYS[4] claim。 // ARGV:now_ms、card_id、cooldown_ms(0 不冷却)、确认延迟(秒,逗号分隔,可空)、期望结果(可空)、 // 是否放入优先队列(1/0;0 用于手动触发:只把到期设为现在并在普通队列中立即到期)、类型…。 // 返回本次实际请求的类型数;卡不在轮询中返回 -1。 const requestCheckLua = ` local now = tonumber(ARGV[1]) local cardId = ARGV[2] local cooldown = tonumber(ARGV[3]) local inPriority = redis.call('ZSCORE', KEYS[2], cardId) local inQueue = redis.call('ZSCORE', KEYS[3], cardId) if not inPriority and not inQueue then return -1 end local requested = 0 local toPriority = ARGV[6] == '1' for i = 7, #ARGV do local t = ARGV[i] local skip = false if cooldown > 0 then local ok = tonumber(redis.call('HGET', KEYS[1], 'ok:' .. t) or '0') if ok > 0 and now - ok < cooldown then skip = true end end if not skip then redis.call('HSET', KEYS[1], 'due:' .. t, now) if ARGV[4] ~= '' then redis.call('HSET', KEYS[1], 'cfm:' .. t, ARGV[4]) if ARGV[5] ~= '' then redis.call('HSET', KEYS[1], 'exp:' .. t, ARGV[5]) else redis.call('HDEL', KEYS[1], 'exp:' .. t) end end requested = requested + 1 end end if requested == 0 then return 0 end if not toPriority then if redis.call('EXISTS', KEYS[4]) == 1 then return requested end if inQueue then redis.call('ZADD', KEYS[3], now, cardId) else redis.call('ZADD', KEYS[2], now, cardId) end return requested end redis.call('HSET', KEYS[1], 'prio', '1') if redis.call('EXISTS', KEYS[4]) == 1 then return requested end redis.call('ZREM', KEYS[3], cardId) redis.call('ZADD', KEYS[2], now, cardId) return requested ` // removeLua 把卡移出所属账户队列(关闭轮询、删除卡等);执行中的卡收尾时因不在队列中而不会被重新加入。 // KEYS[1] 优先队列;KEYS[2] 普通队列;KEYS[3] due。ARGV:card_id。 const removeLua = ` redis.call('ZREM', KEYS[1], ARGV[1]) redis.call('ZREM', KEYS[2], ARGV[1]) redis.call('DEL', KEYS[3]) return 1 ` // releaseLua 发布失败时按 token 删除处理中标记、移出执行中计数,并把 score 恢复为当前时间。 // KEYS[1] claim;KEYS[2] 执行中计数;KEYS[3] 优先队列;KEYS[4] 普通队列。ARGV:token、card_id、now_ms。 const releaseLua = ` if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end redis.call('DEL', KEYS[1]) redis.call('ZREM', KEYS[2], ARGV[1]) redis.call('ZADD', KEYS[3], 'XX', tonumber(ARGV[3]), ARGV[2]) redis.call('ZADD', KEYS[4], 'XX', tonumber(ARGV[3]), ARGV[2]) return 1 ` var ( claimScript = redis.NewScript(claimLua) renewScript = redis.NewScript(renewLua) acquireExecScript = redis.NewScript(acquireExecLua) finishScript = redis.NewScript(finishLua) requestCheckScript = redis.NewScript(requestCheckLua) removeScript = redis.NewScript(removeLua) releaseScript = redis.NewScript(releaseLua) )