强制 Oracle 返回带有 SKIP LOCKED 的 TOP N 行

Question

有一个fewquestions关于如何在Oracle和SQL Server中实现一个类似队列的表（锁定特定行，选择一定数量的行，跳过当前锁定的行）。

假设至少有 N 行符合条件，我如何保证检索到一定数量的 (N) 行？

据我所知，Oracle 在确定要跳过哪些行之前应用了WHERE 谓词。这意味着如果我想从表中提取一行，并且两个线程同时执行相同的 SQL，一个将接收该行，另一个将接收一个空结果集（即使有更多符合条件的行）。

这与 SQL Server 处理UPDLOCK、ROWLOCK 和READPAST 锁定提示的方式相反。在 SQL Server 中，TOP 似乎神奇地限制了在成功获得锁后的记录数。

注意，两个有趣的文章here和here。

甲骨文

CREATE TABLE QueueTest (
    ID NUMBER(10) NOT NULL,
    Locked NUMBER(1) NULL,
    Priority NUMBER(10) NOT NULL
);

ALTER TABLE QueueTest ADD CONSTRAINT PK_QueueTest PRIMARY KEY (ID);

CREATE INDEX IX_QueuePriority ON QueueTest(Priority);

INSERT INTO QueueTest (ID, Locked, Priority) VALUES (1, NULL, 4);
INSERT INTO QueueTest (ID, Locked, Priority) VALUES (2, NULL, 3);
INSERT INTO QueueTest (ID, Locked, Priority) VALUES (3, NULL, 2);
INSERT INTO QueueTest (ID, Locked, Priority) VALUES (4, NULL, 1);

在两个单独的会话中，执行：

SELECT qt.ID
FROM QueueTest qt
WHERE qt.ID IN (
    SELECT ID
    FROM
        (SELECT ID FROM QueueTest WHERE Locked IS NULL ORDER BY Priority)
    WHERE ROWNUM = 1)
FOR UPDATE SKIP LOCKED

注意第一个返回一行，第二个会话不返回一行：

会话 1

ID ---- 4

第 2 节

ID ----

---

SQL 服务器

CREATE TABLE QueueTest (
    ID INT IDENTITY NOT NULL,
    Locked TINYINT NULL,
    Priority INT NOT NULL
);

ALTER TABLE QueueTest ADD CONSTRAINT PK_QueueTest PRIMARY KEY NONCLUSTERED (ID);

CREATE INDEX IX_QueuePriority ON QueueTest(Priority);

INSERT INTO QueueTest (Locked, Priority) VALUES (NULL, 4);
INSERT INTO QueueTest (Locked, Priority) VALUES (NULL, 3);
INSERT INTO QueueTest (Locked, Priority) VALUES (NULL, 2);
INSERT INTO QueueTest (Locked, Priority) VALUES (NULL, 1);

在两个单独的会话中，执行：

BEGIN TRANSACTION
SELECT TOP 1 qt.ID
FROM QueueTest qt
WITH (UPDLOCK, ROWLOCK, READPAST)
WHERE Locked IS NULL
ORDER BY Priority;

请注意，两个会话都返回不同的行。

会话 1

ID ---- 4

第 2 节

ID ---- 3

如何在 Oracle 中获得类似的行为？

Answer 1

“据我所知，Oracle 在确定要跳过哪些行之前应用了 WHERE 谓词。”

是的。这是唯一可能的方法。在确定结果集之前，您不能从结果集中跳过一行。

答案就是不限制 SELECT 语句返回的行数。您仍然可以使用 FIRST_ROWS_n 提示来指示优化器您不会抓取完整的数据集。

调用 SELECT 的软件应该只选择前 n 行。在 PL/SQL 中，它会是

DECLARE
  CURSOR c_1 IS  
    SELECT /*+FIRST_ROWS_1*/ qt.ID
    FROM QueueTest qt
    WHERE Locked IS NULL
    ORDER BY PRIORITY
    FOR UPDATE SKIP LOCKED;
BEGIN
  OPEN c_1;
  FETCH c_1 into ....
  IF c_1%FOUND THEN
     ...
  END IF;
  CLOSE c_1;
END;

Answer 2

Gary Meyers 发布的解决方案几乎是我能想到的，除了使用 AQ，它可以为您完成所有这些以及更多。

如果您真的想避免使用 PLSQL，您应该能够将 PLSQL 转换为 Java JDBC 调用。您需要做的就是准备相同的 SQL 语句，执行它，然后继续对其进行单行提取（或 N 行提取）。

http://download.oracle.com/docs/cd/B10501_01/java.920/a96654/resltset.htm#1023642 的 Oracle 文档提供了一些如何在语句级别执行此操作的线索：

要设置查询的获取大小，请在执行查询之前对语句对象调用 setFetchSize()。如果将提取大小设置为 N，则每次访问数据库都会提取 N 行。

所以你可以用 Java 编写一些看起来像（在伪代码中）的东西：

stmt = Prepare('SELECT /*+FIRST_ROWS_1*/ qt.ID
FROM QueueTest qt
WHERE Locked IS NULL
ORDER BY PRIORITY
FOR UPDATE SKIP LOCKED');

stmt.setFetchSize(10);
stmt.execute();

batch := stmt.fetch();
foreach row in batch {
  -- process row
}
commit (to free the locks from the update)
stmt.close;

更新

根据下面的 cmets，建议使用 ROWNUM 来限制收到的结果，但在这种情况下不起作用。考虑这个例子：

create table lock_test (c1 integer);

begin
  for i in 1..10 loop
    insert into lock_test values (11 - i);
  end loop;
  commit;
end;
/

现在我们有一个有 10 行的表。请注意，我已经小心地以相反的顺序插入了行，包含 10 的行是第一个，然后是 9 等等。

假设您想要前 5 行，按升序排列 - 即从 1 到 5。您的第一次尝试是这样的：

select *
from lock_test
where rownum <= 5
order by c1 asc;

结果如下：

C1
--
6
7
8
9 
10

这显然是错误的，而且几乎是每个人都会犯的错误！查看查询的解释计划：

---

| Id  | Operation           | Name      | Rows  | Bytes | Cost (%CPU)| Time     |
---------------------------------------------------------------------------------
|   0 | SELECT STATEMENT    |           |     5 |    65 |     4  (25)| 00:00:01 |
|   1 |  SORT ORDER BY      |           |     5 |    65 |     4  (25)| 00:00:01 |
|*  2 |   COUNT STOPKEY     |           |       |       |            |          |
|   3 |    TABLE ACCESS FULL| LOCK_TEST |    10 |   130 |     3   (0)| 00:00:01 |
---------------------------------------------------------------------------------

Predicate Information (identified by operation id):
---------------------------------------------------

   2 - filter(ROWNUM<=5)

Oracle 自下而上执行计划 - 请注意，对 rownum 的过滤是在排序之前执行的，Oracle 按照它找到它们的顺序获取行（它们在此处插入的顺序 { 10, 9, 8, 7, 6})，在达到 5 行后停止，然后对该集合进行排序。

因此，要获得正确的前 5 个，您需要先进行排序，然后使用内联视图进行排序：

select * from
(
  select *
  from lock_test
  order by c1 asc
)
where rownum <= 5;

C1
--
1
2
3
4
5

现在，终于进入正题了 - 你能把 for update 跳过锁定在正确的位置吗？

select * from
(
  select *
  from lock_test
  order by c1 asc
)
where rownum <= 5
for update skip locked;

这给出了一个错误：

ORA-02014: cannot select FOR UPDATE from view with DISTINCT, GROUP BY, etc

试图将 for 更新移动到视图中会出现语法错误：

select * from
(
  select *
  from lock_test
  order by c1 asc
  for update skip locked
)
where rownum <= 5;

唯一可行的是以下，会给出错误的结果：

  select *
  from lock_test
  where rownum <= 5
  order by c1 asc
  for update skip locked;

事实上，如果你在会话 1 中运行这个查询，然后在会话 2 中再次运行，会话 2 将给出零行，这真的是错误的！

那你能做什么？打开游标并从中获取您想要的行数：

set serveroutput on

declare
  v_row lock_test%rowtype;
  cursor c_lock_test
  is
  select c1
  from lock_test
  order by c1
  for update skip locked;
begin
  open c_lock_test;
  fetch c_lock_test into v_row;
  dbms_output.put_line(v_row.c1);
  close c_lock_test;
end;
/    

如果您在会话 1 中运行该块，它将打印出“1”，因为它锁定了第一行。然后在会话 2 中再次运行它，它会打印“2”，因为它跳过了第 1 行并获得了下一个空闲行。

这个例子是在 PLSQL 中，但是在 Java 中使用 setFetchSize 你应该能够得到完全相同的行为。

Answer 3

在您的第一个会话中，当您执行时：

SELECT qt.ID
FROM QueueTest qt
WHERE qt.ID IN (
    SELECT ID
    FROM
        (SELECT ID FROM QueueTest WHERE Locked IS NULL ORDER BY Priority)
    WHERE ROWNUM = 1)
FOR UPDATE SKIP LOCKED

您的内部选择尝试仅抓取 id=4 并将其锁定。这是成功的，因为这一行还没有被锁定。

在第二个会话中，您的内部选择仍然尝试抓取 ONLY id=4 并将其锁定。这不成功，因为该单行仍被第一个会话锁定。

现在，如果您在第一个会话中更新了“锁定”字段，则运行该选择的下一个会话将获取 id=3。

基本上，在您的示例中，您依赖于未设置的标志。要使用您的锁定标志，您可能意味着执行以下操作：

1. 根据某些标准选择您想要的 ID。
2. 立即更新这些 ID 的锁定标志 = 1（如果资源繁忙，另一个会话会在此步骤中超过 1 个或多个 ID，请再次转到 1）
3. 对这些 ID 执行任何操作
4. 将锁定标志更新回 null

然后您可以使用您的 select for update 跳过锁定语句，因为您的锁定标志正在维护。

就个人而言，我不喜欢标志的所有更新（您的解决方案可能出于任何原因需要它们），所以我可能只是 try 选择我想要更新的 ID（通过任何标准）在每个会话中：

select * from queuetest where ... for update skip locked;

例如（实际上，我的标准不会基于 id 列表，但 queuetest 表过于简单）：

• sess 1：从 queuetest 中选择 * (4,3) 中的 id 用于更新跳过锁定；

• sess 2：从 queuetest 中选择 * id in (4,3,2) 用于更新跳过锁定；

这里 sess1 会锁定 4,3 而 sess2 只会锁定 2。

据我所知，您不能在 select for update 语句中执行 top-n 或使用 group_by/order_by 等，您将得到 ORA-02014。

Answer 4

我的解决方案 - 是这样编写存储过程：

CREATE OR REPLACE FUNCTION selectQueue 
RETURN SYS_REFCURSOR
AS
  st_cursor SYS_REFCURSOR;
  rt_cursor SYS_REFCURSOR;
  i number(19, 0);

BEGIN

  open st_cursor for
  select id
  from my_queue_table
  for update skip locked;

  fetch st_cursor into i;
  close st_cursor;

  open rt_cursor for select i as id from dual;
  return  rt_cursor;

 END;

这是一个简单的示例 - 返回 TOP FIRST 非阻塞行。要检索 TOP N 行 - 将单次提取替换为局部变量 ("i") 与循环提取到临时表中。

PS：返回光标 - 用于休眠友谊。

Answer 5

我遇到了这个问题，我们花了很多时间来解决它。有的使用for updatefor update skip locked，在oracle 12c中，一个新的方法是使用fetch first n rows only。但是我们使用的是 oracle 11g。

最后，我们尝试了这个方法，发现效果很好。

CURSOR c_1 IS  
   SELECT *
     FROM QueueTest qt
     WHERE Locked IS NULL
     ORDER BY PRIORITY;
   myRow c_1%rowtype;
   i number（5）:=0;
   returnNum := 10;
BEGIN
  OPEN c_1;
  loop 
    FETCH c_1 into myRow 
    exit when c_1%notFOUND 
    exit when i>=returnNum;
    update QueueTest set Locked='myLock' where id=myrow.id and locked is null;
    i := i + sql%rowcount;
  END
  CLOSE c_1;
  commit;
END;

我是用记事本写的，所以可能有问题，您可以将其修改为程序或其他。

Answer 6

首先感谢前 2 个答案..从他们那里学到了很多东西。我测试了以下代码，在运行 Practicedontdel.java main 方法后，我发现这两个类每次都打印不同的行。如果在任何情况下此代码可能会失败，请告诉我。（PS：感谢堆栈溢出）

Practicedontdel.java：

    Connection conn = null;
    PreparedStatement ps = null;
    ResultSet rs =null;
    String val="";
    int count =0;

        conn = getOracleConnection();
        conn.setAutoCommit(false);
        ps = prepareStatement(conn,"SELECT /*+FIRST_ROWS_3*/ t.* from 
        REPROCESS_QUEUE t FOR UPDATE SKIP LOCKED");
        ps.setFetchSize(3);
        boolean rss = ps.execute();
        rs = ps.getResultSet();
        new Practisethread().start();
        while(count<3 && rs.next())
        {
            val = rs.getString(1);
            System.out.println(val);
            count++;
            Thread.sleep(10000);
        }
       conn.commit();
            System.out.println("end of main program");

Practisethread.java：在run()中：

            conn = getOracleConnection();
            conn.setAutoCommit(false);
            ps = prepareStatement(conn,"SELECT /*+FIRST_ROWS_3*/ t.* from REPROCESS_QUEUE t FOR UPDATE SKIP LOCKED");
            ps.setFetchSize(3);
            boolean rss = ps.execute();
            rs = ps.getResultSet();
            while(count<3 && rs.next())
            {
                val = rs.getString(1);
                System.out.println("******thread******");
                System.out.println(val);
                count++;
                Thread.sleep(5000);
            }
            conn.commit();
            System.out.println("end of thread program");