Linux下Python并行任务的集中式日志服务实现问询

针对多Worker进程日志冲突、资源耗尽的问题，实现单机器集中式日志服务完全可行，以下是具体实现方案：

核心思路 #

启动一个独立的日志守护进程，作为机器上唯一的日志写入进程，所有Worker进程通过IPC（Unix域套接字/消息队列）将日志发送给它，由统一的RotatingFileHandler处理日志写入和轮转，从根本上避免多进程写文件的冲突和资源占用。

具体实现步骤 #

1. 编写中央日志守护进程 #

这个进程负责监听日志请求、处理日志轮转和写入，确保机器上只运行一个实例：

import logging
import os
import pickle
from logging.handlers import RotatingFileHandler
from multiprocessing.connection import Listener

def log_processor():
    # 配置日志轮转Handler
    handler = RotatingFileHandler(
        'worker_task.log',
        maxBytes=3 * 1024 * 1024,
        backupCount=99,
        encoding='utf-8'
    )
    formatter = logging.Formatter(
        '%(asctime)s - %(process)d - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    
    logger = logging.getLogger('central_logger')
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False

    # 监听Unix域套接字，接收Worker日志
    socket_path = '/tmp/central_logger.sock'
    if os.path.exists(socket_path):
        os.unlink(socket_path)
    
    listener = Listener(socket_path, authkey=b'your_secret_key')
    print(f"Central logger running, listening on {socket_path}")

    while True:
        conn = listener.accept()
        try:
            while True:
                record = conn.recv()
                if record is None:
                    break
                logger.handle(record)
        except EOFError:
            pass
        finally:
            conn.close()

def main():
    # 用PID文件+进程存活检查避免重复启动
    pid_file = '/var/run/central_logger.pid'
    if os.path.exists(pid_file):
        with open(pid_file, 'r') as f:
            pid = int(f.read().strip())
        try:
            os.kill(pid, 0)  # 检查进程是否存活
            print("Central logger is already running")
            return
        except OSError:
            # PID无效，清理文件
            os.unlink(pid_file)
    
    # 写入当前进程PID
    with open(pid_file, 'w') as f:
        f.write(str(os.getpid()))
    
    try:
        log_processor()
    finally:
        if os.path.exists(pid_file):
            os.unlink(pid_file)
        if os.path.exists('/tmp/central_logger.sock'):
            os.unlink('/tmp/central_logger.sock')

if __name__ == '__main__':
    main()

2. Worker进程日志配置 #

Worker不再直接操作日志文件，而是通过自定义Handler将日志发送给中央守护进程：

import logging
import pickle
import socket
from logging.handlers import SocketHandler

class UnixSocketLogHandler(SocketHandler):
    def __init__(self, socket_path):
        super().__init__('localhost', 0)
        self.socket_path = socket_path

    def createSocket(self):
        # 创建Unix域套接字连接
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(self.socket_path)
        return sock

    def makePickle(self, record):
        # 序列化日志记录，传递给中央进程
        return pickle.dumps(record, pickle.HIGHEST_PROTOCOL)

def init_worker_logger():
    logger = logging.getLogger('worker_logger')
    logger.setLevel(logging.INFO)
    
    # 避免日志传递到根Logger
    logger.propagate = False

    # 添加Unix套接字Handler
    handler = UnixSocketLogHandler('/tmp/central_logger.sock')
    logger.addHandler(handler)
    
    return logger

# 你的Worker计算函数
def compute_task(data):
    logger = init_worker_logger()
    logger.info(f"Start processing data: {data}")
    
    # 核心计算逻辑...
    
    logger.info(f"Finish processing data: {data}")
    return result

3. 部署与运行 #

• 启动中央日志守护进程：python logger_daemon.py，建议用systemd配置为开机自启的服务，避免进程意外退出
• Worker进程正常启动即可，所有日志会自动发送到中央进程统一写入

关键细节优化 #

• 进程唯一性保障：除了PID文件，还可以用fcntl.flock给PID文件加排他锁，防止多进程同时启动守护进程
• 重连机制：在Worker的UnixSocketLogHandler中添加重连逻辑，避免中央进程重启后Worker日志丢失
• 日志标识：在日志格式中加入Worker进程ID、任务ID等信息，方便后续排查问题
• 安全注意：Unix域套接字仅本地可见，无需担心跨网络安全问题；如果需要跨机器，可以改用TCP套接字并添加身份验证

替代方案 #

如果不想自己实现守护进程，也可以：

• 将日志发送给系统syslog服务，通过配置syslog的轮转规则实现类似效果（比如用logging.handlers.SysLogHandler）
• 使用成熟的日志聚合工具收集Worker日志，再统一写入文件并管理轮转，但这需要额外部署工具

内容的提问来源于stack exchange，提问作者Mat