PostgreSQL for update skip locked 实现队列功能

数据准备

create table if not exists queue(id int, content text, status text);
insert into queue values(1, '队列元素1', 'pending');
insert into queue values(2, '队列元素2', 'pending');
insert into queue values(3, '队列元素3', 'pending');
insert into queue values(4, '队列元素4', 'pending');
insert into queue values(5, '队列元素5', 'pending');

skip locked

使用for update时，如果一行并其他session读取，则会一直等待直到timeout；

如果加了nowait，则当选取的行被其他session读取时会直接抛出异常。

如果加了skip locked，则当选取的行被其他session读取时会跳过改行，根据where条件返回或，或其他行。

示例

skip locked 返回空

session 1在事务中选择id为1的行。

z-blog=# begin;
BEGIN
z-blog=*# select * from queue where id = 1 for update skip locked;
 id |  content  | status
----+-----------+---------
  1 | 队列元素1 | pending
(1 row)

session 2同样选择id为1的行，此时返回为空。

z-blog=# select * from queue where id = 1 for update skip locked;
 id | content | status
----+---------+--------
(0 rows)

skip locked 返回其他行

session 1在事务中选择了status为pending的任意一条。

z-blog=# begin;
BEGIN
z-blog=*# select * from queue where status='pending' limit 1 for update skip locked;
 id |  content  | status
----+-----------+---------
  1 | 队列元素1 | pending
(1 row)

session 2同样选择了status为pending的任意一条，返回时跳过了id为1的行，从id为2的行开始选取一条。

z-blog=# select * from queue where status='pending' limit 1 for update skip locked;
 id |  content  | status
----+-----------+---------
  2 | 队列元素2 | pending
(1 row)

队列功能

begin; -- begin transaction
with temp as (
    select id from queue where status='pending' limit 1 for update skip locked
)
update queue set status='succeeded' where queue.id = (select id from temp)
returning *;
-- bussiness logic
commit; -- commit transaction

查看数据

z-blog=# select * from queue;
 id |  content  |  status
----+-----------+-----------
  2 | 队列元素2 | pending
  3 | 队列元素3 | pending
  4 | 队列元素4 | pending
  5 | 队列元素5 | pending
  1 | 队列元素1 | succeeded
(5 rows)