我正在尝试使用java和redis实现一个简单的生产者-消费者。流程是这样的：生产者将项目推入redis。消费者从redis读取项目。为了使消费者更有效率-它不会单独读取每个项目，而是读取一批项目。为此，我使用以下流程：
producer将项目推送到redis中的挂起集。
当集合计数超过给定的阈值时，这些项被打包到一个批中（使用json）并保存到一个“ready hashmap”中。
使用者读取ready hashmap并评估其内容。
使用者从hashmap中删除它所使用的项。
由于这些操作可能导致竞争条件，因此我研究了事务性操作。我知道实现它的最好方法是使用lua，因为redis lua脚本是原子的。
我写的剧本是这样的：

local toJsonList = function(items)
    local jsonList = '[' .. table.concat(items, ', ') .. ']'
    return cjson.encode(jsonList)
end

-- Get the batch size (the threshold)
local batchSize = tonumber(ARGV[1])

-- Add the new item
redis.call('SADD', KEYS[1], ARGV[2])

-- Get the number of pending items in Redis
local currentPendingQueueItems = redis.call('SCARD', KEYS[1])

-- Should we move the items from the pending queue to the ready queue?
if currentPendingQueueItems < batchSize then
    return 1
end

-- Fetch the items stored in the pending queue
local pendingItems = redis.call('SMEMBERS', KEYS[1])

-- Store the items in the ready queue hash map
redis.call('HSET', KEYS[3], cjson.encode(KEYS[4]), toJsonList(pendingItems))

-- Remove the pending queue
redis.call('DEL', KEYS[1])
return 1

执行过程如下：

$ redis-cli --eval addAndSync.lua "pending-queue-key" "ready-queue-key" "unique-key-for-batch" , $THRESHOLD "item to add"

我开始单独测试它，它确实工作得很好。就绪队列已正确同步。我甚至写了这个剧本：

# !/bin/bash

END=$1
for i in $(seq 1 $END);
  do redis-cli --eval syncReadyQueue.lua "pending" "ready" "ready${i}" , "3" "${i}" &
done

和我一起跑的 END=100 一次测试一些插入。
我的问题是，在将它与java集成之后，我开始对它进行压力测试。当压力测试和几个线程同时启动以处理生成的内容时，每个线程都会运行 addAndSync.lua . 在插入了30条记录之后，我检查了就绪队列，发现就绪队列中有重复的记录。这对我来说是出乎意料的，因为redis保证lua脚本将阻止任何其他客户端调用。我预期的行为是，每个lua调用都会阻止对redis的访问，直到它提交结果为止，因此就绪队列将具有唯一的项。
我很想得到任何帮助来理解我错过了什么。谢谢！！