Python防止脚本重复执行

脚本重复执行是一个常见的问题，尤其当我们需要保证脚本只执行一次时。本文将从多个方面介绍如何使用Python防止脚本重复执行。

一、使用文件锁

使用文件锁是一种常见的防止脚本重复执行的方法。我们可以在脚本开始时创建一个特定的文件，并在脚本结束时删除该文件。在脚本执行前，我们可以检查是否存在该文件，如果存在则表示脚本已经在执行，可以选择退出或者等待。

import os
import time

def check_lock_file():
    if os.path.exists('lockfile.txt'):
        print('Script is already running. Exiting...')
        return True
    else:
        with open('lockfile.txt', 'w') as f:
            f.write('lock')
        return False

def remove_lock_file():
    os.remove('lockfile.txt')

if check_lock_file():
    exit()

# 执行脚本的代码片段

remove_lock_file()

二、使用守护进程

守护进程是一种在后台运行的进程，可以独立于终端会话运行。我们可以将脚本编写为守护进程的形式，确保脚本只有一个实例在运行。

import os
import sys
import daemon

def run_script():
    # 执行脚本的代码

with daemon.DaemonContext():
    run_script()

三、使用进程锁

进程锁是一种同步机制，用于确保同一时间只有一个进程可以访问共享资源。我们可以使用Python的multiprocessing库中的锁对象来实现进程锁。

import multiprocessing

def run_script(lock):
    # 获取进程锁
    lock.acquire()
    
    try:
        # 执行脚本的代码
    finally:
        # 释放进程锁
        lock.release()

if __name__ == '__main__':
    lock = multiprocessing.Lock()
    process = multiprocessing.Process(target=run_script, args=(lock,))
    process.start()
    process.join()

四、使用定时任务

定时任务是一种在指定的时间间隔执行指定任务的方法。我们可以使用Python的sched模块来实现定时任务，并设定脚本只执行一次。

import time
import sched

def run_script():
    # 执行脚本的代码

def schedule_once():
    scheduler = sched.scheduler(time.time, time.sleep)
    scheduler.enter(0, 1, run_script)
    scheduler.run()

schedule_once()

五、使用互斥锁

互斥锁是一种同步机制，用于确保同一时间只有一个线程可以访问共享资源。我们可以使用Python的threading库中的Lock对象来实现互斥锁。

import threading

def run_script(lock):
    # 获取互斥锁
    lock.acquire()
    
    try:
        # 执行脚本的代码
    finally:
        # 释放互斥锁
        lock.release()

if __name__ == '__main__':
    lock = threading.Lock()
    thread = threading.Thread(target=run_script, args=(lock,))
    thread.start()
    thread.join()

六、使用标记文件

标记文件是一种简单有效的方法，通过创建一个标记文件来表示脚本是否正在执行。我们可以在需要防止脚本重复执行的部分创建一个特定的文件，脚本开始时检查该文件是否存在，存在则表示脚本已经在执行，可以选择退出或者等待。

import os

def check_flag_file():
    if os.path.exists('flagfile.txt'):
        print('Script is already running. Exiting...')
        return True
    else:
        with open('flagfile.txt', 'w') as f:
            f.write('flag')
        return False

def remove_flag_file():
    os.remove('flagfile.txt')

if check_flag_file():
    exit()

# 执行脚本的代码片段

remove_flag_file()

七、使用信号量

信号量是一种同步机制，用于控制对共享资源的访问。我们可以使用Python的multiprocessing库中的Semaphore对象来实现信号量。

import multiprocessing

def run_script(semaphore):
    # 获取信号量
    semaphore.acquire()
    
    try:
        # 执行脚本的代码
    finally:
        # 释放信号量
        semaphore.release()

if __name__ == '__main__':
    semaphore = multiprocessing.Semaphore(1)
    process = multiprocessing.Process(target=run_script, args=(semaphore,))
    process.start()
    process.join()

八、使用数据库标记

使用数据库标记是一种将脚本是否正在执行的状态存储在数据库中的方法。我们可以在脚本开始时在数据库中创建一个特定的记录，并在脚本结束时删除该记录。

import sqlite3

def check_db_flag():
    connection = sqlite3.connect('script.db')
    cursor = connection.cursor()
    
    cursor.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='flag';")
    result = cursor.fetchone()
    
    if result[0] == 0:
        # 创建flag表
        cursor.execute("CREATE TABLE flag (flag_value INTEGER);")
        # 插入标记数据
        cursor.execute("INSERT INTO flag VALUES (1);")
        connection.commit()
        connection.close()
        return False
    else:
        connection.close()
        return True

def remove_db_flag():
    connection = sqlite3.connect('script.db')
    cursor = connection.cursor()
    
    cursor.execute("DELETE FROM flag;")
    connection.commit()
    connection.close()

if check_db_flag():
    exit()

# 执行脚本的代码片段

remove_db_flag()

九、使用网络通信标记

使用网络通信标记是一种将脚本是否正在执行的状态通过网络进行通信的方法。可以通过发送请求或者创建网络连接来标记脚本的执行状态。

import requests

def check_network_flag():
    try:
        response = requests.get('http://localhost:8000/flag')
        if response.status_code == 200 and response.text == '1':
            return True
    except requests.exceptions.RequestException:
        pass
    return False

def set_network_flag():
    requests.post('http://localhost:8000/flag', data={'value': '1'})

def remove_network_flag():
    requests.post('http://localhost:8000/flag', data={'value': '0'})

if check_network_flag():
    exit()

set_network_flag()

# 执行脚本的代码片段

remove_network_flag()

通过以上多种方法，我们可以有效地防止Python脚本重复执行，确保脚本在需要的时候只执行一次。