Python + Linux 构建高可用数据采集服务

3.3 采集器核心逻辑 (app/collector.py)

import requests
import json
import logging
from datetime import datetime

class DataCollector:
    def __init__(self, shutdown_event):
        self.shutdown_event = shutdown_event
        self.config = {
            'api_url': 'http://httpbin.org/delay/2', # 模拟一个延迟 API
            'data_file': '/var/data/collected_data.jsonl',
            'collection_interval': 30 # 采集间隔，秒
        }
        self.session = requests.Session() # 简单的请求重试配置
        self.session.mount('http://', requests.adapters.HTTPAdapter(max_retries=2))
        logging.info(f"采集器初始化完成，目标 API: {self.config['api_url']}")

    def run_one_cycle(self):
        """执行一个完整的采集、处理、存储周期"""
        if self.shutdown_event.is_set():
            return
        logging.debug("开始新的采集周期")
        try:
            # 1. 采集
            response = self.session.get(self.config['api_url'], timeout=5)
            response.raise_for_status()
            data = response.json()

            # 2. 处理（这里可以添加数据清洗、转换逻辑）
            processed_data = {
                'timestamp': datetime.utcnow().isoformat(),
                'origin': data.get('origin'),
                'url': data.get('url')
            }

            # 3. 存储（追加写入文件）
            with open(self.config['data_file'], 'a') as f:
                f.write(json.dumps(processed_data) + '\n')
            logging.info(f"数据采集并存储成功。来源 IP: {processed_data.get('origin')}")
        except requests.exceptions.RequestException as e:
            logging.warning(f"网络请求失败：{e}")
        except (json.JSONDecodeError, IOError) as e:
            logging.error(f"数据处理或存储失败：{e}", exc_info=True)
        except Exception as e:
            logging.error(f"采集周期内发生未知错误：{e}", exc_info=True)
            raise # 将未知异常抛给主循环处理

    def cleanup(self):
        """服务关闭前的清理工作"""
        logging.info("正在清理采集器资源...")
        self.session.close()
        logging.debug("HTTP 会话已关闭。")

3.4 日志配置模块 (app/logger.py)

import logging
import os
from logging.handlers import RotatingFileHandler

def setup_logging(log_dir='/var/log/data_collector', log_level=logging.INFO):
    """配置日志，支持轮转和控制台输出"""
    os.makedirs(log_dir, exist_ok=True)
    log_file = os.path.join(log_dir, 'collector.log')

    # 格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    # 文件处理器 - 轮转，每个文件 10MB，最多保留 5 个备份
    file_handler = RotatingFileHandler(
        log_file, maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
    )
    file_handler.setFormatter(formatter)

    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)

    # 获取根日志记录器并配置
    root_logger = logging.getLogger()
    root_logger.setLevel(log_level)

    # 避免重复添加处理器
    if not root_logger.handlers:
        root_logger.addHandler(file_handler)
        root_logger.addHandler(console_handler)

    # 关闭过于冗长的第三方库日志
    logging.getLogger('urllib3').setLevel(logging.WARNING)

4. systemd 服务单元文件

这是将我们的 Python 脚本变成系统服务的关键。在 /etc/systemd/system/ 下创建 data-collector.service 文件。

[Unit]
Description=High Availability Data Collector Service
After=network.target # 确保在网络就绪后启动
Requires=network.target

[Service]
Type=simple # 重点：指定运行的用户和组，遵循权限最小化原则
User=datauser
Group=datauser

# 重点：设置工作目录和 Python 路径
WorkingDirectory=/opt/data_collector
# 如果使用虚拟环境，在此指定解释器路径
# ExecStart=/opt/venv/data_collector/bin/python /opt/data_collector/main.py
ExecStart=/usr/bin/python3 /opt/data_collector/main.py

# 优雅停止与重启策略
KillSignal=SIGTERM
TimeoutStopSec=30
Restart=on-failure
RestartSec=10

# 防止日志刷屏，如果 10 秒内重启超过 5 次，则放弃
StartLimitIntervalSec=60
StartLimitBurst=5

# 资源限制（可选，防止程序异常占用过多资源）
# LimitNOFILE=65536
# LimitNPROC=2048

# 标准输出和错误输出重定向到 systemd journal，便于用 journalctl 查看
StandardOutput=journal
StandardError=journal

# 环境变量（如果需要）
# Environment="API_KEY=your_key"

[Install]
WantedBy=multi-user.target

创建专用用户并部署代码：

# 创建无登录权限的系统用户
sudo useradd -r -s /bin/false datauser

# 创建必要的目录并设置权限
sudo mkdir -p /opt/data_collector /var/log/data_collector /var/data
sudo chown -R datauser:datauser /opt/data_collector /var/log/data_collector /var/data

# 将你的项目代码拷贝到 /opt/data_collector

# 安装依赖
sudo -u datauser python3 -m pip install -r /opt/data_collector/requirements.txt

管理服务：

# 重载 systemd 配置
sudo systemctl daemon-reload

# 启动服务
sudo systemctl start data-collector

# 设置开机自启
sudo systemctl enable data-collector

# 查看状态
sudo systemctl status data-collector

# 查看日志（非常强大！）
sudo journalctl -u data-collector -f

5. 生产环境避坑指南

即使代码和 service 文件都写好了，在实际部署时还可能遇到以下问题：

1. 权限问题：切忌用 root 运行你的服务。一定要像上面那样创建专用用户，并只赋予它必要的目录读写权限。这是安全的基本要求。
2. 路径硬编码：代码和配置文件中不要出现 /home/yourname/project/ 这样的绝对路径。使用相对路径（基于 WorkingDirectory）或从配置文件中读取。systemd 的 WorkingDirectory 指令很好地解决了这个问题。
3. 异常未捕获：主循环最外层的 try-except 至关重要，它能防止某个未处理的异常导致整个服务进程崩溃。确保记录详细的错误信息（exc_info=True）以便排查。
4. 资源泄漏：像 requests.Session、数据库连接池这类资源，一定要在 cleanup 或 __del__ 方法中显式关闭。systemd 的 TimeoutStopSec 给了你一个清理的时间窗口。
5. 冷启动延迟：如果服务启动时需要连接数据库或远程配置中心，可能因网络问题超时。可以在 [Service] 部分配置 RestartSec 和 StartLimitIntervalSec 来避免短时间内频繁重启。对于依赖其他服务的，可以用 [Unit] 部分的 After 和 Requires 来管理启动顺序。
6. 日志管理：我们代码中使用了 RotatingFileHandler，但 systemd 自己的 journald 已经提供了强大的日志管理功能（自动轮转、压缩、按时间查询）。将日志输出到 journal（如我们配置的 StandardOutput=journal）通常是更推荐的做法，可以用 journalctl --vacuum-size=500M 来清理旧日志。

6. 进阶思考：让服务更'可观测'

一个高可用的服务，除了稳定运行，还需要可观测。你可以考虑为你的毕设服务增加以下模块：

• 健康检查端点：在服务内部启动一个简单的 HTTP 服务器（比如用 http.server 或 Flask 写一个轻量级子进程），暴露一个 /health 接口。返回服务状态、最近一次采集时间等。这样，外部监控系统（如 systemd 的 ExecStartPost 配合 curl，或 Prometheus）可以定期探测。
• 指标上报：在 DataCollector 中增加计数器，统计采集成功/失败次数、数据条数等。使用 Prometheus 客户端库暴露这些指标，或定期写入日志文件供 ELK 分析。
• 配置热更新：将配置（如 API URL、采集间隔）放入单独的 config.json 文件。服务可以定期检查文件修改时间，或者监听 SIGHUP 信号，实现不重启服务即可重载配置。

通过以上步骤，我们就把一个简单的 Python 脚本，包装成了一个具备生产环境雏形的系统服务。它拥有优雅启停、自动重启、资源管控、集中日志等特性。

动手改造一下你的毕设项目吧！ 试着将你的核心逻辑嵌入到这个框架中，替换掉原来的 nohup 或 cron。这个过程本身，就是对软件工程和运维知识的一次极佳实践。当你看到自己的服务能被 systemctl 优雅地管理，能被 journalctl 清晰地追踪日志时，那种成就感会比单纯实现功能更强烈。你的毕设项目，也会因此脱颖而出。