Redis 實作進階任務佇列與訊息系統

def worker_watch_queues(conn, queues, callbacks):
    while not QUIT:
        packed = conn.blpop(queues, 30)
        if not packed:
            continue
        name, args = json.loads(packed[1])
        if name not in callbacks:
            log_error("Unknown callback %s" % name)
            continue
        callbacks[name](*args)

queues = ['high-priority', 'medium-priority', 'low-priority']
worker_watch_queues(conn, queues, callbacks)

def execute_later(conn, queue, name, args, delay=0):
    identifier = str(uuid.uuid4())
    item = json.dumps([identifier, queue, name, args])
    if delay > 0:
        conn.zadd('delayed:', {item: time.time() + delay})
    else:
        conn.rpush('queue:' + queue, item)
    return identifier

def poll_queue(conn):
    while not QUIT:
        # 取得最早的延遲任務
        item = conn.zrange('delayed:', 0, 0, withscores=True)
        
        # 如果沒有任務或任務尚未到期，稍後重試
        if not item or item[0][1] > time.time():
            time.sleep(0.01)
            continue
        
        # 解析任務內容
        item = item[0][0]
        identifier, queue, function, args = json.loads(item)
        
        # 取得鎖以避免重複執行
        locked = acquire_lock(conn, identifier)
        if not locked:
            continue
        
        # 將任務從延遲佇列移動到目標佇列
        if conn.zrem('delayed:', item):
            conn.rpush('queue:' + queue, item)
        
        # 釋放鎖
        release_lock(conn, identifier, locked)

# 示例佇列順序
queues = ["high-delayed", "high", "medium-delayed", "medium", "low-delayed", "low"]

def send_message(conn, recipient, message):
    conn.rpush(f'mailbox:{recipient}', message)

def fetch_messages(conn, recipient):
    messages = conn.lrange(f'mailbox:{recipient}', 0, 9)
    conn.ltrim(f'mailbox:{recipient}', 10, -1)
    return messages

def create_chat(conn, sender, recipients, message, chat_id=None):
    chat_id = chat_id or str(conn.incr('ids:chat:'))
    recipients.append(sender)
    recipients_dict = dict((r, 0) for r in recipients)
    
    # 將群組成員加入 ZSET
    conn.zadd(f'chat:{chat_id}', recipients_dict)
    
    # 將群組 ID 加入使用者的群組列表
    for recipient in recipients:
        conn.zadd(f'seen:{recipient}', {chat_id: 0})
    
    # 傳送初始訊息
    conn.zadd(f'chat:{chat_id}:messages', {message: conn.incr('ids:message:')})
    
    return chat_id

def create_chat(conn, sender, recipients, message, chat_id=None):
    chat_id = chat_id or str(conn.incr('chat:id:'))
    recipients.append(sender)
    
    pipeline = conn.pipeline(True)
    pipeline.sadd('chat:' + chat_id + ':users', *recipients)
    for recipient in recipients:
        pipeline.zadd('seen:' + recipient, {chat_id: 0})
    pipeline.execute()
    
    return send_message(conn, chat_id, sender, message)

#### 內容解密：
此函式用於建立新的聊天室。首先，它檢查是否提供了 chat_id，如果沒有，則使用 Redis 的 INCR 命令生成一個新的唯一聊天室 ID。然後，它將所有參與者（包括傳送者）新增到聊天室的使用者集合中，並為每個參與者初始化一個已讀訊息的 ZSET，初始分數為 0。最後，呼叫 send_message 函式傳送初始訊息。

def send_message(conn, chat_id, sender, message):
    identifier = acquire_lock(conn, 'chat:' + chat_id)
    if not identifier:
        raise Exception("無法取得鎖")
    try:
        mid = conn.incr('ids:' + chat_id)
        ts = time.time()
        packed = json.dumps({
            'id': mid,
            'ts': ts,
            'sender': sender,
            'message': message,
        })
        conn.zadd('msgs:' + chat_id, {packed: mid})
    finally:
        release_lock(conn, 'chat:' + chat_id, identifier)
    return chat_id

#### 內容解密：
此函式負責在指定的聊天室中傳送訊息。首先，它嘗試取得鎖以防止並發修改。然後，它使用 INCR 命令生成一個新的訊息 ID，將訊息內容封裝成 JSON 格式，並將其新增到聊天室的訊息 ZSET 中。最後，釋放鎖以允許其他操作。

def fetch_pending_messages(conn, recipient):
    seen = conn.zrange('seen:' + recipient, 0, -1, withscores=True)
    pipeline = conn.pipeline(True)
    for chat_id, seen_id in seen:
        pipeline.zrangebyscore('msgs:' + chat_id, seen_id+1, 'inf')
    chat_info = zip(seen, pipeline.execute())
    
    for i, ((chat_id, seen_id), messages) in enumerate(chat_info):
        if not messages:
            continue
        messages[:] = map(json.loads, messages)
        seen_id = messages[-1]['id']
        conn.zadd('chat:' + chat_id, {recipient: seen_id})
        min_id = conn.zrange('chat:' + chat_id, 0, 0, withscores=True)
        pipeline.zadd('seen:' + recipient, {chat_id: seen_id})
        if min_id:
            pipeline.zremrangebyscore('msgs:' + chat_id, 0, min_id[0][1])
        chat_info[i] = (chat_id, messages)
    pipeline.execute()
    return chat_info

#### 內容解密：
此函式用於取得指定接收者的待處理訊息。首先，它取得接收者已讀訊息的最新 ID。然後，它遍歷所有相關的聊天室，取得每個聊天室中接收者尚未閱讀的訊息。對於每個聊天室，它更新接收者的已讀訊息 ID，並清理掉所有使用者都已閱讀的舊訊息。最後，它傳回包含聊天室資訊和訊息內容的列表。

def join_chat(conn, chat_id, user):
    message_id = int(conn.get('ids:' + chat_id))
    pipeline = conn.pipeline(True)
    pipeline.zadd('chat:' + chat_id, {user: message_id})
    pipeline.zadd('seen:' + user, {chat_id: message_id})
    pipeline.execute()

#### 內容解密：
此函式允許使用者加入指定的聊天室。它首先取得當前聊天室的最新訊息 ID。然後，它將使用者新增到聊天室的成員列表中，並將聊天室新增到使用者的已讀列表中，初始已讀訊息 ID 為最新訊息 ID。

def leave_chat(conn, chat_id, user):
    pipeline = conn.pipeline(True)
    pipeline.zrem('chat:' + chat_id, user)
    pipeline.zrem('seen:' + user, chat_id)
    pipeline.zcard('chat:' + chat_id)
    count = pipeline.execute()[-1]
    if not count:
        pipeline.delete('msgs:' + chat_id)
        pipeline.delete('ids:' + chat_id)
        pipeline.execute()
    else:
        oldest = conn.zrange('chat:' + chat_id, 0, 0, withscores=True)
        conn.zremrangebyscore('msgs:' + chat_id, 0, oldest[0][1])

#### 內容解密：
此函式允許使用者離開指定的聊天室。它首先從聊天室的成員列表和使用者的已讀列表中移除使用者。如果聊天室中不再有任何成員，它將刪除聊天室的所有相關資料。否則，它將清理掉所有成員都已閱讀的舊訊息。