使用二进制数据记录
struct 模块提供了 pack() 和 unpack() 函数,用于处理可变长度的二进制记录格式。下面的示例演示如何在不使用 zipfile 模块的情况下循环遍历 ZIP 文件中的头信息。封装码 “H” 和 “I” 分别表示2字节和4字节的无符号数。“<” 表示它们是标准大小,并且按低字节序顺序排列:
准备3个文件,压缩到一个文件中,文件名命名为:myfile.zip
import struct
with open('myfile.zip', 'rb') as f:
data = f.read()
start = 0
for i in range(3): # 显示前3个文件头
start += 14
fields = struct.unpack('<IIIHH', data[start:start+16])
crc32, comp_size, uncomp_size, filenamesize, extra_size = fields
start += 16
filename = data[start:start+filenamesize]
start += filenamesize
extra = data[start:start+extra_size]
print(filename, hex(crc32), comp_size, uncomp_size)
start += extra_size + comp_size
复制
运行结果:
b'file1.txt' 0x1ed605cd 99 94
b'file2.txt' 0x8517a1ea 95 90
b'file3.txt' 0xc34c995f 95 90
复制
多线程
线程是一种用于解耦不依赖于顺序的任务的技术。当其他任务在后台运行时,线程可以用来提高接受用户输入的应用程序的响应能力。一个相关的用例是在一个线程中计算时如何在另一个线程中并行运行I/O处理。
以下代码显示了在主程序持续运行时,高级别 threading 模块如何在后台运行任务:
import threading, zipfile
class AsyncZip(threading.Thread):
def __init__(self, infile, outfile):
threading.Thread.__init__(self)
self.infile = infile
self.outfile = outfile
def run(self):
f = zipfile.ZipFile(self.outfile, 'w', zipfile.ZIP_DEFLATED)
f.write(self.infile)
f.close()
print('后台完成压缩:', self.infile)
background = AsyncZip('file1.txt', 'myarchive.zip')
background.start()
print('主程序持续在前台运行。')
background.join() # 等待后台任务结束
print('主程序一直等到后台任务结束')
复制
运行结果:
主程序持续在前台运行。
后台完成压缩: file1.txt
主程序一直等到后台任务结束
复制
多线程应用程序的主要挑战是协调共享数据或其他资源的线程。为此,threading 模块提供了许多同步原语,包括锁、事件、条件变量和信号量。
虽然这些工具非常强大,但轻微的设计错误可能会导致难以重现的问题。因此,任务协调的首选方法是将对资源的所有访问集中在一个线程中,然后使用 queue 模块将来自其他线程的请求提供给该线程。使用 Queue 对象进行线程间通信和协调的应用程序更容易设计,可读性更强,也更可靠。
日志记录
logging 模块提供了一个功能齐全、灵活的日志系统。最简单的方法是将日志消息发送到文件或 sys.stderr:
import logging
logging.debug('Debugging information')
logging.info('Informational message')
logging.warning('Warning:config file %s not found', 'server.conf')
logging.error('Error occurred')
logging.critical('Critical error -- shutting down')
复制
这将产生以下输出:
WARNING:root:Warning:config file server.conf not found
ERROR:root:Error occurred
CRITICAL:root:Critical error -- shutting down
复制
默认情况下,信息和调试消息被抑制,输出被发送到标准错误。其他输出选项包括通过电子邮件、数据报、套接字路由消息,或将消息路由到 HTTP 服务器。新的过滤器可以根据消息优先级:DEBUG、INFO、WARNING、ERROR 和 CRITICAL 选择不同的路由。
日志系统可以直接从 Python 配置,也可以从用户可编辑的配置文件中加载,以便在不改变应用程序的情况下定制日志记录。
官方文档:
https://docs.python.org/3.9/tutorial/stdlib2.html