温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

python fcntl 文件锁

发布时间:2020-07-11 23:07:25 来源:网络 阅读:1556 作者:qw87112 栏目:编程语言

此模块只有在 unix 系统上才有,windows 没有。

文档地址:

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

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

多进程示例程序
import fcntl
import os
import time
from multiprocessing import Pool

def 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 Flask
from service.extensions import scheduler
import logging
from logging.handlers import RotatingFileHandler
import os
import fcntl, atexit

basedir = 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 app

def register_apscheduler(app):
    scheduler.init_app(app)
    from service import aliyuncron
    scheduler.start()

app = create_app()

@app.route('/')
def index():
    return '<h2>Hello World!</h2>'
向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI