Postgresql FOR UPDATE SKIP LOCKED 仍然选择重复的行答案

【问题标题】：Postgresql FOR UPDATE SKIP LOCKED still selects duplicated rowsPostgresql FOR UPDATE SKIP LOCKED 仍然选择重复的行
【发布时间】：2016-08-14 02:59:20
【问题描述】：

我使用 PostgreSQL 作为作业队列。以下是我检索作业并更新其状态的查询：

        UPDATE requests AS re
        SET
          started_at = NOW(),
          finished_at = NULL
        FROM (
          SELECT
            _re.*
          FROM requests AS _re
          WHERE
            _re.state = 'pending'
          AND
            _re.started_at IS NULL
          LIMIT 1
          FOR UPDATE SKIP LOCKED
        ) AS sub
        WHERE re.id = sub.id
        RETURNING
          sub.*

现在，我有几台机器，在每台机器上我有 1 个带有多个线程的进程，并且在每个线程上我有一个工人。同一进程中的所有工作人员共享一个连接池，通常有 10 - 20 个连接。

问题是，上面的查询会多次返回一些行！

我找不到任何原因。有人可以帮忙吗？

更详细地说，我使用的是 Python3 和 psycopg2。

更新：

我试过@a_horse_with_no_name 的回答，但似乎不行。

我注意到，两个查询检索到一个请求，started_at 更新为：

2016-04-21 14:23:06.970897+08

和

2016-04-21 14:23:06.831345+08

仅相差 0.14 秒。

我想知道当这两个连接执行内部 SELECT 子查询时，两个锁都没有建立？

更新：

更准确地说，我在 1 台机器上的 1 个进程中有 200 个工人（即 200 个线程）。

【问题讨论】：

跨帖dba.stackexchange.com/q/136030/1822
在 postgresql 9.1 中是否可以使用 FOR UPDATE SKIP LOCKED？我不这么认为
@e4c5，我使用的是 9.5
对不起，我以为我看到了 9.1
您的员工数量是否多于您的人脉数量？

标签： multithreading postgresql psycopg2

【解决方案1】：

还请注意，如果您不希望它们相互干扰，则每个线程都有自己的连接是必不可少的。

如果您的应用程序使用多个执行线程，它们不能同时共享一个连接。您必须明确控制访问连接（使用互斥锁）或为每个连接使用一个连接线。如果每个线程都使用自己的连接，则需要使用 AT 子句指定线程将使用哪个连接。

来自：http://www.postgresql.org/docs/9.5/static/ecpg-connect.html

如果两个线程共享同一个连接，就会发生各种奇怪的事情。我相信这就是你的情况。如果您对一个连接进行锁定，则使用同一连接的所有其他线程都可以访问锁定的对象。

请允许我提出一种替代方法，这非常简单。使用redis作为队列。您可以简单地使用 redis-py 和 lpush/rpop 方法，也可以使用 python-rq。

【讨论】：

谢谢！我稍后会尝试。只是一个简单的问题，因为我有 200 个工人，这意味着我必须建立 200 个连接。那么如果我有 50 台机器，那么总共有 10000 个连接。 postgresql 可以容纳这么多的连接吗？
用建议的替代方法更新答案
@HanXu：10000 个连接肯定太多了。我什至怀疑拥有 200 个并发工作人员会不会让事情变得更快，除非你有一个非常、非常功能强大且具有许多个内核 (> 100) 的数据库服务器和很多个硬盘。您应该将您的工作人员（在数据库中执行某些操作）限制为不会使数据库超载的合理数字。或者使用连接池，确保单个连接永远不会被两个并发线程使用。

【解决方案2】：

有可能在选择时尚未发出锁定事务，或者在选择结果准备好并且更新语句开始时锁定已丢失。您是否尝试过明确开始交易？

BEGIN;
  WITH req AS (
    SELECT id
    FROM requests AS _re
    WHERE _re.state = 'pending' AND _re.started_at IS NULL
    LIMIT 1 FOR UPDATE SKIP LOCKED
    )
  UPDATE requests SET started_at = NOW(), finished_at = NULL
  FROM req
  WHERE requests.id = req.id;
COMMIT;

【讨论】：