如何用Python gevent读取百万行Nginx日志？能否通过pysqml存入MySQL？

嘿，这个问题我刚好有实践经验，来给你拆解一下：

一、关于用gevent读取大日志文件的问题 #

首先得明确：读取本地大文件本身，单线程往往是最优解——因为磁盘IO是瓶颈，多协程/线程反而可能因为上下文切换带来额外开销，并不会让读文件更快。

但如果你的需求是「读文件→解析日志→写入MySQL」这一整套流程，gevent就有用武之地了：解析和写入MySQL属于CPU/网络IO操作，用协程可以在等待MySQL响应的间隙，去处理下一行日志，整体提升吞吐量。

二、用pymysql将数据存入MySQL的注意事项 #

直接用pymysql结合gevent的话，要注意两个关键点：

1. 打gevent猴子补丁：pymysql基于socket实现，必须通过monkey.patch_all()让socket支持协程异步，否则协程会被阻塞。
2. 使用连接池：如果每个协程都创建新的MySQL连接，很快会耗尽MySQL的连接数，用连接池可以复用连接，避免这个问题。

三、完整实践代码 #

先安装依赖：

pip install gevent pymysql dbutils

下面是可直接参考的代码（根据你的实际环境调整配置）：

import gevent
from gevent import monkey, pool
# 打补丁让socket、IO等支持协程异步
monkey.patch_all()

import pymysql
from dbutils.pooled_db import PooledDB
import re
from datetime import datetime

# 配置MySQL连接信息（根据你的实际情况修改）
DB_CONFIG = {
    'host': 'localhost',
    'user': 'your_db_user',
    'password': 'your_db_password',
    'database': 'your_db_name',
    'charset': 'utf8mb4'
}

# 创建MySQL连接池
db_pool = PooledDB(
    creator=pymysql,
    maxconnections=10,  # 连接池最大连接数，别超过MySQL的max_connections配置
    **DB_CONFIG
)

# 适配你的Nginx日志格式的正则表达式（这里是默认的combined格式，根据你的日志调整）
NGINX_LOG_REGEX = re.compile(
    r'^(?P<remote_addr>\S+) - (?P<remote_user>\S+) \[(?P<time_local>[^\]]+)\] '
    r'"(?P<request>[^"]+)" (?P<status>\d+) (?P<body_bytes_sent>\S+) '
    r'"(?P<http_referer>[^"]+)" "(?P<http_user_agent>[^"]+)"'
)

def parse_and_insert_log(line):
    """解析单条日志并插入MySQL"""
    line = line.strip()
    if not line:
        return
    
    # 解析日志行
    match_result = NGINX_LOG_REGEX.match(line)
    if not match_result:
        print(f"跳过无效日志行: {line}")
        return
    
    log_data = match_result.groupdict()
    # 转换Nginx时间格式为MySQL支持的datetime
    try:
        time_local = datetime.strptime(log_data['time_local'], '%d/%b/%Y:%H:%M:%S %z')
        log_data['time_local'] = time_local.strftime('%Y-%m-%d %H:%M:%S')
    except ValueError as e:
        print(f"时间格式解析失败: {e}")
        return
    
    # 处理body_bytes_sent为空的情况
    log_data['body_bytes_sent'] = log_data['body_bytes_sent'] if log_data['body_bytes_sent'] != '-' else 0

    # 从连接池获取连接并插入数据
    conn = None
    cursor = None
    try:
        conn = db_pool.connection()
        cursor = conn.cursor()
        insert_sql = """
        INSERT INTO nginx_access_logs (remote_addr, remote_user, time_local, request, status, body_bytes_sent, http_referer, http_user_agent)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        cursor.execute(insert_sql, (
            log_data['remote_addr'],
            log_data['remote_user'],
            log_data['time_local'],
            log_data['request'],
            log_data['status'],
            log_data['body_bytes_sent'],
            log_data['http_referer'],
            log_data['http_user_agent']
        ))
        conn.commit()
    except Exception as e:
        print(f"插入数据失败: {e}")
        if conn:
            conn.rollback()
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

def process_log_file(log_file_path, worker_count=10):
    """批量处理日志文件"""
    # 创建协程池，控制并发数
    worker_pool = pool.Pool(worker_count)
    
    # 逐行读取日志（大文件推荐逐行读，避免内存溢出）
    with open(log_file_path, 'r', encoding='utf-8', errors='ignore') as log_file:
        for line in log_file:
            # 提交日志行到协程池处理
            worker_pool.spawn(parse_and_insert_log, line)
    
    # 等待所有协程任务完成
    worker_pool.join()

if __name__ == '__main__':
    # 替换成你的Nginx日志文件路径
    process_log_file('/var/log/nginx/access.log')

四、额外优化建议 #

• 批量插入：如果日志量极大，可以攒几十条甚至上百条日志再批量插入，比单条插入效率高很多，减少与MySQL的交互次数。
• 错误日志收集：可以把解析失败或插入失败的日志行写入单独的文件，后续排查问题。
• 协程池大小：根据你的服务器性能和MySQL配置调整worker_count和连接池大小，避免并发过高导致资源耗尽。
• 日志格式适配：务必根据你实际的Nginx日志格式修改正则表达式，否则会解析失败。

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