内存映射文件如何写 (内存映射后怎么取数)

1. 现象

内存映射一个二进制文件到一个可变字节数组中,目的可能是为了随机访问它的内容或者是原地做些修改

2. 原因分析

可以随机访问文件的内容,使得访问文件内容更加高效和优雅

3. 问题解决

使用mmap模块来内存映射文件。

内存映射一个文件并不会导致这个文件被读取到内存中。文件并没有被复制到内存缓存或数组中。

import mmap


def memory_map(file_name, length=0, access=None):
    fd = os.open(file_name, os.O_RDWR)
    return mmap.mmap(fd, length, access)

size = 10
with open('sorted_file_5', 'wb') as f_obj:
    f_obj.seek(size - 1)
    f_obj.write(b'A')

m = memory_map('sorted_file_5')
print(len(m)) # 10
print(m[0:5]) # b'\x00\x00\x00\x00\x00'
print(m[0]) # 0
m[0:5] = b'Hello'
m.close()

with open('sorted_file_5') as f_obj:
    print(f_obj.read(10)) # 'Hello A'

mmap() 返回的 mmap 对象同样也可以作为一个上下文管理器来使用,这时候底层的文件会被自动关闭

with memory_map('sorted_file_5') as m:
    print(len(m)) # 10
    print(m[0:5]) # b'\x00\x00\x00\x00\x00'
    print(m[0]) # 0
    m[0:5] = b'Hello'

默认情况下, memeory map函数打开的文件同时支持读和写操作。任何的修改内容都会复制回原来的文件中。如果需要只读的访问模式,可以给参数 access 赋值为mmap.ACCESS_READ

memory_map(file_name, length=0, access=mmap.ACCESS_READ)

如果想在本地修改数据,但是又不想将修改写回到原始文件中,可以使用mmap.ACCESS COPY

import mmap


def memory_map(file_name, length=0, access=None):
    fd = os.open(file_name, os.O_RDWR)
    return mmap.mmap(fd, length, access=access)


size = 10
with open('sorted_file_5', 'wb') as f_obj:
    f_obj.seek(size - 1)
    f_obj.write(b'A')

from mmap import ACCESS_COPY

# access 换成mmap.ACCESS_COPY
with memory_map('sorted_file_5', access=ACCESS_COPY) as m:
    print(len(m)) # 10
    print(m[0:5]) # b'\x00\x00\x00\x00\x00'
    print(m[0]) # 0
    m[0:5] = b'Hello'
    print(m[0:5]) # b'Hello'

with open('sorted_file_5') as f_obj:
    print(f_obj.read(10)) # ' A'

为了随机访问文件的内容,使用 mmap 将文件映射到内存中是一个高效和优雅的方法。无需打开一个文件并执行大量的 seek() , read() , write() 调用,只需要简单的映射文件并使用切片操作访问数据即可。

一般来讲, mmap方法所暴露的内存看上去就是一个二进制数组对象。但是,你可以使用一个内存视图来解析其中的数据。

size = 8

with open('sorted_file_5', 'wb') as f_obj:
    f_obj.seek(size - 1)
    f_obj.write(b'A')

    f_obj.seek(0)
    f_obj.write(b'Hello')

m = memory_map('sorted_file_5', access=mmap.ACCESS_COPY)
print(len(m)) # '8'
# I 表示转换为无符号整数 unsigned int 占用四个字节
v = memoryview(m).cast('I')
v[0] = 8
print(m[0:4]) # b'\x08\x00\x00\x00'
m[0:4] = b'\x07\x01\x00\x00'
# byte order 字节顺序是 'little'
print(v[0]) # '263'
print(v[1]) # '1090519151'

查询本地电脑字节顺序

import sys
print(sys.byteorder) # 'little'

需要强调的一点是,内存映射一个文件并不会导致整个文件被读取到内存中。也就是说,文件并没有被复制到内存缓存或数组中相反,操作系统仅仅为文件内容保留了一段虚拟内存。当访问文件的不同区域时,这些区域的内容才根据需要被读取并映射到内存区域中。而那些从没被访问到的部分还是留在磁盘上。所有这些过程是透明的,在幕后完成

如果多个 Python 解释器内存映射同一个文件,得到的 mmap 对象能够被用来在解释器直接交换数据。也就是说,所有解释器都能同时读写数据,并且其中一个解释器所做的修改会自动呈现在其他解释器中。很明显,这里需要考虑同步的问题。但是这种方法有时候可以用来在管道或套接字间传递数据。

4. 错误经历

4.1 nonzero cur-relative seeks

关键语句:io.UnsupportedOperation: can't do nonzero cur-relative seeks

内存映射文件的方法,内存映射怎么实现

相关代码:

with open('sorted_file_0', 'wt', encoding='utf-8') as f_obj:
    f_obj.write('第一行\n')
    # 一个汉字 占用 3个字节
    f_obj.write('ABC\n')
    f_obj.write('第三行\n')
    f_obj.write('DEF\n')

with open('sorted_file_0', 'rt', encoding='utf-8') as f_obj:
    print('当前位置:{}'.format(f_obj.tell()))
    # 换行符 \n 占用 2两个字节
    content = f_obj.read()
    print('当前位置:{}'.format(f_obj.tell()))
    f_obj.seek(0)
    print('当前位置:{}'.format(f_obj.tell()))
    f_obj.seek(6)
    data_1 = f_obj.readline() # '行\n'
    print('当前位置:{}'.format(f_obj.tell()))

    # 从当前位置SEEK_CUR 1 偏移 3个字节 byte
    # SEEK_CUR 为 1 时,Offset必须为负数
    f_obj.seek(-2, 1)
    data_2 = f_obj.readline() # '一行\n'
    print('当前位置:{}'.format(f_obj.tell())) # '11'

错误原因:Python 3.2 中解释说明,文本模式查找seek只允许从文件开头查找,其他方式不允许,二进制模式无此限制

with open('sorted_file_0', 'wb') as f_obj:
    f_obj.write('第一行\n'.encode('utf-8'))
    # 一个汉字 占用 3个字节
    f_obj.write(b'ABC\n')
    # 二进制 \n 一个字节
    f_obj.write('第三行\n'.encode('utf-8'))
    f_obj.write(b'DEF\n')


with open('sorted_file_0', 'rb') as f_obj:
    print('当前位置:{}'.format(f_obj.tell()))
    # 换行符 \n 占用 1 个字节
    content = f_obj.read() # 'b'\xe7\xac\xac\xe4\xb8\x80\xe8\xa1\x8c\nABC\n\xe7\xac\xac\xe4\xb8\x89\xe8\xa1\x8c\nDEF\n''
    print('当前位置:{}'.format(f_obj.tell())) # '28'
    f_obj.seek(0)
    print('当前位置:{}'.format(f_obj.tell())) # '0'
    f_obj.seek(6)
    data_1 = f_obj.readline() # '行\n'
    print('当前位置:{}'.format(f_obj.tell())) # '10'