a4e3fe4d8f
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m24s
- 开关开启但分发器心跳缺失时回退为当场执行，避免指令无人消费 - 待确认指令先查上游状态再判断；执行期间被覆盖时把待确认带到新指令 - 同方向失败退避中重复写入不提前执行时间 - 领取容量把执行上限与本秒剩余 QPS 分开计算 - 队列复机共用单卡复机锁；执行上下文取消后按结果未知确认 - 消费者并行停止；读取最近调用时间失败时不覆盖；成功但终结日志失败按成功处理 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
232 lines
10 KiB
Go
232 lines
10 KiB
Go
package carrierqueue
|
||
|
||
import "github.com/redis/go-redis/v9"
|
||
|
||
// 停复机指令 Hash 字段（见 constants.RedisCardStopResumeCmdKey）。
|
||
const (
|
||
srFieldAction = "action" // stop / start
|
||
srFieldReason = "reason" // 停机原因（复机为空）
|
||
srFieldScenes = "scenes" // 触发场景（逗号分隔）
|
||
srFieldSource = "source" // 请求来源
|
||
srFieldCarrier = "carrier" // 所属运营商账户
|
||
srFieldVersion = "ver" // 指令版本，每次写入 HINCRBY
|
||
srFieldFail = "fail" // 连续失败次数
|
||
srFieldFirst = "ffail" // 首次失败（毫秒）
|
||
srFieldError = "err" // 最近错误（类别|信息）
|
||
srFieldPending = "pending" // 1 表示结果未知、待确认
|
||
srFieldUnknown = "unknown" // 连续结果未知次数
|
||
srFieldAt = "at" // 最近写入时间（毫秒）
|
||
)
|
||
|
||
// srWriteLua 写入或覆盖停复机指令：方向相反时清零失败与待确认，递增版本，放入账户停复机队列。
|
||
// 指令执行中（存在处理中标记）时只更新指令内容与版本，由收尾按版本保留；待确认中、或同方向失败退避中时保持原执行时间。
|
||
//
|
||
// KEYS[1] 指令 Hash；KEYS[2] 账户停复机队列；KEYS[3] 处理中标记。
|
||
// ARGV：action、reason、scenes、source、carrier、now_ms、card_id。返回新版本号。
|
||
const srWriteLua = `
|
||
local prev = redis.call('HGET', KEYS[1], 'action')
|
||
if prev and prev ~= ARGV[1] then
|
||
redis.call('HDEL', KEYS[1], 'fail', 'ffail', 'err', 'pending', 'unknown')
|
||
end
|
||
redis.call('HSET', KEYS[1], 'action', ARGV[1], 'reason', ARGV[2], 'scenes', ARGV[3], 'source', ARGV[4], 'carrier', ARGV[5], 'at', ARGV[6])
|
||
local ver = redis.call('HINCRBY', KEYS[1], 'ver', 1)
|
||
if redis.call('EXISTS', KEYS[3]) == 1 then return ver end
|
||
local backingOff = prev == ARGV[1] and tonumber(redis.call('HGET', KEYS[1], 'fail') or '0') > 0
|
||
if redis.call('HGET', KEYS[1], 'pending') == '1' or backingOff then
|
||
-- 待确认或同方向失败退避中：不提前执行时间（ZADD NX 只补缺失）。
|
||
redis.call('ZADD', KEYS[2], 'NX', tonumber(ARGV[6]), ARGV[7])
|
||
else
|
||
redis.call('ZADD', KEYS[2], 'LT', tonumber(ARGV[6]), ARGV[7])
|
||
end
|
||
return ver
|
||
`
|
||
|
||
// srClaimLua 领取账户到期停复机指令：先清理过期执行中，按容量领取；未到单卡最小重试间隔的推后到可调用时间。
|
||
// 待确认的指令只查状态，不受最小间隔约束。指令已不存在的成员直接移出队列。
|
||
//
|
||
// KEYS[1] 账户停复机队列；KEYS[2] 账户停复机执行中计数。
|
||
// ARGV：now_ms、lease_ms、执行上限、token 前缀、单卡最小间隔 ms、本秒剩余 QPS（-1 表示不按 QPS 限制）。
|
||
// 返回 {过期数, card_id, token, ver, action, pending, ...}。
|
||
const srClaimLua = `
|
||
local now = tonumber(ARGV[1])
|
||
local lease = tonumber(ARGV[2])
|
||
local limit = tonumber(ARGV[3])
|
||
local prefix = ARGV[4]
|
||
local minInterval = tonumber(ARGV[5])
|
||
local budget = tonumber(ARGV[6])
|
||
local expired = redis.call('ZREMRANGEBYSCORE', KEYS[2], '-inf', now)
|
||
local result = {expired}
|
||
local capacity = limit - redis.call('ZCARD', KEYS[2])
|
||
if budget >= 0 and budget < capacity then capacity = budget end
|
||
if capacity <= 0 then return result end
|
||
local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, capacity * 2 + 10)
|
||
local taken = 0
|
||
for _, id in ipairs(ids) do
|
||
if taken >= capacity then break end
|
||
local cmdKey = 'polling:card:' .. id .. ':stop_resume_cmd'
|
||
local fields = redis.call('HMGET', cmdKey, 'action', 'ver', 'pending')
|
||
if not fields[1] then
|
||
redis.call('ZREM', KEYS[1], id)
|
||
else
|
||
local pending = fields[3] == '1'
|
||
local blockedUntil = 0
|
||
if not pending then
|
||
local last = tonumber(redis.call('GET', 'polling:card:' .. id .. ':sr_last_call') or '0')
|
||
if last > 0 and last + minInterval > now then blockedUntil = last + minInterval end
|
||
end
|
||
if blockedUntil > 0 then
|
||
redis.call('ZADD', KEYS[1], 'XX', blockedUntil, id)
|
||
else
|
||
local claimKey = 'polling:card:' .. id .. ':sr_claim'
|
||
local token = prefix .. ':' .. id
|
||
if redis.call('SET', claimKey, token, 'NX', 'PX', lease) then
|
||
redis.call('ZADD', KEYS[1], 'XX', now + lease, id)
|
||
redis.call('ZADD', KEYS[2], now + lease, token)
|
||
taken = taken + 1
|
||
table.insert(result, id)
|
||
table.insert(result, token)
|
||
table.insert(result, fields[2] or '0')
|
||
table.insert(result, fields[1])
|
||
table.insert(result, pending and '1' or '0')
|
||
else
|
||
local ttl = redis.call('PTTL', claimKey)
|
||
if ttl < 0 then ttl = lease end
|
||
redis.call('ZADD', KEYS[1], 'XX', now + ttl, id)
|
||
end
|
||
end
|
||
end
|
||
end
|
||
return result
|
||
`
|
||
|
||
// srFinishLua 停复机指令收尾：token 不一致返回 0 不做任何修改；版本已变大（执行期间写入了新指令）时保留新指令、立即可领取（返回 2）；
|
||
// 否则 done=1 删除指令并移出队列，done=0 写入字段并按 next_ms 重新排队（返回 1）。
|
||
//
|
||
// KEYS[1] 指令 Hash；KEYS[2] 处理中标记；KEYS[3] 执行权；KEYS[4] 执行中计数；KEYS[5] 账户停复机队列。
|
||
// ARGV：token、card_id、领取时版本、done(1/0)、next_ms、now_ms、本次结果未知(1/0)、set 字段对数 n、字段1、值1、…、删除字段…。
|
||
// 版本已变大且本次结果未知时，把待确认带到新指令上：新指令执行前先查状态，避免上游已生效而库里未知。
|
||
const srFinishLua = `
|
||
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]
|
||
if (redis.call('HGET', KEYS[1], 'ver') or '0') ~= ARGV[3] then
|
||
if redis.call('EXISTS', KEYS[1]) == 1 then
|
||
if ARGV[7] == '1' then redis.call('HSET', KEYS[1], 'pending', '1') end
|
||
redis.call('ZADD', KEYS[5], tonumber(ARGV[6]), cardId)
|
||
else
|
||
redis.call('ZREM', KEYS[5], cardId)
|
||
end
|
||
return 2
|
||
end
|
||
if ARGV[4] == '1' then
|
||
redis.call('DEL', KEYS[1])
|
||
redis.call('ZREM', KEYS[5], cardId)
|
||
return 1
|
||
end
|
||
local n = tonumber(ARGV[8])
|
||
local idx = 9
|
||
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
|
||
redis.call('ZADD', KEYS[5], tonumber(ARGV[5]), cardId)
|
||
return 1
|
||
`
|
||
|
||
// srReleaseLua 发布失败：按 token 删除处理中标记、移出执行中计数，score 恢复为当前时间。
|
||
// KEYS[1] 处理中标记；KEYS[2] 执行中计数；KEYS[3] 队列。ARGV：token、member、now_ms。
|
||
const srReleaseLua = `
|
||
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])
|
||
return 1
|
||
`
|
||
|
||
// deviceEvalRequestLua 请求设备判断：递增版本、合并场景、设备进入判断队列（同一设备一个成员，ZADD LT 不推后已有更早时间）。
|
||
// KEYS[1] 版本；KEYS[2] 场景集合；KEYS[3] 设备判断队列；KEYS[4] 处理中标记。ARGV：device_id、now_ms、场景…。
|
||
const deviceEvalRequestLua = `
|
||
redis.call('INCR', KEYS[1])
|
||
for i = 3, #ARGV do redis.call('SADD', KEYS[2], ARGV[i]) end
|
||
if redis.call('EXISTS', KEYS[4]) == 1 then return 1 end
|
||
redis.call('ZADD', KEYS[3], 'LT', tonumber(ARGV[2]), ARGV[1])
|
||
return 1
|
||
`
|
||
|
||
// deviceEvalClaimLua 领取到期设备判断：写处理中标记并记下版本与场景快照。
|
||
// KEYS[1] 设备判断队列；KEYS[2] 执行中计数。ARGV：now_ms、lease_ms、上限、token 前缀。
|
||
// 返回 {过期数, device_id, token, ver, scenes, ...}。
|
||
const deviceEvalClaimLua = `
|
||
local now = tonumber(ARGV[1])
|
||
local lease = tonumber(ARGV[2])
|
||
local limit = tonumber(ARGV[3])
|
||
local expired = redis.call('ZREMRANGEBYSCORE', KEYS[2], '-inf', now)
|
||
local result = {expired}
|
||
local capacity = limit - redis.call('ZCARD', KEYS[2])
|
||
if capacity <= 0 then return result end
|
||
local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, capacity * 2 + 10)
|
||
local taken = 0
|
||
for _, id in ipairs(ids) do
|
||
if taken >= capacity then break end
|
||
local claimKey = 'polling:device:' .. id .. ':eval_claim'
|
||
local token = ARGV[4] .. ':' .. id
|
||
if redis.call('SET', claimKey, token, 'NX', 'PX', lease) then
|
||
redis.call('ZADD', KEYS[1], 'XX', now + lease, id)
|
||
redis.call('ZADD', KEYS[2], now + lease, token)
|
||
taken = taken + 1
|
||
local ver = redis.call('GET', 'polling:device:' .. id .. ':eval_ver') or '0'
|
||
local scenes = redis.call('SMEMBERS', 'polling:device:' .. id .. ':eval_scenes')
|
||
table.insert(result, id)
|
||
table.insert(result, token)
|
||
table.insert(result, ver)
|
||
table.insert(result, table.concat(scenes, ','))
|
||
else
|
||
local ttl = redis.call('PTTL', claimKey)
|
||
if ttl < 0 then ttl = lease end
|
||
redis.call('ZADD', KEYS[1], 'XX', now + ttl, id)
|
||
end
|
||
end
|
||
return result
|
||
`
|
||
|
||
// deviceEvalFinishLua 设备判断收尾：只移除本次已处理的场景；版本变大时立即再判断（返回 2）；
|
||
// defer_ms > 0 表示本次未判断（旧路径持设备锁），推后再判断，场景不移除。
|
||
// KEYS[1] 处理中标记；KEYS[2] 执行权；KEYS[3] 执行中计数；KEYS[4] 版本；KEYS[5] 场景集合；KEYS[6] 队列。
|
||
// ARGV：token、device_id、领取时版本、now_ms、defer_ms、已处理场景…。
|
||
const deviceEvalFinishLua = `
|
||
if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end
|
||
redis.call('DEL', KEYS[1])
|
||
redis.call('DEL', KEYS[2])
|
||
redis.call('ZREM', KEYS[3], ARGV[1])
|
||
local now = tonumber(ARGV[4])
|
||
local deferMs = tonumber(ARGV[5])
|
||
if deferMs > 0 then
|
||
redis.call('ZADD', KEYS[6], now + deferMs, ARGV[2])
|
||
return 3
|
||
end
|
||
for i = 6, #ARGV do redis.call('SREM', KEYS[5], ARGV[i]) end
|
||
if (redis.call('GET', KEYS[4]) or '0') ~= ARGV[3] then
|
||
redis.call('ZADD', KEYS[6], now, ARGV[2])
|
||
return 2
|
||
end
|
||
redis.call('ZREM', KEYS[6], ARGV[2])
|
||
redis.call('DEL', KEYS[5])
|
||
return 1
|
||
`
|
||
|
||
var (
|
||
srWriteScript = redis.NewScript(srWriteLua)
|
||
srClaimScript = redis.NewScript(srClaimLua)
|
||
srFinishScript = redis.NewScript(srFinishLua)
|
||
srReleaseScript = redis.NewScript(srReleaseLua)
|
||
deviceRequestScript = redis.NewScript(deviceEvalRequestLua)
|
||
deviceClaimScript = redis.NewScript(deviceEvalClaimLua)
|
||
deviceFinishScript = redis.NewScript(deviceEvalFinishLua)
|
||
)
|