基于redis乐观锁实现并发排队

基于Redis乐观锁实现并发排队的攻略需要分为以下几步：

第一步：确定Redis存储设计

首先需要确定Redis存储设计，用于存储患者排队信息。可以考虑使用Redis Sorted Set，Set中的元素是待处理的患者排队信息，Sorted Set中的元素则是已处理的患者排队信息，Score则是时间戳用来排序，患者排队信息应该至少包含患者姓名、身份证号码、就诊日期、预约时间和就诊科室等信息。

zadd queue 1632289200 王大锤_411111111111111111_2021-09-22_09:00_内科

第二步：实现排队逻辑

患者到达医院，需要进行排队，可以将该患者的排队信息放入Redis中。需要使用Redis乐观锁来保证多个并发客户端对同一个患者排队信息进行修改时，不会造成数据混乱。修改排队信息时可以使用Redis事务，先对ZSET中待处理的患者排队信息进行检查，若患者排队信息已存在，则不做处理，若不存在，则把该患者排队信息加入ZSET。

def enqueue(patient_info):  
    # redis 连接池配置
    redis_pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    # 初始化 redis 连接对象
    redis_conn = redis.Redis(connection_pool=redis_pool)

    # 当前时间戳
    curr_time = int(time.time())
    with redis_conn.pipeline() as pipeline:
        while True:
            try:
                pipeline.watch("queue")
                #获取队列中已有的患者信息
                patients = pipeline.zrange("queue", 0, -1)
                #检查当前患者是否有预约记录
                for item in patients:
                    if item.split('_')[1] == patient_info["ID"]:
                        return "您已有预约记录，请勿重复预约。"

                #检查当前患者有无重复排队信息，若无则加入待排队队列
                if patient_info["name"]+'_'+patient_info["ID"]+'_'+patient_info["date"]+'_'+patient_info["time"]+'_'+patient_info["department"] not in patients:
                    pipeline.multi()
                    pipeline.zadd("queue", {patient_info["name"]+'_'+patient_info["ID"]+'_'+patient_info["date"]+'_'+patient_info["time"]+'_'+patient_info["department"]: curr_time})
                    pipeline.execute()
                    return "预约成功！"
                else:
                    return "已在排队队列中，请耐心等待。"
            except redis.WatchError:
                continue
            finally:
                #归还连接到连接池中
                pipeline.reset()
                redis_pool.disconnect()

第三步：实现出队逻辑

已处理的患者排队信息需要从待处理队列中删除并加入Sorted Set中。为保证出队操作原子性需要使用Redis Pipeline事务。

def dequeue():
    # redis 连接池配置
    redis_pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=0, decode_responses=True)
    # 初始化 redis 连接对象
    redis_conn = redis.Redis(connection_pool=redis_pool)

    with redis_conn.pipeline(transaction=True) as pipeline:
        while True:
            try:
                pipeline.watch("queue")
                #获取待处理队列中的患者信息
                patients = pipeline.zrange("queue", 0, -1)
                #取出第一个患者信息
                if patients:
                    first_patient = patients[0]
                    #把取出的第一个患者排队信息放入已处理队列
                    pipeline.multi()
                    pipeline.zadd("done", {first_patient: int(time.time())})
                    pipeline.zrem("queue", first_patient)
                    pipeline.execute()

                    #返回患者的排队信息
                    return first_patient
                else:
                    return "当前队列为空，无法出队。"
            except redis.WatchError:
                continue
            finally:
                #归还连接到连接池中
                pipeline.reset()
                redis_pool.disconnect()

第四步：调用排队和出队接口

在flask中定义两个接口，一个用于排队，一个用于出队。

from flask import Flask, request

app = Flask(__name__)

@app.route("/enqueue", methods=["POST"])
def en_queue():
    patient_info = request.json
    msg = enqueue(patient_info)
    return msg

@app.route("/dequeue")
def de_queue():
    msg = dequeue()
    return msg

if __name__ == "__main__":
    app.run(debug=True)

示例

假设当前时间是2021年9月22日08:00，患者小张要预约09:00号的内科检查。多个患者同时调用排队接口进行预约，最终只有一个患者预约成功。

1. 患者小李调用排队接口请求预约09:00点内科检查。

POST /enqueue HTTP/1.1
content-type: application/json

{
    "name": "李华",
    "ID": "420123198012122222",
    "date": "2021-09-22",
    "time": "09:00",
    "department": "内科"
}

1. 患者小张调用排队接口请求预约09:00点内科检查。

POST /enqueue HTTP/1.1
content-type: application/json

{
    "name": "张三",
    "ID": "420123198012123333",
    "date": "2021-09-22",
    "time": "09:00",
    "department": "内科"
}

1. 患者小李再次调用排队接口请求预约09:00点内科检查。

POST /enqueue HTTP/1.1
content-type: application/json

{
    "name": "李华",
    "ID": "420123198012122222",
    "date": "2021-09-22",
    "time": "09:00",
    "department": "内科"
}

1. 患者小王调用排队接口请求预约10:00点外科检查。

POST /enqueue HTTP/1.1
content-type: application/json

{
    "name": "王五",
    "ID": "420123198012124444",
    "date": "2021-09-22",
    "time": "10:00",
    "department": "外科"
}

1. 患者小张调用出队接口进行就诊。

GET /dequeue HTTP/1.1

执行5成功后，第一次的李华请求排队的数据在逻辑上已经过期了，可以从待处理队列中删除，实际上原始汇报中的代码的第14行实现了这个操作（在425行有具体操作）。