优化事件驱动系统中‘小于指定时间的最大事件’查找算法

问题分析 #

当前handle_event函数的性能瓶颈非常明确：每次处理事件时，都需要遍历整个_all_events列表生成past_events，再调用max()筛选符合条件的前序事件。这导致单次操作时间复杂度为O(n)，整体复杂度为O(n²)——对于100万级别的事件量，这种二次复杂度会让处理时间指数级增长，完全无法满足需求。

结合系统给出的4个特性，我们可以针对性地优化索引结构，将整体复杂度降至*O(n log n)*级别，彻底解决性能问题。

方案1：基于bisect模块维护有序索引（纯标准库实现） #

利用Python标准库的bisect模块，维护一个按(occurrence_time, processing_time)升序排列的事件列表。通过二分查找快速定位目标事件位置，避免全量遍历；同时针对70%-90%的有序到达事件做特殊优化，进一步降低平均耗时。

适配Python 3.10+版本（支持bisect的key参数） #

from datetime import datetime
from dataclasses import dataclass
import bisect


@dataclass
class Event:
    event_id: int
    occurrence_time: datetime
    processing_time: datetime

    def __post_init__(self) -> None:
        if self.occurrence_time > self.processing_time:
            raise ValueError("Impossible event! An event can never be processed before it occurs.")


_all_events: list[Event] = []
# 维护按(occurrence_time, processing_time)升序排列的事件列表
_sorted_by_occurrence: list[Event] = []
# 记录当前已处理事件的最大occurrence_time，优化有序到达场景的查找效率
_last_max_occurrence: datetime | None = None


def handle_event(curr_event: Event) -> None:
    global _last_max_occurrence
    prev_event = None

    if _sorted_by_occurrence:
        # 优先处理70%-90%的有序到达事件，O(1)操作
        if _last_max_occurrence <= curr_event.occurrence_time:
            prev_event = _sorted_by_occurrence[-1]
            _sorted_by_occurrence.append(curr_event)
            _last_max_occurrence = curr_event.occurrence_time
        else:
            # 处理晚到事件，通过二分查找定位前序事件，O(log n)查找+O(n)插入
            curr_key = (curr_event.occurrence_time, curr_event.processing_time)
            idx = bisect.bisect_left(
                _sorted_by_occurrence, curr_key,
                key=lambda x: (x.occurrence_time, x.processing_time)
            )
            prev_event = _sorted_by_occurrence[idx-1] if idx > 0 else None
            bisect.insort(
                _sorted_by_occurrence, curr_event,
                key=lambda x: (x.occurrence_time, x.processing_time)
            )
            # 更新最大occurrence_time（如果当前事件的occurrence_time更大）
            if curr_event.occurrence_time > _last_max_occurrence:
                _last_max_occurrence = curr_event.occurrence_time
    else:
        # 处理第一个事件
        _sorted_by_occurrence.append(curr_event)
        _last_max_occurrence = curr_event.occurrence_time

    _all_events.append(curr_event)
    do_something(prev_event, curr_event)

适配Python 3.9及以下版本（无bisect key参数） #

如果使用低版本Python，可将事件包装为元组存储，利用元组的自然排序实现二分查找：

from datetime import datetime
from dataclasses import dataclass
import bisect


@dataclass
class Event:
    event_id: int
    occurrence_time: datetime
    processing_time: datetime

    def __post_init__(self) -> None:
        if self.occurrence_time > self.processing_time:
            raise ValueError("Impossible event! An event can never be processed before it occurs.")


_all_events: list[Event] = []
# 存储格式：(occurrence_time, processing_time, Event)
_sorted_by_occurrence: list[tuple[datetime, datetime, Event]] = []
_last_max_occurrence: datetime | None = None


def handle_event(curr_event: Event) -> None:
    global _last_max_occurrence
    prev_event = None
    curr_tuple = (curr_event.occurrence_time, curr_event.processing_time, curr_event)

    if _sorted_by_occurrence:
        if _last_max_occurrence <= curr_event.occurrence_time:
            prev_event = _sorted_by_occurrence[-1][2]
            _sorted_by_occurrence.append(curr_tuple)
            _last_max_occurrence = curr_event.occurrence_time
        else:
            idx = bisect.bisect_left(_sorted_by_occurrence, curr_tuple)
            prev_event = _sorted_by_occurrence[idx-1][2] if idx > 0 else None
            bisect.insort(_sorted_by_occurrence, curr_tuple)
            if curr_event.occurrence_time > _last_max_occurrence:
                _last_max_occurrence = curr_event.occurrence_time
    else:
        _sorted_by_occurrence.append(curr_tuple)
        _last_max_occurrence = curr_event.occurrence_time

    _all_events.append(curr_event)
    do_something(prev_event, curr_event)

方案2：使用SortedList实现O(log n)插入与查找（第三方库） #

如果允许引入第三方库，sortedcontainers中的SortedList是更优选择——它的插入、查找操作均为*O(log n)*时间复杂度，无需手动维护排序逻辑，对于晚到事件占比较高的场景性能提升更明显。

实现代码 #

from datetime import datetime
from dataclasses import dataclass
from sortedcontainers import SortedList


@dataclass
class Event:
    event_id: int
    occurrence_time: datetime
    processing_time: datetime

    def __post_init__(self) -> None:
        if self.occurrence_time > self.processing_time:
            raise ValueError("Impossible event! An event can never be processed before it occurs.")


_all_events: list[Event] = []
# 按(occurrence_time, processing_time)作为排序key
_sorted_by_occurrence = SortedList(key=lambda x: (x.occurrence_time, x.processing_time))


def handle_event(curr_event: Event) -> None:
    # 二分查找定位当前事件的插入位置，O(log n)
    idx = _sorted_by_occurrence.bisect_left(curr_event)
    # 前序事件为插入位置的前一个元素（如果存在）
    prev_event = _sorted_by_occurrence[idx-1] if idx > 0 else None

    _sorted_by_occurrence.add(curr_event)
    _all_events.append(curr_event)
    do_something(prev_event, curr_event)

优化效果说明 #

• 原方案的O(n²)复杂度降至O(n log n)，100万事件的总操作数从1e12级降至2e7级；
• 方案1中70%-90%的有序事件可实现*O(1)*查找与插入，进一步降低平均耗时；
• 方案2的SortedList可让所有操作稳定在O(log n)，适合晚到事件占比较高的场景；
• 结合do_something()的1ms单次耗时，100万事件的总处理时间可从数天降至十几分钟（主要耗时在do_something()本身）。

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