python自带库fcntl文件锁

作用：保证只用1次，一般在多线程，但是只需要1次运行

文件锁fcntl库 简介

Python 的 fcntl 库提供了对文件描述符的控制操作，类似于 Unix 中的 fcntl() 和 ioctl() 系统调用。这个库允许你对文件描述符进行锁定，以实现进程间的同步，以及执行其他文件控制操作。fcntl 库的主要作用包括：

1. 文件锁定：通过 fcntl.flock() 或 fcntl.lockf() 函数，可以在文件描述符上执行锁定操作，防止其他进程同时修改文件，从而避免数据冲突。这些锁定操作包括排他锁（LOCK_EX）、共享锁（LOCK_SH）以及解锁（LOCK_UN）。

2. 非阻塞锁定：可以使用 LOCK_NB 标志来执行非阻塞锁定，如果锁定失败，函数会立即返回而不是等待。

3. 文件控制操作：fcntl.fcntl() 函数可以执行各种文件控制操作，如改变文件描述符的属性或获取文件状态标志。

4. I/O 控制：fcntl.ioctl() 函数用于执行 I/O 控制操作，这通常涉及到设备特定的操作，如获取或设置终端属性。

5. 审计事件：在某些操作中，fcntl 库会触发审计事件，这对于系统安全和监控非常有用。

6. 文件描述符的非阻塞模式：fcntl 库可以设置文件描述符为非阻塞模式，使得文件操作如读取不会阻塞进程，如果没有数据可读，会立即返回错误。

fcntl 库在需要对文件进行进程间同步或执行特定文件控制操作时非常有用，尤其是在 Unix 或类 Unix 系统上进行系统编程时。

文件锁fcntl

import fcntl
import atexit
f = open("scheduler.lock", "wb")
try:
    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except:
    pass
def unlock():
    fcntl.flock(f, fcntl.LOCK_UN)
    f.close()
atexit.register(unlock)

flock() : flock(f, operation)

文件锁fcntl 类型

operation : 包括：
LOCK_SH 共享锁:所有进程没有写访问权限，即使是加锁进程也没有。所有进程有读访问权限。
LOCK_EX 排他锁:除加锁进程外其他进程没有对已加锁文件读写访问权限。
LOCK_NB 非阻塞锁:如果指定此参数，函数不能获得文件锁就立即返回，否则，函数会等待获得文件锁。
LOCK_NB可以同LOCK_SH或LOCK_NB进行按位或（|）运算操作。 fcnt.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB)

文件锁fcntl例子：

import fcntl
import atexit
import multiprocessing
import time
class fcntl_fun():
    def __init__(self, lock_file, open_type):
        self.lock_file = lock_file
        self.file = open(self.lock_file, open_type)
        # file = self.open_file()
    # def open_file(self):
    #     self.file = open(self.lock_file, "wb")
    # 共享锁_不等
    def add_share_no_wait(self):
        try:
            print("共享锁_不等")
            fcntl.flock(self.file, fcntl.LOCK_SH | fcntl.LOCK_NB)
            return True
        except:
            print("共享锁_不等_出错")
            return False
    # 排他锁_不等
    def add_other_no_wait(self):
        try:
            print("排他锁_不等")
            fcntl.flock(self.file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            return True
        except:
            print("排他锁_不等_出错")
            return False
    # 共享锁_等
    def add_share_wait(self):
        try:
            print("共享锁_等")
            fcntl.flock(self.file, fcntl.LOCK_SH)
            return True
        except:
            print("共享锁_等_出错")
            return False
    # 排他锁_等
    def add_other_wait(self):
        try:
            print("排他锁_等")
            fcntl.flock(self.file, fcntl.LOCK_EX)
            return True
        except:
            print("排他锁_等_出错")
            return False
    # 解锁
    def unlock(self):
        fcntl.flock(self.file, fcntl.LOCK_UN)
        self.file.close()
    def write(self, i):
        self.file.write(str(i))
        self.file.write("\n")
# atexit.register(unlock)
def func_1(number,lock_file, open_type):
    print("子进程_1")
    f = fcntl_fun(lock_file, open_type)
    result = f.add_other_wait()  # 换这里
    print(result)
    if result:
        for i in range(number):
            time.sleep(1)
            print("子进程_1_" + str(i))
def func_2(number,lock_file, open_type):
    print("子进程_2")
    f = fcntl_fun(lock_file, open_type)
    result = f.add_other_wait()  # 换这里
    print(result)
    if result:
        for i in range(number):
            time.sleep(1)
            f.write(i)
            print("子进程_2_" + str(i))
if __name__ == "__main__":
    begin_time = time.time()
    pool = multiprocessing.Pool(processes=2) 
    # 可以用来换上面验证
    # result = f.add_share_no_wait()  # 共享锁_不等
    # result = f.add_other_no_wait()  # 排他锁_不等
    # result = f.add_share_wait()  # 共享锁_等
    # result = f.add_other_wait()  # 排他锁_等
    # print(result)
    # lock_file = "scheduler.lock"
    lock_file = "/Users/chu/Desktop/scheduler.txt"
    open_type = "w"
    res = pool.apply_async(func_1, (10,lock_file, open_type))
    time.sleep(3)
    res = pool.apply_async(func_2, (10,lock_file, open_type))
    pool.close()
    pool.join() 
    end_time = time.time()
    print(end_time - begin_time)

文件锁fcntl 总结