filelock , 一个神奇的 Python 库

2025-06-09 07:37:03 栏目：宝塔面板 95 阅读

filelock 是一个 Python 第三方库，用于实现跨平台的文件锁功能。它允许多个线程或进程在访问共享文件时进行互斥操作，避免并发写入等问题导致的数据混乱。

安装

pip install filelock

特性

支持基于文件的锁定机制
提供了上下文管理器来自动管理锁的获取和释放
支持超时机制，防止死锁情况发生
可以管理多个文件的锁定状态

基础示例

from filelock import FileLock

# 定义锁文件（通常是在目标文件后加.lock）和数据文件
lock_file = "data.txt.lock"
data_file = "data.txt"

# 创建一个锁对象
lock = FileLock(lock_file)

# 使用with语句获取锁（阻塞直到获取锁）
with lock:
    with open(data_file, "a") as f:
        f.write("新条目
")
# 锁会在with块结束后自动释放

data.txt.lock 是锁文件的名称。当获取文件锁时，会创建这个文件，释放锁时，会删除这个文件（如果文件系统支持删除操作，否则可能只是释放锁，文件仍然存在）。

(1) 非阻塞锁

通过设置锁的timeout超时时间为0，设置为非阻塞锁

try:
    # 设置timeout=0表示非阻塞，如果获取不到锁立即抛出Timeout异常
    with lock.acquire(timeout=0):
        with open(data_file, "a") as f:
            f.write("快速更新
")
except Timeout:
    print("锁被占用，稍后再试。")

(2) 手动控制锁

lock.acquire()   # 获取锁（阻塞）
try:
    with open(data_file, "a") as f:
        f.write("手动加锁
")
finally:
    lock.release()  # 务必在finally中释放锁，避免死锁

实际应用

(1) 数据库连接池中的文件锁定

在数据库连接池的实现中，通常会使用文件锁来控制对数据库连接文件的并发访问。

from filelock import FileLock

# 创建数据库连接池文件锁
db_lock = FileLock("db_pool.lock")

# 获取数据库连接
def get_db_connection():
    with db_lock:
        # 在锁定范围内执行获取数据库连接的操作
        print("获取数据库连接")
        # 返回数据库连接对象
        return db_connection

使用文件锁 db_lock 来控制对数据库连接池文件的并发访问，确保同一时间只有一个线程可以获取数据库连接。

(2) 多线程/多进程环境下的文件访问控制

在多线程或多进程的环境中，文件锁可以用来控制对共享文件的并发访问，避免数据竞争和冲突。

from filelock import FileLock
from concurrent.futures import ThreadPoolExecutor

# 创建文件锁
file_lock = FileLock("shared_file.txt.lock")

# 多线程处理文件写入操作
def write_to_file(data):
    with file_lock:
        # 在锁定范围内执行文件写入操作
        with open("shared_file.txt", "a") as file:
            file.write(data + "
")

# 创建线程池
executor = ThreadPoolExecutor(max_workers=5)

# 提交多个写入任务
data_list = ["Data 1", "Data 2", "Data 3", "Data 4", "Data 5"]
for data in data_list:
    executor.submit(write_to_file, data)

使用文件锁file_lock来控制多线程环境下对共享文件的并发写入操作，确保数据写入的安全性和一致性。

(3) 日志文件的并发写入控制

在日志系统中，多个进程或线程同时写入日志文件时，可以使用文件锁来控制对日志文件的并发访问，避免日志信息错乱和丢失

from filelock import FileLock
import logging

# 创建日志文件锁
log_lock = FileLock("app.log.lock")

# 配置日志系统
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 写入日志信息的函数
def write_log_message(message):
    with log_lock:
        # 在锁定范围内执行写入日志信息的操作
        logging.info(message)

# 测试写入日志信息
write_log_message("This is a log message.")

使用文件锁log_lock来控制对日志文件的并发写入操作，确保日志信息按照正确的顺序写入文件中。