yixiongandClaude
d22fdcfaad
- 应用骨架: Solon + MyBatis-Plus + PostgreSQL + Redis + Sa-Token - 核心链路: 回调网关 → 消息落库(幂等) → 分发队列(FOR UPDATE SKIP LOCKED) → worker消费/重试/死信 - 多租户隔离: 前缀法账号映射, 所有业务表带 tenant_id - 管理后台: FreeMarker 渲染, 租户/授权/队列/用量管理页面 - 历史消息补拉: 水位线驱动, getRoamMsg/getGroupMsg 增量拉取, 兜底回调丢失 - msg_key 统一算法(回调/补拉共享, 跨路径去重) Co-Authored-By: Claude <noreply@anthropic.com>
82 lines
3.2 KiB
Java
82 lines
3.2 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);
|
||
}
|