影刀RPA 文件锁机制防止多流程同时操作同一文件作者林焱什么情况用这个你设了两个定时任务一个每天9点跑一个每天9点05分跑。正常情况下没问题但某天第一个任务因为数据量大跑了10分钟还没结束第二个任务9点05分准时启动——两个流程同时操作同一个Excel文件。结果数据写乱、文件损坏、流程崩溃。这就是文件竞争问题。但凡有多个流程或多台机器上的流程可能同时操作同一个文件你就需要文件锁——让后来的流程排队等前面的流程操作完再动文件。怎么做第一步最简单的文件锁——锁文件创建一个.lock文件作为互斥信号importosimporttimeclassSimpleFileLock:简单的文件锁——通过创建锁文件实现def__init__(self,file_path):# 锁文件名为原文件名加 .lock 后缀self.lock_pathfile_path.lockself.lockedFalsedefacquire(self,timeout30): 获取锁——如果已有锁则等待 timeout: 最多等多少秒 start_timetime.time()whiletime.time()-start_timetimeout:try:# 尝试创建锁文件排他性创建fdos.open(self.lock_path,os.O_CREAT|os.O_EXCL|os.O_RDWR)os.write(fd,str(os.getpid()).encode())os.close(fd)self.lockedTrueprint(f✓ 获取文件锁:{self.lock_path})returnTrueexceptFileExistsError:# 锁文件已存在检查是否是僵尸锁ifself._is_stale_lock():print(检测到僵尸锁强制删除)os.remove(self.lock_path)continue# 等待后重试time.sleep(0.5)raiseTimeoutError(f获取文件锁超时({timeout}秒):{self.lock_path})defrelease(self):释放锁——删除锁文件ifself.lockedandos.path.exists(self.lock_path):try:os.remove(self.lock_path)self.lockedFalseprint(f✓ 释放文件锁:{self.lock_path})except:print(f⚠ 释放锁文件失败可能已被其他进程删除)def_is_stale_lock(self,stale_seconds600):检查是否是僵尸锁锁文件超过10分钟没更新try:mtimeos.path.getmtime(self.lock_path)[video(video-rUmwVQOx-1784314413640)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)]returntime.time()-mtimestale_secondsexcept:returnFalsedef__enter__(self):支持 with 语句self.acquire()returnselfdef__exit__(self,*args):离开 with 块时自动释放self.release()# 使用方式 excel_pathD:/data/销售数据.xlsx# 方式1手动获取和释放lockSimpleFileLock(excel_path)lock.acquire(timeout60)try:# 安全地操作文件modify_excel(excel_path)finally:lock.release()# 无论如何都要释放锁# 方式2with语句推荐——自动释放withSimpleFileLock(excel_path)aslock:modify_excel(excel_path)# with块结束后自动释放锁第二步带标识信息的文件锁知道是谁锁住了文件对排查问题很重要classIdentifiableFileLock(SimpleFileLock):带标识信息的文件锁——知道谁在锁defacquire(self,ownerunknown,timeout30):self.ownerowner start_timetime.time()whiletime.time()-start_timetimeout:try:fdos.open(self.lock_path,os.O_CREAT|os.O_EXCL|os.O_RDWR)# 写入锁持有者信息infofPID:{os.getpid()}\nOwner:{owner}\nTime:{datetime.now().isoformat()}os.write(fd,info.encode())os.close(fd)self.lockedTruereturnTrueexceptFileExistsError:# 读取现有锁的信息try:withopen(self.lock_path,r)asf:current_infof.read()print(f文件被锁定:{current_info[:50]}...)except:print(文件被锁定无法读取锁信息)time.sleep(1)raiseTimeoutError(f等待文件锁超时({timeout}秒))第三步数据库记录锁——跨机器的文件锁文件锁只能在同一台机器上工作。多台机器共享文件NAS/网络盘时用数据库实现分布式锁importsqlite3classDatabaseLock:基于数据库的分布式锁——适用于多台机器def__init__(self,db_path,resource_name):self.db_pathdb_path self.resource_nameresource_name self.lock_idf{os.getpid()}_{int(time.time())}self._init_db()def_init_db(self):初始化锁表connsqlite3.connect(self.db_path)conn.execute( CREATE TABLE IF NOT EXISTS distributed_locks ( resource TEXT PRIMARY KEY, lock_id TEXT NOT NULL, acquired_at TEXT NOT NULL, expire_at TEXT NOT NULL ) )conn.commit()conn.close()defacquire(self,timeout30,lock_ttl300):获取分布式锁start_timetime.time()expire_atdatetime.now()timedelta(secondslock_ttl)whiletime.time()-start_timetimeout:connsqlite3.connect(self.db_path)try:# 先清理过期锁conn.execute(DELETE FROM distributed_locks WHERE resource ? AND expire_at ?,(self.resource_name,datetime.now().isoformat()))# 尝试插入锁主键冲突已被锁定conn.execute(INSERT INTO distributed_locks VALUES (?, ?, ?, ?),(self.resource_name,self.lock_id,datetime.now().isoformat(),expire_at.isoformat()))conn.commit()returnTrueexceptsqlite3.IntegrityError:# 已被锁定conn.rollback()time.sleep(1)finally:conn.close()raiseTimeoutError(f获取分布式锁超时:{self.resource_name})defrelease(self):释放锁connsqlite3.connect(self.db_path)conn.execute(DELETE FROM distributed_locks WHERE resource ? AND lock_id ?,(self.resource_name,self.lock_id))conn.commit()conn.close()defrefresh(self,lock_ttl300):续期——延长锁的有效期适用于长时间操作connsqlite3.connect(self.db_path)new_expiredatetime.now()timedelta(secondslock_ttl)conn.execute(UPDATE distributed_locks SET expire_at ? WHERE resource ? AND lock_id ?,(new_expire.isoformat(),self.resource_name,self.lock_id))conn.commit()conn.close()第四步带队列的等待机制等锁的时候不是傻等而是有序排队classQueuedFileLock(SimpleFileLock):带队列的文件锁——先到先得def__init__(self,file_path):super().__init__(file_path)self.queue_pathfile_path.queuedefacquire(self,timeout60):排队获取锁my_idf{os.getpid()}_{int(time.time()*1000)}# 加入队列self._join_queue(my_id)queue_posself._get_queue_position(my_id)print(f排队中... 当前第{queue_pos}位)start_timetime.time()whiletime.time()-start_timetimeout:# 检查是否轮到自己ifself._is_my_turn(my_id):# 尝试获取锁try:fdos.open(self.lock_path,os.O_CREAT|os.O_EXCL|os.O_RDWR)os.close(fd)self.lockedTrueself._leave_queue(my_id)print(轮到你了已获取锁)returnTrueexceptFileExistsError:time.sleep(0.5)continuetime.sleep(1)self._leave_queue(my_id)raiseTimeoutError(排队超时)def_join_queue(self,my_id):withopen(self.queue_path,a)asf:f.write(my_id\n)def_get_queue_position(self,my_id):ifnotos.path.exists(self.queue_path):return1withopen(self.queue_path,r)asf:queue[line.strip()forlineinfifline.strip()]returnqueue.index(my_id)1ifmy_idinqueueelselen(queue)1def_is_my_turn(self,my_id):ifnotos.path.exists(self.queue_path):returnTruewithopen(self.queue_path,r)asf:firstf.readline().strip()returnfirstmy_iddef_leave_queue(self,my_id):ifnotos.path.exists(self.queue_path):returnwithopen(self.queue_path,r)asf:linesf.readlines()withopen(self.queue_path,w)asf:f.writelines(lineforlineinlinesifline.strip()!my_id)有什么坑坑一流程崩溃导致锁没释放僵尸锁现象流程中途崩溃锁文件留在磁盘上后续所有流程都卡住了。解决在锁文件中写入时间戳超过合理时长如10分钟自动判断为僵尸锁用try-finally或with语句确保死也释放锁坑二网络盘上的文件锁不可靠现象NAS上的.lock文件创建成功了但另一台机器看不到缓存延迟。解决网络共享文件用数据库锁SQLite也行放在本地不要依赖文件系统锁。坑三锁的粒度过粗TEMU店群如何管理运营现象所有流程共享一个锁文件D:/data/.lock导致无关流程也在排队。解决按文件或按业务类型细分锁# 不同数据文件用不同的锁lock_salesSimpleFileLock(D:/data/销售数据.xlsx)lock_inventorySimpleFileLock(D:/data/库存数据.xlsx)# 两个锁互不影响坑四while循环等锁导致CPU空转现象不加sleep的while循环占满一个CPU核心。解决循环中必须有time.sleep()间隔建议0.3-1秒。坑五多台机器时间不同步现象分布式锁中依赖时间戳判断过期但机器A和机器B时间差5分钟导致误判。解决分布式环境用NTP同步时间锁的过期时间设得宽松一些如10分钟。总结文件锁从最简方案开始创建一个.lock文件操作完删除。当你有跨机器的需求时升级到数据库锁。记住两条铁律锁一定要能自动过期防僵尸锁、释放锁写在finally里防异常不释放。