Redis 应用组件

def add_or_update_contact(conn, user, contact):
    # 添加或更新用户的最近联系人列表
    contact_list = 'recent:' + user
    pipe = conn.pipeline(True)
 
    # 如果联系人已存在，先删除
    pipe.lrem(contact_list, contact)
    # 将联系人添加到列表开头
    pipe.lpush(contact_list, contact)
    # 确保列表不超过100个联系人
    pipe.ltrim(contact_list, 0, 99)
    pipe.execute()

def get_autocomplete_list(conn, user, prefix):
    # 获取匹配前缀的联系人列表
    candidates = conn.lrange('recent:' + user, 0, -1)
    matches = []
    
    for candidate in candidates:
        if candidate.lower().startswith(prefix.lower()):
            matches.append(candidate)
    
    return matches

def find_prefix_range(prefix):
    # 计算前缀匹配的字符范围
    valid_characters = '`abcdefghijklmnopqrstuvwxyz{'
    
    # 找到前缀最后一个字符的位置
    pos = bisect.bisect_left(valid_characters, prefix[-1:])
    # 获取前一个字符作为起始边界
    suffix = valid_characters[(pos or 1) - 1]
    
    return prefix[:-1] + suffix + '{', prefix + '{'

def autocomplete_on_prefix(conn, guild, prefix):
    # 使用 ZSET 实现高效的前缀自动完成
    start, end = find_prefix_range(prefix)
    identifier = str(uuid.uuid4())
    start += identifier
    end += identifier
 
    zset_name = 'members:' + guild
 
    # 添加边界标记
    conn.zadd(zset_name, {start: 0, end: 0})
 
    pipeline = conn.pipeline(True)
    while True:

def acquire_lock(conn, lock_name, acquire_timeout=10):
    # 获取分布式锁
    identifier = str(uuid.uuid4())
    end = time.time() + acquire_timeout
 
    while time.time() < end:
        if conn.setnx('lock:' + lock_name, identifier):
            return identifier
        time.sleep(0.001)
 
    return False

def release_lock(conn, lock_name, identifier):
    # 释放分布式锁
    pipe = conn.pipeline(True)
    lock_name = 'lock:' + lock_name
 
    while True:
        try:
            pipe.watch(lock_name)
            if pipe.get(lock_name) == identifier:
                pipe.multi()
                pipe.delete(lock_name)
                pipe.execute()
                return True
            pipe.unwatch()
            break
        except redis.exceptions.WatchError:

def acquire_lock_with_timeout(conn, lock_name, acquire_timeout=10, lock_timeout=10):
    # 获取带超时机制的分布式锁
    identifier = str(uuid.uuid4())
    lock_timeout = int(math.ceil(lock_timeout))
    end = time.time() + acquire_timeout
 
    while time.time() < end:
        if conn.setnx(lock_name, identifier):
            conn.expire(lock_name, lock_timeout)
            return identifier
        elif not conn.ttl(lock_name):
            conn.expire(lock_name, lock_timeout)

def purchase_item_with_lock(conn, buyer_id, item_id, seller_id):
    # 使用细粒度锁保护物品交易
    buyer = "users:%s" % buyer_id
    seller = "users:%s" % seller_id
    item = "%s.%s" % (item_id, seller_id)
    inventory = "inventory:%s" % buyer_id
 
    end_time = time.time() + 30

def 获取信号量(连接, 信号量名, 限制数量, 超时=10):
    """获取计数信号量"""
    标识符 = str(uuid.uuid4())
    当前时间 = time.time()
    管道 = 连接.pipeline(True)
    
    # 清理过期的信号量持有者
    管道.zremrangebyscore(信号量名, '-inf', 当前时间 - 超时)
    # 尝试获取信号量
    管道.zadd(信号量名, 标识符, 当前时间)
    管道.zrank(信号量名, 标识符)
    
    if 管道.execute()[-1] < 限制数量:

def release_semaphore(conn, semaphore_name, identifier):
    # 释放计数信号量
    return conn.zrem(semaphore_name, identifier)

def acquire_fair_semaphore(conn, semaphore_name, limit, timeout=10):
    # 获取公平的计数信号量
    identifier = str(uuid.uuid4())
    owner_zset = semaphore_name + ':owner'
    counter_key = semaphore_name + ':counter'
    now = time.time()
    
    pipe = conn.pipeline(True)
    # 清理过期的信号量持有者
    pipe.zremrangebyscore(semaphore_name, '-inf', now - timeout)
    # 更新所有者ZSET，只保留有效的持有者

def refresh_fair_semaphore(conn, semaphore_name, identifier):
    # 刷新公平信号量的超时时间
    if conn.zadd(semaphore_name, {identifier: time.time()}):
        # 如果添加成功，说明标识符已经不存在，信号量已丢失
        release_fair_semaphore(conn, semaphore_name, identifier)
        return False
    return True

def enqueue_sold_email_task(conn, seller, item, price, buyer):
    # 添加邮件任务到队列
    data = {
        'seller_id': seller,
        'item_id': item,
        'price': price,
        'buyer_id': buyer,
        'time': time.time()
    }
    conn.rpush('queue:email', json.dumps(data))

def process_sold_email_queue(conn):
    # 处理邮件队列中的任务
    while not exit_flag:
        packed_data = conn.blpop(['queue:email'], 30)
        if not packed_data:
            continue
        
        to_send = json.loads(packed_data[1])
        try:
            fetch_data_and_send_sold_email(to_send)
        except EmailSendError as err:
            log_error("Failed to send sold email", err, to_send)
        else

def process_multi_queue_tasks(conn, queue_list, callbacks):
    # 处理多个优先级队列中的任务
    while not exit_flag:
        packed_data = conn.blpop(queue_list, 30)
        if not packed_data:
            continue
 
        queue_name, task_data = packed_data
        func_name, args = json.loads(task_data)
 
        if func_name not in callbacks:
            log_error("Unknown callback function %s" % func_name)
            continue
 

def enqueue_delayed_task(conn, queue, func_name, args, delay=0):
    # 创建延迟任务
    identifier = str(uuid.uuid4())
    task_data = json.dumps([identifier, queue, func_name, args])
 
    if delay > 0:
        # 延迟执行，添加到ZSET
        conn.zadd('delayed:', {task_data: time.time() + delay})
    else:
        # 立即执行，添加到LIST
        conn.rpush('queue:' + queue, task_data)
 
    return identifier

def poll_delayed_queue(conn):
    # 轮询延迟队列，将到期的任务移动到执行队列
    while not exit_flag:
        item = conn.zrange('delayed:', 0, 0, withscores=True)
        if not item or item[0][1] > time.time():
            time.sleep(0.01)
            continue
 
        task_data = item[0

def send_message(conn, recipient, sender, message):
    # 发送消息给指定接收者
    message_data = {
        'sender': sender,
        'msg': message,
        'ts': time.time()
    }
    conn.rpush('mailbox:' + recipient, json.dumps(message_data))

def fetch_messages(conn, user, count=10):
    # 获取用户邮箱中的消息
    message_list = conn.lrange('mailbox:' + user, 0, count-1)
    if message_list:
        conn.ltrim('mailbox:' + user, count, -1)
    
    return [json.loads(message) for message in message_list]

def create_chat(conn, sender, recipient_list, message, chat_id=None):
    # 创建新的群聊会话
    chat_id = chat_id or str(conn.incr('ids:chat:'))
    recipient_list.append(sender)
    recipient_dict = dict((recipient, 0) for recipient in recipient_list)
    
    conn.zadd('chat:' + chat_id, **recipient_dict)
    for recipient in recipient_list:
        conn.zadd(

def send_group_message(conn, chat_id, sender, message):
    # 在群聊中发送消息
    token = acquire_lock(conn, 'chat:' + chat_id)
    if not token:
        raise Exception("无法获取锁")
    
    try:
        msg_id = conn.incr('ids:' + chat_id)
        timestamp = time.time()
        packed_msg = json.dumps({
            'id': msg_id,
            'ts': timestamp,

def get_pending_messages(conn, recipient):
    # 获取用户在所有群聊中的待处理消息
    seen_info = conn.zrange('seen:' + recipient, 0, -1, withscores=True)
    pipe = conn.pipeline(True)
    
    for chat_id, seen_id in seen_info:
        pipe.zrangebyscore('msgs:' + chat_id, seen_id + 1, 'inf')
    

def daily_country_aggregate(conn, log_line):
    # 聚合每日的国家级访问数据
    if log_line:
        fields = log_line.split()
        ip_addr = fields[0]
        date = fields[1]
        country = lookup_city_by_ip(ip_addr)[2]
        aggregate_data[date][country] += 1
        return
 
    # 处理完一天的数据，写入Redis
    for date, aggregate in aggregate_data.items():
        conn.zadd('daily:country:' +

def copy_logs_to_redis(conn, path, channel, num_workers=10, limit=2**30):
    # 将日志文件复制到Redis并通知处理节点
    bytes_in_redis = 0
    wait_queue = deque()
    create_group_chat(conn, 'source', map(str, range(num_workers)), '', channel)
    num_workers = str(num_workers)
    
    for log_file in sorted(os.listdir(path)):

def process_logs_from_redis(conn, node_id, callback):
    # 从Redis中读取并处理日志文件
    while True:
        file_data = get_pending_messages(conn, node_id)
        for channel, messages in file_data:
            for msg in messages:
                log_file = msg['message']
                if log_file == ':done':
                    return
                elif not log_file:
                    continue
 
                chunk_reader =

def read_lines(conn, key, chunk_reader):
    # 从Redis中逐行读取文件内容
    buffer = ''
    for chunk in chunk_reader(conn, key):
        buffer += chunk
        pos = buffer.rfind('\n')
        if pos >= 0:
            for line in buffer[:pos].split('\n'):
                yield line + '\n

def read_chunks_gz(conn, key):
    # 从Redis中逐块读取并解压缩gzip文件
    buffer = ''
    decompressor = None
 
    for chunk in read_chunks(conn, key, 2**17):
        if not decompressor:
            buffer += chunk
            try:
                # 解析gzip头部
                if buffer[:3] != "\x1f\x8b\x08":
                    raise