连续内存布局：用array和numpy实现高速批量读写

说实话，很多做Python性能优化的朋友，一开始都容易忽略内存布局这件事。我当年也是踩过坑的——有一次处理上亿条传感器数据，用Python列表读写，结果程序跑了整整一个下午还没跑完。后来换成numpy数组，同样的数据量，十几秒就搞定了。差别在哪？说白了，就是内存布局。

为什么连续内存这么快？

你想想看，CPU从内存里读数据，不是一个个字节读的，而是一块块读的。这一块通常叫“缓存行”，大小一般是64字节。如果你的数据在内存里是连续排列的，CPU一次缓存行读取就能拿到好几个数据。反之，如果数据分散在各处，CPU就得频繁去内存里“取货”，那速度自然就慢了。

我习惯把连续内存比作“书架上的书”——按顺序排好，拿一本翻一本，效率极高。而Python列表里存的是对象的引用，每个对象散落在内存各处，就像书被扔在房间各个角落，找起来费劲得很。

array模块：轻量级的连续内存容器

Python标准库里的array模块，其实是个被很多人忽略的宝贝。它存储的是真正的数值，而不是Python对象。每个元素在内存里紧挨着排好，类型统一。

来看个例子：

from array import array
import time

# 用列表创建1000万个整数
start = time.perf_counter()
list_data = list(range(10_000_000))
print(f"列表创建耗时: {time.perf_counter() - start:.3f}秒")

# 用array创建同样的数据
start = time.perf_counter()
arr_data = array('i', range(10_000_000))  # 'i' 表示有符号整型
print(f"array创建耗时: {time.perf_counter() - start:.3f}秒")

# 批量读取测试
start = time.perf_counter()
sum_list = sum(list_data)
print(f"列表求和耗时: {time.perf_counter() - start:.3f}秒")

start = time.perf_counter()
sum_arr = sum(arr_data)
print(f"array求和耗时: {time.perf_counter() - start:.3f}秒")

我在项目中实测过，array的创建速度比列表快2-3倍，批量读取更是快上5-10倍。为什么？因为array不需要处理Python对象的引用计数，数据在内存里就是赤裸裸的二进制。

容器类型	内存布局	元素类型	批量读取速度
list	非连续（存对象引用）	任意Python对象	慢
array	连续（存原始值）	单一C类型	快
numpy.ndarray	连续（存原始值）	单一类型（支持广播）	极快

numpy数组：工业级连续内存方案

如果说array是轻量级选手，那numpy就是重型武器了。numpy数组不仅内存连续，还支持向量化操作——说白了，就是一次操作整个数组，不用写循环。

我记得有一次做金融高频数据回测，需要计算百万级K线的移动平均。用Python列表写for循环，一次回测要跑40分钟。换成numpy后，同样的逻辑，3秒搞定。嗯，你没看错，是3秒。

import numpy as np
import time

# 创建1000万个随机浮点数
n = 10_000_000
data = np.random.randn(n)

# 向量化计算：所有元素乘以2再加1
start = time.perf_counter()
result = data * 2 + 1
print(f"numpy向量化耗时: {time.perf_counter() - start:.3f}秒")

# 对比：用Python列表做同样的操作
list_data = data.tolist()
start = time.perf_counter()
list_result = [x * 2 + 1 for x in list_data]
print(f"列表推导式耗时: {time.perf_counter() - start:.3f}秒")

你可能会问：numpy为什么这么快？除了连续内存，还有两个关键点：

• C语言底层实现：numpy的核心计算是用C写的，没有Python解释器的开销
• SIMD指令优化：现代CPU支持单指令多数据流，numpy能自动利用这些指令

实战：高速批量数据写入

批量写入的场景，我遇到过最典型的就是日志系统和时序数据库。假设我们要把100万条记录写入文件：

import numpy as np
import struct
import time

# 生成测试数据
n = 1_000_000
timestamps = np.arange(n, dtype=np.int64)
values = np.random.randn(n).astype(np.float64)

# 方法1：逐个写入（慢）
start = time.perf_counter()
with open('data_slow.bin', 'wb') as f:
    for i in range(n):
        f.write(struct.pack('qd', timestamps[i], values[i]))
print(f"逐个写入耗时: {time.perf_counter() - start:.3f}秒")

# 方法2：批量写入（快）
start = time.perf_counter()
# 将两个数组合并成结构化数组
combined = np.column_stack((timestamps, values))
with open('data_fast.bin', 'wb') as f:
    f.write(combined.tobytes())
print(f"批量写入耗时: {time.perf_counter() - start:.3f}秒")

我曾经帮一个团队优化过他们的数据采集系统。原来他们用Python的pickle逐条序列化写入，每秒只能写几千条。改成numpy的tobytes()批量写入后，每秒能写几十万条。差别就是这么大。

读取性能对比：眼见为实

写完了还得读回来，对吧？来看看不同方式的读取性能：

# 方法1：逐个读取
start = time.perf_counter()
timestamps_read = np.empty(n, dtype=np.int64)
values_read = np.empty(n, dtype=np.float64)
with open('data_fast.bin', 'rb') as f:
    for i in range(n):
        data = f.read(16)  # 8字节int64 + 8字节float64
        timestamps_read[i], values_read[i] = struct.unpack('qd', data)
print(f"逐个读取耗时: {time.perf_counter() - start:.3f}秒")

# 方法2：批量读取
start = time.perf_counter()
with open('data_fast.bin', 'rb') as f:
    raw_data = f.read()
data_array = np.frombuffer(raw_data, dtype=np.dtype([('ts', np.int64), ('val', np.float64)]))
print(f"批量读取耗时: {time.perf_counter() - start:.3f}秒")

实测下来，批量读取比逐个读取快50倍以上。原因很简单：批量读取只做一次系统调用，而逐个读取要做100万次系统调用。系统调用是有开销的，每次都要切换用户态和内核态。

知识体系总览

下面这张图，是我梳理的连续内存布局优化核心逻辑：

最后说一句，连续内存布局这个思路，不光适用于Python。我在写C++程序时，也会刻意把热点数据放在连续的内存块里，用std::vector而不是std::list。道理是一样的——让数据排好队，CPU才能跑得更快。

---

公众号:蓝海资料掘金营，微信deep3321