多线程场景下PostgreSQL按recdate选记录乱序问题求助

当前问题的核心是：多线程并发调用函数时，可能跳过某otherid的旧消息，先选中同otherid的新消息并锁定，导致处理顺序颠倒。根源在于原函数仅全局按recdate取最旧单条消息，未确保选中的是对应otherid的最旧未处理消息，且并发场景下无行锁保护，导致线程间的竞态问题。

1. 核心逻辑调整 #

确保每次仅选中每个otherid的最旧未处理消息，再全局按这些消息的recdate排序，优先处理全局最旧的otherid的最旧消息。

2. 并发控制优化 #

使用FOR UPDATE SKIP LOCKED锁定选中的消息行，避免其他线程重复选中；同时处理locktbl的唯一约束冲突，防止线程因竞态报错。

3. 索引优化 #

添加合适的索引提升查询效率，确保排序和过滤的正确性：

-- 优化tbl1的查询和排序性能
CREATE INDEX idx_tbl1_otherid_status_recdate ON tbl1(otherid, statuscode, recdate);
-- locktbl已存在otherid唯一约束，无需额外索引

4. 修改后的函数代码 #

CREATE OR REPLACE FUNCTION public.fn_getnextmessage(
    )
    RETURNS TABLE(ret_recid bigint, ret_otherid bigint, ret_msgbody text) 
    LANGUAGE 'plpgsql'
    COST 100
    VOLATILE PARALLEL SAFE 
    ROWS 1000

AS $BODY$
declare
    _recid bigint := null;
    _otherid bigint := null;
    _msgbody text := null;
    _lock_inserted boolean := false;
BEGIN
    -- 用窗口函数筛选每个otherid的最旧未处理消息，再全局选最旧的一条
    WITH eligible_messages AS (
        SELECT 
            recid,
            otherid,
            msgbody,
            recdate,
            -- 给每个otherid的消息按recdate升序编号，取第一条（最旧）
            ROW_NUMBER() OVER (PARTITION BY otherid ORDER BY recdate ASC) AS rn
        FROM tbl1
        WHERE recdate < (now() - interval '15 second')
          AND statuscode = 'New'
          AND NOT EXISTS (SELECT 1 FROM locktbl l WHERE l.otherid = tbl1.otherid)
    )
    SELECT recid, otherid, msgbody
    INTO _recid, _otherid, _msgbody
    FROM eligible_messages
    WHERE rn = 1 -- 仅保留每个otherid的最旧消息
    ORDER BY recdate ASC -- 全局按消息时间升序，优先处理最旧的
    FOR UPDATE SKIP LOCKED -- 锁定选中行，其他线程跳过已锁定行
    LIMIT 1;

    IF _recid IS NOT NULL THEN 
        -- 插入锁表，处理唯一约束冲突（避免多线程同时锁定同一otherid）
        BEGIN
            INSERT INTO locktbl (otherid, lockdate)
            VALUES (_otherid, now())
            ON CONFLICT (otherid) DO NOTHING;
            -- 检查是否成功插入锁记录
            GET DIAGNOSTICS _lock_inserted = ROW_COUNT;
        EXCEPTION
            WHEN unique_violation THEN
                _lock_inserted := false;
        END;

        IF _lock_inserted THEN
            -- 更新消息状态为处理中
            UPDATE tbl1 t1
            SET statuscode = 'InProcess'
            WHERE t1.recid = _recid;            

            -- 返回选中的消息
            RETURN QUERY SELECT _recid, _otherid, _msgbody;
        ELSE
            -- 其他线程已锁定该otherid，返回空结果
            RETURN QUERY SELECT NULL::bigint, NULL::bigint, NULL::text LIMIT 0;
        END IF;
    ELSE
        -- 无符合条件的消息
        RETURN QUERY SELECT NULL::bigint, NULL::bigint, NULL::text LIMIT 0;    
    END IF;
END;
$BODY$;

• 窗口函数PARTITION BY otherid ORDER BY recdate ASC：确保每个otherid仅筛选出最旧的未处理消息，从根源上避免选中同otherid的新消息。
• FOR UPDATE SKIP LOCKED：锁定选中的消息行，其他线程会跳过已锁定的行，防止重复选中同一消息。
• ON CONFLICT DO NOTHING：处理多线程同时尝试锁定同一otherid的竞态问题，插入失败时返回空结果，让线程重新尝试。

内容的提问来源于stack exchange，提问作者M M