如何写实时统计访问流量脚本

wen 实用脚本 2026-06-27 27

从零搭建高效监控系统

目录导读

为什么需要实时流量统计脚本？
核心概念与原理
编写流量统计脚本的6个关键步骤
实战案例：用Python实现Nginx日志实时解析
常见问题问答（FAQ）
脚本优化与性能调优技巧
总结与推荐工具

为什么需要实时流量统计脚本？

在网站运营中，实时掌握访问流量是优化用户体验、发现异常流量、调整服务器资源的核心能力，传统统计工具（如Google Analytics）通常有延迟（5-30分钟），而自建实时脚本可以做到秒级响应,适用于以下场景：

电商大促：监控瞬时并发量，防止服务器过载
安全检测：识别DDoS攻击或爬虫异常
资源弹性伸缩：根据流量自动增减服务器实例
运营决策：即时查看推广活动效果

核心概念与原理

1 流量统计的三大指标

PV（页面浏览量）：用户每次刷新或切换页面算一次
UV（独立访客数）：基于IP或Cookie去重
QPS（每秒查询率）：服务器每秒处理的请求数

2 数据来源

Web服务器日志：Nginx/Apache的access.log，每行记录一次HTTP请求
反向代理日志：如HAProxy、CDN的访问记录
网络抓包：tcpdump捕获网卡数据包（较复杂）

3 实时统计的工作流

日志 -> 读取 -> 解析 -> 聚合（按时间/IP/URL）-> 存储（内存/时序数据库）-> 展示（图表/告警）

编写流量统计脚本的6个关键步骤

步骤1：选择编程语言

推荐 Python（生态完善）或 Go（高性能）,本文以Python为例。

步骤2：确定日志格式

Nginx默认格式：

$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"

示例：

168.1.1 - - [15/Oct/2023:10:30:45 +0000] "GET /index.html HTTP/1.1" 200 1234 "-" "Mozilla/5.0"

步骤3：设计统计维度和窗口

滑动时间窗口：例如每5秒统计一次前60秒的PV
关键指标：PV、UV、Top URL、状态码分布、平均响应时间

步骤4：实现高效日志读取

避免逐行读取大文件，使用 tail -f 模拟实时流：

import subprocess
def follow_log(file_path):
    f = subprocess.Popen(['tail','-F', file_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    for line in iter(f.stdout.readline, b''):
        yield line.decode('utf-8').strip()

步骤5：解析并聚合数据

使用正则表达式提取字段：

import re
pattern = r'(\S+) - - \[(.+?)\] "(\S+) (\S+) \S+" (\d+) (\d+) "(\S*)" "(.+)"'
def parse_line(line):
    match = re.match(pattern, line)
    if match:
        return {
            'ip': match.group(1),
            'time': match.group(2),
            'method': match.group(3),
            'url': match.group(4),
            'status': int(match.group(5)),
            'bytes': int(match.group(6)),
            'referer': match.group(7),
            'user_agent': match.group(8)
        }

步骤6：存储与输出

开发阶段：输出到控制台或CSV
生产环境：写入InfluxDB（时序数据库）或Redis（高性能缓存）

实战案例：用Python实现Nginx日志实时解析

完整脚本示例

import time
from collections import defaultdict, deque
import threading
class RealTimeTrafficMonitor:
    def __init__(self, log_file, window_seconds=60):
        self.log_file = log_file
        self.window_seconds = window_seconds
        self.traffic_queue = deque()
        self.ip_count = defaultdict(int)
        self.url_count = defaultdict(int)
        self.lock = threading.Lock()
    def process_log_line(self, line):
        parsed = parse_line(line)
        if not parsed:
            return
        current_time = time.time()
        self.lock.acquire()
        self.traffic_queue.append((current_time, parsed))
        self.ip_count[parsed['ip']] += 1
        self.url_count[parsed['url']] += 1
        # 移除窗口外的数据
        while self.traffic_queue and current_time - self.traffic_queue[0][0] > self.window_seconds:
            _, old = self.traffic_queue.popleft()
            self.ip_count[old['ip']] -= 1
            if self.ip_count[old['ip']] == 0:
                del self.ip_count[old['ip']]
            self.url_count[old['url']] -= 1
        self.lock.release()
    def get_stats(self):
        self.lock.acquire()
        total_requests = len(self.traffic_queue)
        unique_ips = len(self.ip_count)
        top_urls = sorted(self.url_count.items(), key=lambda x: x[1], reverse=True)[:10]
        qps = total_requests / self.window_seconds
        self.lock.release()
        return {
            'PV': total_requests,
            'UV': unique_ips,
            'QPS': round(qps, 2),
            'Top URLs': top_urls
        }
    def start(self):
        for line in follow_log(self.log_file):
            self.process_log_line(line)
# 使用示例
if __name__ == "__main__":
    monitor = RealTimeTrafficMonitor('/var/log/nginx/access.log', window_seconds=60)
    # 启动数据输出线程
    def display():
        while True:
            time.sleep(5)
            print(monitor.get_stats())
    threading.Thread(target=display, daemon=True).start()
    monitor.start()

运行效果

每5秒输出一次最近60秒的统计：

{'PV': 1523, 'UV': 487, 'QPS': 25.38, 'Top URLs': [('/index.html', 230), ('/api/product', 189), ...]}

常见问题问答（FAQ）

Q1：如何处理百万级QPS的日志？

答：使用Go语言编写，或采用流式处理框架如Flink；避免单进程，可部署多个实例通过消息队列（Kafka）分发。

Q2：UV去重使用IP不准确怎么办？

答：升级为基于Cookie+IP的混合去重，或在日志中加入user_id字段。

Q3：如何防止脚本内存溢出？

答：设置最大缓存行数（如10万条），超限时强制清理；使用LRU缓存策略。

Q4：如何将统计结果实时展示在网页上？

答：脚本将数据推送到WebSocket，前端用Chart.js或ECharts渲染折线图、仪表盘。

Q5：能否统计每个URL的平均响应时间？

答：在解析时增加$request_time字段（需在Nginx配置中开启）,聚合时计算平均值。

脚本优化与性能调优技巧

1 使用异步IO

用 asyncio 替代阻塞读取,适合IO密集型场景：

import asyncio
async def tail_log(file_path):
    proc = await asyncio.create_subprocess_exec('tail', '-F', file_path, stdout=asyncio.subprocess.PIPE)
    while True:
        line = await proc.stdout.readline()
        if not line:
            break
        yield line.decode('utf-8').rstrip()