Блог

Я пытаюсь реализовать простого производителя-потребителя, используя Java и Redis. Поток таков: производитель помещает элементы в Redis. Потребитель считывает элементы из Redis. Чтобы сделать потребителя более эффективным, он не будет читать каждый элемент отдельно, а вместо этого будет читать пакет элементов. Для этого я использую этот поток:

• Производитель отправляет элемент в ожидающий набор в Redis.
• Когда установленное количество превышает заданный порог, элементы упаковываются в пакет (с использованием JSON) и сохраняются в «готовую хэш-карту».
• Потребитель считывает готовую хэш-карту и оценивает ее содержимое.
• Потребитель удаляет элементы, которые он использовал, из 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}" amp;
done
  

С которым я работал, END=100 чтобы протестировать несколько вставок сразу.

Моя проблема в том, что после интеграции его с Java я начал его стресс-тестирование. Когда стресс-тестирование и несколько потоков запускались одновременно для обработки созданного содержимого — каждый из них выполнялся addAndSync.lua . Я проверил очередь готовности после вставки даже только 30 записей и заметил, что в очереди готовности есть дубликаты. Для меня это неожиданно, поскольку Redis гарантировал, что сценарии Lua будут блокировать любой другой вызов клиента. Мое ожидаемое поведение заключается в том, что каждый вызов Lua будет блокировать доступ к Redis до тех пор, пока он не зафиксирует свои результаты, и, таким образом, очередь готовности будет иметь уникальные элементы.

Я хотел бы получить любую помощь в понимании того, чего мне не хватает. Спасибо!!