python fcntl 文件锁，,此模块只有在 uni

python fcntl 文件锁，,此模块只有在 uni

文档地址：

https://docs.python.org/3.7/library/fcntl.html

https://www.docs4dev.com/docs/zh/python/3.7.2rc1/all/library-fcntl.html

多进程示例程序

import fcntlimport osimport timefrom multiprocessing import Pooldef worker():    print(‘task: %s‘ % os.getpid())    try:        f = open(‘scheduler.lock‘, ‘wb‘)        ‘‘‘加锁，同步锁，其他进程获取不到需等待‘‘‘        fcntl.flock(f, fcntl.LOCK_EX)        print(‘I am get file %s‘ % os.getpid())        time.sleep(10)        ‘‘‘解锁‘‘‘        fcntl.flock(f, fcntl.LOCK_UN)        f.close()    except Exception as e:        print(‘file is already locked: %s‘ % os.getpid())if __name__ == ‘__main__‘:    p = Pool(4)    for i in range(5):        p.apply_async(worker)    print(‘Waiting for all subprocesses done...‘)    p.close()    p.join()    print(‘All subprocesses done.‘)

fcntl 详细参数

‘‘‘f 需传入文件对象，operator 传入加锁方式‘‘‘fcntl.flock(f, operator)fcntl.LOCK_SH    ‘共享锁‘fcntl.LOCK_EX    ‘排他锁‘fcntl.LOCK_NB    ‘非阻塞锁——如果指定此参数，函数不能获得文件锁就立即返回，否则，函数会等待获得文件锁。LOCK_NB可以同LOCK_SH或LOCK_NB进行按位或（|）运算操作。 fcntl.flock (f,fcntl.LOCK_EX|fcntl.LOCK_NB)‘fcntl.LOCK_UN    ‘解锁‘

解决 gunicorn flask-apscheduler 重复执行问题

from flask import Flaskfrom service.extensions import schedulerimport loggingfrom logging.handlers import RotatingFileHandlerimport osimport fcntl, atexitbasedir = os.path.abspath(‘‘)def create_app():    app = Flask(__name__)    f = open("scheduler.lock", "wb")    try:        ‘‘‘使用非阻塞锁‘‘‘        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)        register_apscheduler(app)    except Exception as e:        pass    def unlock():        fcntl.flock(f, fcntl.LOCK_UN)        f.close()    ‘‘‘注册一个退出回调函数，用于在程序退出时进行一些清理工作，关闭文件，解除锁‘‘‘    atexit.register(unlock)    return appdef register_apscheduler(app):    scheduler.init_app(app)    from service import aliyuncron    scheduler.start()app = create_app()@app.route(‘/‘)def index():    return ‘<h1>Hello World!</h1>‘

python fcntl 文件锁