如何实现带消息优先级检查的字符串Priority Queue及优先级判定方法？

核心方案：用优先级队列（堆）实现 #

直接基于优先级队列（底层用堆实现）就能满足需求，核心是给每个消息生成一个排序键，通过键的大小控制送达顺序——键值越小，消息优先级越高。

消息优先级判断规则 #

明确送达优先级从高到低：

1. 高优先级消息：无视接收时间，优先送达
2. 已到指定送达时间的定时消息：优先级高于普通消息
3. 普通消息：按接收时间戳先到先送

对应排序键生成逻辑：

• 高优先级消息：键 = (0, 接收时间戳)
• 已到时间的定时消息：键 = (1, 定时送达时间)
• 普通消息：键 = (2, 接收时间戳)
• 未到时间的定时消息：暂不处理，留到下次上线时再判断时间

代码实现（Python） #

用heapq模块实现小顶堆，模拟离线收消息、按键触发上线的完整流程：

import heapq
import time
from datetime import datetime

class Message:
    def __init__(self, content, receive_time, priority=0, scheduled_time=None):
        self.content = content
        self.receive_time = receive_time  # 接收时间戳
        self.priority = priority  # 0=普通，1=高优先级
        self.scheduled_time = scheduled_time  # 定时送达时间戳，None为普通消息

    def __repr__(self):
        recv_time = datetime.fromtimestamp(self.receive_time).strftime("%Y-%m-%d %H:%M:%S")
        if self.scheduled_time:
            sch_time = datetime.fromtimestamp(self.scheduled_time).strftime("%Y-%m-%d %H:%M:%S")
            return f"[定时消息] 内容: {self.content} | 接收时间: {recv_time} | 送达时间: {sch_time}"
        elif self.priority == 1:
            return f"[高优先级] 内容: {self.content} | 接收时间: {recv_time}"
        else:
            return f"[普通消息] 内容: {self.content} | 接收时间: {recv_time}"

def simulate_messenger():
    message_heap = []
    print("=== 用户已离线，开始接收消息 ===")

    # 模拟离线时接收各类消息
    msg1 = Message("今天下午开部门会", time.time())
    heapq.heappush(message_heap, (2, msg1.receive_time, msg1))
    
    msg2 = Message("紧急：服务器宕机了", time.time(), priority=1)
    heapq.heappush(message_heap, (0, msg2.receive_time, msg2))
    
    scheduled_ts = time.time() + 5
    msg3 = Message("记得明天早上9点提交周报", time.time(), scheduled_time=scheduled_ts)
    heapq.heappush(message_heap, (1, scheduled_ts, msg3))
    
    msg4 = Message("下班一起去吃火锅", time.time())
    heapq.heappush(message_heap, (2, msg4.receive_time, msg4))

    print("\n当前缓冲区消息：")
    for item in message_heap:
        print(item[2])

    # 按键触发上线
    input("\n按回车键触发用户上线...")
    print("\n=== 用户已上线，开始按规则送达消息 ===")

    current_time = time.time()
    temp_heap = []

    # 处理所有符合条件的消息
    while message_heap:
        priority_key, sort_key, msg = heapq.heappop(message_heap)
        # 过滤未到时间的定时消息，放回临时堆
        if msg.scheduled_time and msg.scheduled_time > current_time:
            heapq.heappush(temp_heap, (priority_key, sort_key, msg))
            continue
        # 送达并展示消息
        print(f"✅ 已送达：{msg}")

    # 把未到时间的定时消息放回原堆，留到下次上线处理
    while temp_heap:
        heapq.heappush(message_heap, heapq.heappop(temp_heap))

    print("\n=== 本次上线消息处理完成，剩余定时消息将在下一次上线时处理 ===")

if __name__ == "__main__":
    simulate_messenger()

关键逻辑说明 #

1. 排序键设计：用元组(优先级组, 排序子键)作为堆元素的键，元组比较会先对比第一个元素，再对比第二个，完美匹配我们的优先级规则。
2. 定时消息处理：上线时先判断当前时间是否满足定时要求，未到时间的消息暂时放回堆中，等待下次上线再处理。
3. 交互模拟：离线阶段直接往堆中添加消息并展示缓冲区；用input()模拟按键触发上线的操作。

扩展优化方向 #

• 给定时消息单独维护延迟队列，避免每次上线遍历整个堆
• 支持离线时持续接收新消息（比如加个循环接收输入）
• 给高优先级消息细分等级（如1=紧急、2=重要），只需调整排序键的第一个元素即可

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