yixiongandClaude
b4c6652f66
集中修复 T8/T9/T11/T15 已实现代码的审查问题： - M2 令牌桶 Lua 改用 Redis TIME（消除多节点时钟偏斜） - M3 after 跨租户标记复查授权 + 无授权告警兜底 - m1 before 审计 MsgSeq/MsgRandom null 防御 - m2 HealthService 加 forceDown 开关，实测 DOWN 真返 503 - m3 countByStatus 排除 done（WHERE status IN (0,1,3)） - m5 配额查询异常不缓存（下次重试） - 附带 app.yml console charset GBK→UTF-8 M1 限流审计/事务收窄暂缓。 Co-Authored-By: Claude <noreply@anthropic.com>
90 lines
3.5 KiB
Java
90 lines
3.5 KiB
Java
package com.imutil.mapper;
|
||
|
||
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
||
import com.imutil.entity.DistQueue;
|
||
import org.apache.ibatis.annotations.Param;
|
||
import org.apache.ibatis.annotations.Update;
|
||
|
||
import java.time.OffsetDateTime;
|
||
import java.util.List;
|
||
|
||
/**
|
||
* 分发队列 Mapper
|
||
* <p>
|
||
* 提供 FOR UPDATE SKIP LOCKED 抢占消费的 SQL。
|
||
*
|
||
* @author imutil
|
||
*/
|
||
public interface DistQueueMapper extends BaseMapper<DistQueue> {
|
||
|
||
/**
|
||
* 抢占式拉取待分发记录：SELECT ... FOR UPDATE SKIP LOCKED
|
||
*
|
||
* @param limit 每轮条数
|
||
* @param lockedBy 抢占者标识
|
||
* @param lockTime 锁定时间
|
||
* @return 抢占到的记录
|
||
*/
|
||
@org.apache.ibatis.annotations.Select(
|
||
"SELECT id, msg_key, tenant_id, conv_id, target_url, payload, status, retry_count " +
|
||
"FROM dist_queue " +
|
||
"WHERE status = 0 AND next_retry_at <= #{now} " +
|
||
"ORDER BY next_retry_at, id " +
|
||
"LIMIT #{limit} " +
|
||
"FOR UPDATE SKIP LOCKED")
|
||
List<DistQueue> fetchPending(@Param("now") OffsetDateTime now,
|
||
@Param("limit") int limit);
|
||
|
||
/**
|
||
* 标记抢占：将 pending 置为 processing
|
||
*/
|
||
@Update("UPDATE dist_queue SET status = 1, locked_by = #{lockedBy}, locked_at = #{lockTime}, " +
|
||
"updated_at = #{lockTime} " +
|
||
"WHERE id = #{id} AND status = 0")
|
||
int lock(@Param("id") Long id,
|
||
@Param("lockedBy") String lockedBy,
|
||
@Param("lockTime") OffsetDateTime lockTime);
|
||
|
||
/**
|
||
* 标记分发成功
|
||
*/
|
||
@Update("UPDATE dist_queue SET status = 2, updated_at = #{now} WHERE id = #{id} AND status = 1")
|
||
int markDone(@Param("id") Long id, @Param("now") OffsetDateTime now);
|
||
|
||
/**
|
||
* 标记失败并安排重试（指数退避）
|
||
*/
|
||
@Update("UPDATE dist_queue SET status = CASE WHEN retry_count + 1 >= #{maxRetry} THEN 3 ELSE 0 END, " +
|
||
"retry_count = retry_count + 1, next_retry_at = #{nextRetryAt}, updated_at = #{now} " +
|
||
"WHERE id = #{id} AND status = 1")
|
||
int markRetry(@Param("id") Long id,
|
||
@Param("maxRetry") int maxRetry,
|
||
@Param("nextRetryAt") OffsetDateTime nextRetryAt,
|
||
@Param("now") OffsetDateTime now);
|
||
|
||
/**
|
||
* 巡检：重置超时未回执的 processing 回到 pending（工作线程宕机恢复）
|
||
*/
|
||
@Update("UPDATE dist_queue SET status = 0, locked_by = NULL, locked_at = NULL, updated_at = #{now} " +
|
||
"WHERE status = 1 AND locked_at < #{threshold}")
|
||
int recoverStuck(@Param("threshold") OffsetDateTime threshold, @Param("now") OffsetDateTime now);
|
||
|
||
/**
|
||
* 死信重发：将 status=3(dead) 的记录重置为 pending（retry_count 归零），供工作线程重新消费
|
||
*
|
||
* @return 影响行数（0 表示该记录非死信或已处理）
|
||
*/
|
||
@Update("UPDATE dist_queue SET status = 0, retry_count = 0, next_retry_at = now(), " +
|
||
"locked_by = NULL, locked_at = NULL, updated_at = now() " +
|
||
"WHERE id = #{id} AND status = 3")
|
||
int redeliverDead(@Param("id") Long id);
|
||
|
||
/**
|
||
* 按状态聚合计数（健康检查/监控用，一次扫描拿全状态分布）
|
||
*
|
||
* @return 每项含 status（0=pending/1=processing/2=done/3=dead）与 cnt
|
||
*/
|
||
@org.apache.ibatis.annotations.Select("SELECT status, count(*) AS cnt FROM dist_queue WHERE status IN (0, 1, 3) GROUP BY status")
|
||
java.util.List<java.util.Map<String, Object>> countByStatus();
|
||
}
|