如何高效合并百万级跨序列重叠的大IP地址范围？

核心问题分析 #

你当前的性能瓶颈主要来自两个方面：一是Python列表pop(0)的O(n)时间复杂度（百万级数据下会导致频繁内存移位）；二是Python对象的交互开销（原生tuple、列表操作在循环中会产生大量额外消耗）。以下是针对性的优化方案，覆盖Numba、Cython及基础逻辑优化，同时支持IPv6的uint128类型。

1. 基础Python逻辑优化（零依赖提效） #

先修正最影响性能的基础操作，不用扩展就能大幅提速：

• 替换list.pop(0)为双指针：两组列表都是已排序的，无需弹出元素，用两个指针i、j分别跟踪两组当前处理项，完全避免O(n)的列表修改开销。
• 用collections.deque替代列表：如果必须弹出首项，deque.popleft()是O(1)操作，比列表快100倍以上。
• 预分配输出容器：提前预估结果大小（比如len(list1)+len(list2)），用[None]*(预估大小)初始化输出列表，再通过索引填充，减少频繁append导致的内存扩容。

2. Numba优化（正确利用JIT加速） #

之前Numba无效的原因是依赖Python对象操作，需改用NumPy数组+原生类型逻辑：

处理uint128（IPv6）的方法 #

Numba不直接支持uint128，可拆分为两个uint64（高64位+低64位）模拟，实现自定义比较函数：

import numba as nb
import numpy as np

# IPv6地址拆分为高/低64位的比较函数
@nb.njit(nb.boolean(nb.uint64, nb.uint64, nb.uint64, nb.uint64))
def ipv6_gt(high1, low1, high2, low2):
    if high1 > high2:
        return True
    elif high1 == high2:
        return low1 > low2
    return False

@nb.njit(nb.boolean(nb.uint64, nb.uint64, nb.uint64, nb.uint64))
def ipv6_lt(high1, low1, high2, low2):
    if high1 < high2:
        return True
    elif high1 == high2:
        return low1 < low2
    return False

核心处理函数（双指针+NumPy数组） #

将三元组转换为结构化数组，用Numba编译纯函数处理：

# 定义IPv4/IPv6的结构化数组类型
ipv4_dtype = np.dtype([('start', 'u4'), ('end', 'u4'), ('data', 'O')])
ipv6_dtype = np.dtype([
    ('start_high', 'u8'), ('start_low', 'u8'),
    ('end_high', 'u8'), ('end_low', 'u8'),
    ('data', 'O')
])

@nb.njit
def process_ipv4_ranges(arr1, arr2):
    n1, n2 = arr1.shape[0], arr2.shape[0]
    i = j = 0
    result = []
    while i < n1 and j < n2:
        s1, e1, d1 = arr1[i]['start'], arr1[i]['end'], arr1[i]['data']
        s2, e2, d2 = arr2[j]['start'], arr2[j]['end'], arr2[j]['data']
        
        if e1 < s2:
            result.append((s1, e1, d1, None))
            i += 1
        elif e2 < s1:
            result.append((s2, e2, None, d2))
            j += 1
        else:
            # 拆分重叠与非重叠部分
            overlap_start = max(s1, s2)
            overlap_end = min(e1, e2)
            result.append((overlap_start, overlap_end, d1, d2))
            
            # 更新剩余未处理范围
            if e1 > overlap_end:
                arr1[i]['start'] = overlap_end + 1
            else:
                i += 1
            if e2 > overlap_end:
                arr2[j]['start'] = overlap_end + 1
            else:
                j += 1
    # 处理剩余项
    while i < n1:
        s, e, d = arr1[i]['start'], arr1[i]['end'], arr1[i]['data']
        result.append((s, e, d, None))
        i += 1
    while j < n2:
        s, e, d = arr2[j]['start'], arr2[j]['end'], arr2[j]['data']
        result.append((s, e, None, d))
        j += 1
    return result

使用时将列表转换为数组：

arr1 = np.array(ipv4_list1, dtype=ipv4_dtype)
arr2 = np.array(ipv4_list2, dtype=ipv4_dtype)
processed = process_ipv4_ranges(arr1, arr2)

3. Cython优化（极致性能，原生支持uint128） #

Cython可直接使用GCC/Clang支持的unsigned __int128类型，完全规避Python对象开销：

核心Cython代码片段 #

# distutils: language = c
# distutils: extra_compile_args = -std=c11

cdef struct RangeIPv4:
    unsigned int start
    unsigned int end
    object data

cdef struct RangeIPv6:
    unsigned __int128 start
    unsigned __int128 end
    object data

cdef list process_ipv6_ranges(list ranges1, list ranges2):
    cdef int n1 = len(ranges1), n2 = len(ranges2)
    cdef RangeIPv6* arr1 = <RangeIPv6*>malloc(n1 * sizeof(RangeIPv6))
    cdef RangeIPv6* arr2 = <RangeIPv6*>malloc(n2 * sizeof(RangeIPv6))
    
    # 填充C数组（将Python元组转换为原生类型）
    for i in range(n1):
        arr1[i].start = ranges1[i][0]
        arr1[i].end = ranges1[i][1]
        arr1[i].data = ranges1[i][2]
    for j in range(n2):
        arr2[j].start = ranges2[j][0]
        arr2[j].end = ranges2[j][1]
        arr2[j].data = ranges2[j][2]
    
    cdef int i = 0, j = 0
    cdef list result = []
    cdef unsigned __int128 overlap_start, overlap_end
    
    while i < n1 and j < n2:
        if arr1[i].end < arr2[j].start:
            result.append((arr1[i].start, arr1[i].end, arr1[i].data, None))
            i += 1
        elif arr2[j].end < arr1[i].start:
            result.append((arr2[j].start, arr2[j].end, None, arr2[j].data))
            j += 1
        else:
            overlap_start = arr1[i].start if arr1[i].start > arr2[j].start else arr2[j].start
            overlap_end = arr1[i].end if arr1[i].end < arr2[j].end else arr2[j].end
            result.append((overlap_start, overlap_end, arr1[i].data, arr2[j].data))
            
            if arr1[i].end > overlap_end:
                arr1[i].start = overlap_end + 1
            else:
                i += 1
            if arr2[j].end > overlap_end:
                arr2[j].start = overlap_end + 1
            else:
                j += 1
    
    # 处理剩余项
    while i < n1:
        result.append((arr1[i].start, arr1[i].end, arr1[i].data, None))
        i += 1
    while j < n2:
        result.append((arr2[j].start, arr2[j].end, None, arr2[j].data))
        j += 1
    
    free(arr1)
    free(arr2)
    return result

编译时需指定支持__int128的编译器（如GCC 5+）。

4. 合并排序法的改进（解决全量数据处理问题） #

原合并排序法速度快但无法处理全量数据，大概率是内存不足导致，可采用分块处理：

1. 分块拆分：将两组数据拆分为多个内存可承受的块（如每块100万条）。
2. 块内处理：对每个块执行合并排序+重叠拆分，输出到临时文件。
3. 归并合并：对所有临时文件做外部归并排序，最终合并为完整结果。

内容的提问来源于stack exchange，提问作者Ξένη Γήινος