Python多线程与非线程场景下Redis索引锁无法序列化问题求助

核心思路是抛弃本地进程锁，利用Redis本身的原子操作保证索引唯一性——Redis的单线程特性可天然保证命令/脚本的原子性，完全适配多进程（gunicorn多Worker）、多线程混合场景，同时避免锁序列化问题。

1. 修改 rotator.py（替换本地锁为Redis原子操作） #

用Redis Lua脚本实现「自增+阈值重置」的原子逻辑，替代原本地锁+分步读写的非原子方案：

import redis
from datetime import datetime

# 初始化Redis连接（建议用连接池，减少连接开销）
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 定义Lua脚本，原子完成索引自增、阈值判断、重置、返回结果
INDEX_ROTATE_SCRIPT = """
local current = redis.call('GET', KEYS[1])
if not current then
    current = 0
end
current = current + 1
if tonumber(current) > tonumber(ARGV[1]) then
    current = 0
end
redis.call('SET', KEYS[1], current)
return current
"""
# 预加载脚本，避免每次调用重复传输
script_sha = redis_client.script_load(INDEX_ROTATE_SCRIPT)

def increment_index():
    time_before = datetime.now()
    # 执行Lua脚本，原子获取并更新索引
    new_index = redis_client.evalsha(script_sha, 1, "app_index", 200)
    return int(new_index)

def rotator_of_index():
    new_index = increment_index()
    # 保留原日志等业务逻辑
    return new_index

2. salary_worker.py 无需修改 #

原代码直接复用即可，rotator_of_index的实现已改为Redis原子操作，单线程/多线程调用均安全：

from rotator import rotator_of_index

def do_work(n):
    index = rotator_of_index()
    # 保留原业务逻辑
    return f"result_{n}_index_{index}"

# 单线程调用逻辑
def do_more_work(n):
    something = do_work(n)
    # 保留原后续逻辑
    return something

3. do_with_threading.py 无需修改 #

线程池可正常使用，不再有锁序列化问题，所有并发控制由Redis原子操作处理：

from salary_worker import do_work
from multiprocessing.dummy import Pool as ThreadPool

def do_work_with_threading():
    list_of_n = [1, 2, 3, 4, 5]
    pool = ThreadPool(15)
    list_of_results = pool.map(do_work, list_of_n)
    pool.close()
    pool.join()
    print(list_of_results)

关键优势 #

• 跨场景安全：Redis原子操作覆盖gunicorn多Worker（多进程）和线程池（多线程）场景，不会出现重复索引
• 无序列化问题：彻底移除本地multiprocessing.Lock，避免线程池传递时的序列化报错
• 代码结构兼容：完全符合3个文件拆分的要求，无需调整现有调用关系
• 性能更优：Redis原子操作比本地锁+分步Redis读写减少网络往返，效率更高

内容的提问来源于stack exchange，提问作者Nanno