import fcntl

# 打開文件
with open("data.txt", "r+") as f:
    # 獲取排他鎖
    fcntl.flock(f, fcntl.LOCK_EX)
    
    # 執(zhí)行文件操作
    content = f.read()
    f.seek(0)
    f.write(content + "new data")
    f.truncate()
    
    # 釋放鎖（退出with塊時文件關(guān)閉，鎖也會自動釋放）

import fcntl
import errno

with open("data.txt", "r") as f:
    try:
        # 嘗試獲取共享鎖，不阻塞
        fcntl.flock(f, fcntl.LOCK_SH | fcntl.LOCK_NB)
        print("成功獲取共享鎖")
        # 執(zhí)行讀取操作
        content = f.read()
        # 釋放鎖
        fcntl.flock(f, fcntl.LOCK_UN)
    except BlockingIOError as e:
        if e.errno == errno.EWOULDBLOCK:
            print("無法獲取鎖，文件已被鎖定")
        else:
            raise

import fcntl
import struct

def lock_range(fd, start, length, exclusive=True):
    # 構(gòu)建鎖結(jié)構(gòu)
    lock_type = fcntl.F_WRLCK if exclusive else fcntl.F_RDLCK
    lock_struct = struct.pack('hhllhh', lock_type, 0, start, length, 0, 0)
    fcntl.fcntl(fd, fcntl.F_SETLK, lock_struct)

def unlock_range(fd, start, length):
    lock_struct = struct.pack('hhllhh', fcntl.F_UNLCK, 0, start, length, 0, 0)
    fcntl.fcntl(fd, fcntl.F_SETLK, lock_struct)

# 使用示例
with open("large_file.dat", "r+") as f:
    # 鎖定文件的第100-200字節(jié)
    lock_range(f, 100, 100)
    
    # 操作該范圍的數(shù)據(jù)
    f.seek(100)
    data = f.read(100)
    # 處理數(shù)據(jù)...
    f.seek(100)
    f.write(processed_data)
    
    # 解鎖
    unlock_range(f, 100, 100)

import msvcrt
import os

def lock_file(f, exclusive=True):
    # 鎖定整個文件
    f.seek(0, os.SEEK_END)
    file_size = f.tell()
    f.seek(0)
    
    # 鎖定類型：1表示共享鎖，2表示排他鎖
    lock_type = 2 if exclusive else 1
    msvcrt.locking(f.fileno(), lock_type, file_size)

def unlock_file(f):
    # 解鎖整個文件
    f.seek(0, os.SEEK_END)
    file_size = f.tell()
    f.seek(0)
    
    # 8表示解鎖
    msvcrt.locking(f.fileno(), 8, file_size)

# 使用示例
with open("data.txt", "r+") as f:
    try:
        # 獲取排他鎖
        lock_file(f)
        # 執(zhí)行文件操作
        content = f.read()
        f.seek(0)
        f.write(content + "new data")
        f.truncate()
    finally:
        # 確保解鎖
        unlock_file(f)

pip install filelock

from filelock import FileLock

# 創(chuàng)建鎖文件和數(shù)據(jù)文件
lock = FileLock("data.txt.lock")
data_file = "data.txt"

with lock:
    # 在with塊中，鎖被獲取
    with open(data_file, "a") as f:
        f.write("新的數(shù)據(jù)行\(zhòng)n")
# 退出with塊后，鎖自動釋放

from filelock import FileLock, Timeout

lock = FileLock("data.txt.lock", timeout=0)  # timeout=0表示非阻塞

try:
    with lock.acquire(timeout=0):  # 也可以在這里指定timeout
        with open("data.txt", "r") as f:
            content = f.read()
        print(f"讀取內(nèi)容: {content}")
except Timeout:
    print("無法獲取鎖，已被其他進程占用")

lock = FileLock("data.txt.lock")

try:
    # 獲取鎖，最多等待5秒
    lock.acquire(timeout=5)
    # 執(zhí)行文件操作
    with open("data.txt", "r+") as f:
        content = f.read()
        # 處理內(nèi)容...
finally:
    # 確保釋放鎖
    if lock.is_locked:
        lock.release()

import multiprocessing
import time
import random
from filelock import FileLock

def write_log(process_id, lock_file, log_file):
    lock = FileLock(lock_file)
    
    for i in range(5):
        # 生成日志內(nèi)容
        log_message = f"進程 {process_id} 日志 {i}: {time.ctime()}\n"
        
        # 獲取鎖并寫入日志
        with lock:
            with open(log_file, "a") as f:
                f.write(log_message)
        
        # 隨機休眠，模擬不同進程的執(zhí)行間隔
        time.sleep(random.uniform(0.1, 0.5))

if __name__ == "__main__":
    log_file = "app.log"
    lock_file = "app.log.lock"
    
    # 清除現(xiàn)有日志
    with open(log_file, "w") as f:
        pass
    
    # 創(chuàng)建5個進程
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=write_log, args=(i, lock_file, log_file))
        processes.append(p)
        p.start()
    
    # 等待所有進程完成
    for p in processes:
        p.join()
    
    print("所有進程完成，日志已寫入")

import multiprocessing
import json
from filelock import FileLock

def process_data(process_id, data_file, lock_file):
    lock = FileLock(lock_file)
    
    for _ in range(3):
        # 讀取數(shù)據(jù)（需要共享鎖）
        with lock.acquire(timeout=10):
            with open(data_file, "r") as f:
                data = json.load(f)
        
        # 處理數(shù)據(jù)（不需要鎖）
        processed_value = data["value"] * 2
        print(f"進程 {process_id} 處理后的值: {processed_value}")
        
        # 寫入結(jié)果（需要排他鎖）
        with lock.acquire(timeout=10):
            with open(data_file, "r+") as f:
                data = json.load(f)
                data["value"] = processed_value
                f.seek(0)
                json.dump(data, f)
                f.truncate()

if __name__ == "__main__":
    data_file = "data.json"
    lock_file = "data.json.lock"
    
    # 初始化數(shù)據(jù)文件
    with open(data_file, "w") as f:
        json.dump({"value": 1}, f)
    
    # 創(chuàng)建3個進程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=process_data, args=(i, data_file, lock_file))
        processes.append(p)
        p.start()
    
    # 等待所有進程完成
    for p in processes:
        p.join()
    
    # 查看最終結(jié)果
    with open(data_file, "r") as f:
        final_data = json.load(f)
    print(f"最終結(jié)果: {final_data}")

import multiprocessing
import fcntl
import os

def write_chunk(process_id, file_path, start, length):
    with open(file_path, "r+") as f:
        # 鎖定指定范圍
        fcntl.fcntl(f, fcntl.F_SETLK, fcntl struct.pack('hhllhh', fcntl.F_WRLCK, 0, start, length, 0, 0))
        
        # 寫入數(shù)據(jù)
        f.seek(start)
        data = f"進程 {process_id} 寫入的內(nèi)容".ljust(length, 'x')
        f.write(data)
        
        # 解鎖（可選，文件關(guān)閉時會自動解鎖）
        # fcntl.fcntl(f, fcntl.F_SETLK, struct.pack('hhllhh', fcntl.F_UNLCK, 0, start, length, 0, 0))

if __name__ == "__main__":
    file_path = "chunked_file.dat"
    file_size = 1000  # 1000字節(jié)的文件
    
    # 創(chuàng)建空文件并設(shè)置大小
    with open(file_path, "w") as f:
        f.write('\0' * file_size)
    
    # 每個進程負責(zé)寫入不同的塊
    processes = []
    chunk_size = 200
    for i in range(5):
        start = i * chunk_size
        p = multiprocessing.Process(target=write_chunk, args=(i, file_path, start, chunk_size))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print("所有塊寫入完成")

# 不推薦：長時間持有鎖
with lock:
    data = read_data()
    result = complex_processing(data)  # 耗時操作
    write_result(result)

# 推薦：只在必要時持有鎖
data = read_data()  # 可能需要讀鎖
result = complex_processing(data)  # 不需要鎖
with lock:  # 只在寫入時持有鎖
    write_result(result)

# 推薦：使用上下文管理器
with lock:
    # 操作文件

# 替代方案：使用try-finally
try:
    lock.acquire()
    # 操作文件
finally:
    lock.release()

from filelock import FileLock, Timeout

lock = FileLock("data.lock")

try:
    # 最多等待5秒
    with lock.acquire(timeout=5):
        # 操作文件
except Timeout:
    # 處理獲取鎖失敗的情況
    print("無法獲取鎖，操作超時")

# 進程1
with lock1:
    # 操作1
    with lock2:
        # 操作2

# 進程2
with lock2:
    # 操作3
    with lock1:
        # 操作4

import multiprocessing
import time
import tempfile
import os
from filelock import FileLock

def test_operation(lock_file, data_file, result_queue):
    try:
        lock = FileLock(lock_file)
        for _ in range(10):
            with lock:
                # 讀取當前值
                with open(data_file, "r") as f:
                    value = int(f.read())
                
                # 模擬處理時間
                time.sleep(0.001)
                
                # 寫入新值
                with open(data_file, "w") as f:
                    f.write(str(value + 1))
        
        result_queue.put(True)
    except Exception as e:
        result_queue.put(str(e))

def test_concurrent_access(num_processes=10):
    # 創(chuàng)建臨時文件
    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(b"0")
        data_file = f.name
    
    lock_file = data_file + ".lock"
    result_queue = multiprocessing.Queue()
    
    # 啟動多個進程
    processes = []
    for _ in range(num_processes):
        p = multiprocessing.Process(
            target=test_operation,
            args=(lock_file, data_file, result_queue)
        )
        processes.append(p)
        p.start()
    
    # 等待所有進程完成
    for p in processes:
        p.join()
    
    # 檢查結(jié)果
    errors = []
    for _ in range(num_processes):
        result = result_queue.get()
        if result is not True:
            errors.append(result)
    
    # 檢查最終值
    with open(data_file, "r") as f:
        final_value = int(f.read())
    
    # 清理臨時文件
    os.unlink(data_file)
    if os.path.exists(lock_file):
        os.unlink(lock_file)
    
    # 輸出測試結(jié)果
    print(f"測試完成: {num_processes}個進程，每個進程遞增10次")
    print(f"預(yù)期結(jié)果: {num_processes * 10}")
    print(f"實際結(jié)果: {final_value}")
    if errors:
        print(f"錯誤: {len(errors)}個錯誤發(fā)生")
        for err in errors:
            print(f"- {err}")
    else:
        print("測試成功，無錯誤發(fā)生")

if __name__ == "__main__":
    test_concurrent_access(20)

import os
from filelock import FileLock

lock_file = "data.lock"

# 檢查并清理舊的鎖文件
if os.path.exists(lock_file):
    # 嘗試獲取鎖，如果失敗，說明鎖正在被使用
    try:
        with FileLock(lock_file, timeout=1):
            pass  # 鎖可以被獲取，說明是殘留文件
        os.remove(lock_file)
    except Timeout:
        pass  # 鎖正在被使用，不做處理

# 正常使用鎖
with FileLock(lock_file):
    # 操作文件

import sys

if sys.platform.startswith('win'):
    # Windows平臺代碼
    import msvcrt
    # ...
else:
    # Unix平臺代碼
    import fcntl
    # ...