基于LSH与Minhash的类Omegle应用高效等待机制实现问询

我正在开发一款类似Omegle的陌生人匹配应用，基于用户共同兴趣做匹配。目前用LSH（局部敏感哈希）结合Minhash技术实现匹配逻辑，但在处理「API调用后未立即找到匹配用户」的等待机制时遇到了瓶颈。

当前我用time.sleep()设置等待时长后返回"Failed"状态，但这个函数会阻塞其他API调用，导致其他用户请求延迟。想知道Omegle这类网站是怎么处理这种场景的，以及实现高效等待机制的正确流程。

以下是我的代码片段：

from fastapi import FastAPI, Body
from typing import Annotated
from pydantic import BaseModel
from sonyflake import SonyFlake
import redis
import time
from datasketch import MinHash, MinHashLSH

app = FastAPI()
sf = SonyFlake()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
lsh = MinHashLSH(num_perm=128, threshold=0.5, storage_config={
    'type': 'redis',
    'redis': {'host': '127.0.0.1', 'port': 6379}
}, prepickle=True)


class Partner(BaseModel):
    client_id: int
    partner_id: str
    status: str = 'Failed'


@app.post("/start", response_model=Partner)
async def start(interests: Annotated[list[str] | None, Body()] = None) -> Partner:
    client_id=sf.next_id()
    partner_id = ''

    minhash = MinHash()
    if not interests:
        return Partner(client_id = client_id, partner_id = partner_id)

    client_hash = f"user:{client_id}:interests:hash"
    minhash.update_batch([*(map(lambda item: item.encode('utf-8'), interests))])
    lsh.insert(client_hash, minhash)

    matches = lsh.query(minhash)
    matches.remove(client_hash)

    if not matches:
        time.sleep(5)

    matches = lsh.query(minhash)
    matches.remove(client_hash)

    if not matches:
        lsh.remove(client_hash)
        return Partner(client_id = client_id, partner_id = partner_id)

    lsh.remove(client_hash)
    lsh.remove(matches[0])
    return Partner(client_id = client_id, partner_id = matches[0], status="Success")

希望得到以下帮助：

1. 该场景下实现高效等待机制的见解与最佳实践；
2. 代码优化或提升响应性的建议；
3. 相关学习资源或链接。

一、高效等待机制的最佳实践 #

Omegle这类平台不会让请求线程原地阻塞等待，核心是把等待逻辑从服务端线程转移到客户端或用非阻塞方式处理，常见方案：

• 客户端轮询：用户发起匹配请求后，服务端返回匹配会话ID，客户端每隔1-2秒调用查询接口获取状态，直到找到匹配或超时。
• WebSocket长连接：建立双向连接，用户发起请求后服务端保持连接，找到匹配后实时推送结果，超时则主动通知客户端。
• 队列化匹配请求：将未匹配用户按兴趣分组存入Redis队列，新用户先查对应队列，有等待用户直接匹配，无则加入队列；后台用异步任务定期扫描队列做批量匹配，避免每个请求轮询LSH。
• 绝对避免阻塞线程：time.sleep()会占用FastAPI的ASGI事件循环线程，导致所有请求卡顿，必须用非阻塞替代方案。

二、代码优化建议 #

1. 替换阻塞sleep为异步等待 #

用asyncio.sleep()替代time.sleep()，它不会阻塞事件循环线程：

import asyncio

# ... 其他代码 ...

if not matches:
    # 异步等待，不影响其他请求
    await asyncio.sleep(5)

2. 改用WebSocket实现实时匹配 #

将/start接口改为WebSocket端点，实时推送匹配结果：

from fastapi import WebSocket

@app.websocket("/match")
async def websocket_match(websocket: WebSocket):
    await websocket.accept()
    client_id = sf.next_id()
    interests = await websocket.receive_json()
    
    minhash = MinHash()
    if not interests:
        await websocket.send_json({"client_id": client_id, "partner_id": "", "status": "Failed"})
        await websocket.close()
        return
    
    client_hash = f"user:{client_id}:interests:hash"
    minhash.update_batch([item.encode('utf-8') for item in interests])
    lsh.insert(client_hash, minhash)
    
    # 设置30秒超时
    timeout = 30
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        matches = lsh.query(minhash)
        matches.remove(client_hash)
        if matches:
            partner_hash = matches[0]
            partner_id = partner_hash.split(":")[1]
            # 移除双方匹配记录
            lsh.remove(client_hash)
            lsh.remove(partner_hash)
            # 通知当前用户匹配成功
            await websocket.send_json({
                "client_id": client_id,
                "partner_id": partner_id,
                "status": "Success"
            })
            await websocket.close()
            return
        # 异步等待1秒后再查询，减少LSH访问频率
        await asyncio.sleep(1)
    
    # 超时未匹配
    lsh.remove(client_hash)
    await websocket.send_json({"client_id": client_id, "partner_id": "", "status": "Failed"})
    await websocket.close()

3. 引入异步任务队列处理匹配 #

用Celery+Redis实现异步匹配流程：

1. 用户调用/start接口，服务端生成client_id、计算minhash存入LSH，同时将用户信息加入Redis队列，立即返回「等待匹配」响应。
2. 后台Celery任务定期扫描队列，取出用户做LSH查询，找到匹配后更新Redis中的匹配状态。
3. 客户端通过轮询/check_match/{client_id}接口获取结果。

4. 优化LSH匹配逻辑 #

• 找到匹配或超时后再从LSH中移除用户，避免重复插入。
• 按兴趣标签分组使用LSH（比如给不同兴趣组加前缀），缩小查询范围，提升匹配效率。

三、相关学习资源推荐 #

• FastAPI异步框架：重点学习异步函数、WebSocket使用，理解ASGI事件循环原理。
• Redis数据结构：掌握List、Hash在任务队列、状态存储中的应用。
• Celery异步任务：学习任务队列的基本配置与使用，实现后台批量匹配。
• LSH与Minhash进阶：深入理解算法原理，根据场景调整num_perm、threshold参数。
• WebSocket协议：学习双向通信机制，掌握Python中高效实现实时通信的方法。

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