在Redis中，如何保证在多客户端环境中从列表中获取N个项目？

Question

假设Redis中有一个键K，它保存着一个值列表。

许多producer客户端正在使用LPUSH或RPUSH将元素逐个添加到此列表中。

另一方面，另一组消费者客户端正在从列表中弹出元素，尽管有一定的限制。只有当列表中至少包含N个项目时，使用者才会尝试弹出N个项目。这样可以确保消费者在完成弹出过程后将N件物品放在手中

如果列表包含少于N个项目，消费者甚至不应该尝试从列表中弹出元素，因为他们最后不会有至少N个项目。

如果只有1个消费者客户端，客户端可以简单地运行LLEN命令来检查列表是否包含至少N个项目，并使用LPOP/RPOP减去N。

但是，如果有许多消费者客户端，则可能存在竞争条件，在阅读LLEN

使用单独的锁定系统似乎是解决此问题的一种方法，但我很好奇，这种类型的操作是否只能使用Redis命令来完成，例如Multi/Exec/Watch等。

我检查了Multi/Exec方法，似乎它们不支持回滚。此外，在Multi/Exec事务之间执行的所有命令都将返回“QUEUED”，因此我无法知道我将在事务中执行的N个LPOP是否都将返回元素。

Answer 1

您可以使用预取器。

你可以有一个预取器来构建一个大小=6的包，而不是每个消费者贪婪地从队列中挑选一个项目，这会导致“到处都是水，但一滴也喝不到”的问题。当预取器有一个完整的数据包时，它可以将项目放在一个单独的数据包队列中（另一个带有数据包列表的redis键），并在一个事务中从主队列中弹出项目。本质上，你写的：

如果只有一个客户机，客户机只需运行LLEN命令检查列表是否包含至少N个项目，然后使用LPOP/RPOP减去N。

如果预取器没有一个完整的数据包，它什么也不做，一直等待主队列大小达到6。

在用户端，他们只需查询预取的数据包队列，然后弹出最上面的数据包，就可以了。它始终是1个预构建包（大小=6项）。如果没有可用的数据包，它们将等待。

在生产者方面，不需要进行任何更改。他们可以继续插入主队列。

顺便说一句，可以有多个预取器任务同时运行，它们可以在它们之间同步对主队列的访问。

可以使用buffet表类比来描述预取器的实现。可以把排队的队伍想象成餐厅的自助餐台，客人可以在那里取食物离开。礼仪要求客人排队等候。预取程序也会遵循类似的方法。以下是算法：

Algorithm Prefetch
Begin
   while true
      check = main queue has 6 items or more    // this is a queue read. no locks required
      if(check == true)
         obtain an exclusive lock on the main queue
         if lock successful
            begin a transaction
            create a packet and fill it with top 6 items from 
              the queue after popping them
            add the packet to the prefetch queue

            if packet added to prefetch queue successfully
              commit the transaction
            else
              rollback the transaction
            end if

            release the lock
            
         else
            // someone else has the excl lock, we should just wait
            sleep for xx millisecs
         end if
      end if
   end while

End

为了简单起见，我在这里展示了一个无限的轮询循环。但是这可以通过Redis通知使用pub/sub模式来实现。因此，预取器只是等待主队列键正在接收LPUSH的通知，然后在上面的循环正文中执行内的逻辑。

还有其他方法可以做到这一点。但这会给你一些想法。

Answer 2

因此，您所需要的只是一种原子方法来检查列表长度并有条件地弹出。

这就是Lua脚本的用途，请参见EVAL命令。

下面是一个Lua脚本，让您开始：

local len = redis.call('LLEN', KEYS[1])
if len >= tonumber(ARGV[1]) then
  local res = {n=len}
  for i=1,len do
    res[i] = redis.call('LPOP', KEYS[1])
  end
  return res
else
  return false
end

用作

EVAL "local len = redis.call('LLEN', KEYS[1]) \n if len >= tonumber(ARGV[1]) then \n   local res = {n=len} \n   for i=1,len do \n     res[i] = redis.call('LPOP', KEYS[1]) \n   end \n   return res \n else \n   return false \n end" 1 list 3

如果列表至少有那么多元素，这只会从列表中弹出ARGV[1]元素（键名后面的数字）。

Lua脚本以原子方式运行，因此读取客户端之间不存在竞争条件。

正如OP在注释中指出的那样，存在数据丢失的风险，比如因为LPOP和脚本返回之间的电源故障。您可以使用RPOPLPUSH代替LPOP，将元素存储在临时列表中。然后还需要一些跟踪、删除和恢复逻辑。请注意，您的客户端也可能会死亡，留下一些未处理的元素。

您可能想看看Redis Streams。此数据结构非常适合在多个客户端之间分配负载。当与消费者组一起使用时，它有一个挂起条目列表（PEL），充当临时列表。

然后，客户端执行XACK以在处理后从PEL中删除元素。然后，您还可以免受客户端故障的影响。

Redis Streams对于解决您试图解决的复杂问题非常有用。你可能想免费上这门课。