如何改进异步锁机制

Question

下面是我的代码，我在其中实现了异步锁定机制，如果具有相同名称的请求已经在执行并且尚未完成，那么该机制应该阻止方法中的请求，这工作正常，但问题是如果请求带有不同的名称也被阻止，这并不理想，理想情况下，如果请求带有不同的请求名称，则应该开始执行而无需等待

import asyncio
from contextlib import asynccontextmanager
 
@asynccontextmanager
async def get_lock(req_name_):
    locks = {}
    logger.info(f"Creating lock for stack {req_name_} if not created")
    if not locks.get(req_name_):
        logger.info("creating key for a lock")
        locks[req_name_] = asyncio.Lock()
    async with locks[req_name_]:
        yield
    if len(locks[req_name_]._waiters) == 0:
        del locks[req_name_]
        logger.info(f"lock released")
    logger.info(len(locks))

async def handle_lock_request(req_json_):
    logger.info(f"ocupying lock")
    req_name = req_json_.get('req_name')
    async with get_lock(req_name):
        logger.info(f"lock acquired by stack {req_name}")
        await _handle_request(req_json_)

async def _req_handler():
    tasks = []
    loop = asyncio.get_running_loop()
    logger.debug("Await receiver.recv_string")
    req = await receiver.recv_string()
    logger.debug(f"Request received {req}")
    req_json = json.loads(req)
    logger.debug("Await create_task")
    tasks.append(loop.create_task(handle_lock_request(req_json)))
    await asyncio.gather(*[task for task in tasks if not task.done()])
  
def _handle_request(req_json_):
    # ...
    # ...
    logger.info(f"Request finished with req name {req_name} for action patch stack")

Answer 1

更新 从代码中删除 asyncio.gather() 部分，然后调用 create_task

import asyncio
from contextlib import asynccontextmanager

locks = {}
    @asynccontextmanager
    async def get_lock(req_name_):
        logger.info(f"Creating lock for stack {req_name_} if not created")
        if not locks.get(req_name_):
            logger.info("creating key for a lock")
            locks[req_name_] = asyncio.Lock()
        async with locks[req_name_]:
            yield
        if len(locks[req_name_]._waiters) == 0:
            del locks[req_name_]
            logger.info(f"lock released")
        logger.info(len(locks))


    async def handle_lock_request(req_json_):
        logger.info(f"ocupying lock")
        req_name = req_json_.get('req_name')
        async with get_lock(req_name):
            logger.info(f"lock acquired by stack {req_name}")
            await _handle_request(req_json_)
     
     async def _req_handler():
         tasks = []
         loop = asyncio.get_running_loop()
         logger.debug("Await receiver.recv_string")
          req = await receiver.recv_string()
          logger.debug(f"Request received {req}")
          req_json = json.loads(req)
          logger.debug("Await create_task")
          loop.create_task(handle_lock_request(req_json))
          
  
     def _handle_request(req_json_):
       ...
       ...
       logger.info(f"Request finished with req name {req_name} for action patch stack")