Flock 锁的详细使用指南

flock [options] <file|directory|fd> <command>

#!/bin/bash
LOCK_FILE="/tmp/my_script.lock"

# 尝试获取排他锁，如果失败则退出
exec 200>"$LOCK_FILE"
flock -n 200 || {
    echo "Another instance is running"
    exit 1
}

# 脚本主体
echo "Running critical section at $(date)"
sleep 10

# 自动释放锁（脚本结束时文件描述符关闭）

#!/bin/bash
LOG_FILE="/var/log/app.log"
LOCK_FILE="/var/lock/app.log.lock"

# 使用flock命令直接保护写入操作
flock -x "$LOCK_FILE" -- \
    echo "$(date): New log entry" >> "$LOG_FILE"

#!/bin/bash
LOCK_FILE="/tmp/process.lock"

# 非阻塞模式，如果无法获取锁立即返回
if ! flock -n 9; then
    echo "Cannot acquire lock, exiting"
    exit 1
fi 9>"$LOCK_FILE"

# 临界区代码
echo "Processing..."

import fcntl
import time

def write_with_lock(filename, content):
    with open(filename, 'a') as f:
        # 获取排他锁
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            # 写入操作
            f.write(content + '\n')
            f.flush()
        finally:
            # 释放锁
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

import fcntl
import os

class FileLock:
    def __init__(self, lock_file, timeout=None):
        self.lock_file = lock_file
        self.timeout = timeout
        self.fd = None
    
    def __enter__(self):
        self.fd = open(self.lock_file, 'w+')
        start_time = time.time()
        
        while True:
            try:
                fcntl.flock(self.fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except IOError as e:
                if self.timeout is not None and time.time() - start_time > self.timeout:
                    raise TimeoutError("Could not acquire lock within timeout")
                time.sleep(0.1)
        
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        fcntl.flock(self.fd.fileno(), fcntl.LOCK_UN)
        self.fd.close()

# 使用示例
try:
    with FileLock("/tmp/myapp.lock", timeout=5):
        print("Lock acquired, performing critical operation")
        # 执行临界区代码
except TimeoutError:
    print("Could not acquire lock within timeout period")

import json
import fcntl
import os

def update_config(config_path, update_func):
    """安全更新JSON配置文件"""
    # 读取现有配置
    with open(config_path, 'r+') as f:
        # 获取排他锁
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        
        try:
            # 读取内容
            config = json.load(f)
            
            # 应用更新
            updated_config = update_func(config)
            
            # 回到文件开始位置
            f.seek(0)
            f.truncate()
            
            # 写入更新后的内容
            json.dump(updated_config, f, indent=2)
            f.flush()
            os.fsync(f.fileno())  # 确保写入磁盘
        finally:
            # 释放锁
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

# 1. 将文件描述符200关联到锁文件（打开文件）
exec 200>"$LOCK_FILE"

# 2. 对文件描述符200应用非阻塞排他锁
flock -n 200 || {
    echo "Another instance is running"
    exit 1
}

# 3. 脚本执行主体
echo "Critical section running"

# 4. 脚本结束时自动关闭文件描述符200，释放锁
# （也可以显式关闭：exec 200>&-）

int fd = open("/tmp/my_script.lock", O_RDWR|O_CREAT, 0644); // fd=200
if (flock(fd, LOCK_EX|LOCK_NB) != 0) { // 非阻塞尝试获取排他锁
    printf("Another instance is running\n");
    exit(1);
}
// ... 执行临界区代码 ...
close(fd); // 自动释放锁

exec 3>"$LOCK_FILE"  # 使用文件描述符3
flock -n 3 || exit 1

flock -n /tmp/lockfile.lock -c "command_to_run"