用 Python 造一个轻量级流处理引擎顺便把 Git、Docker、CI/CD 全串起来前言你是否有过这样的需求统计过去 5 秒内 API 的请求次数、监控传感器数据的突变、或者对直播间的弹幕进行限流这些场景都离不开实时数据流处理。而流处理的核心往往是一个精巧的滑动窗口算法。今天我们不依赖 Flink、Spark Streaming 这类重型框架而是从零开始用 Python 实现一个基于滑动窗口的实时计数器然后把它容器化、纳入 Git 版本管理、配置 CI/CD 流水线——最终跑在自己的 Docker 环境中。整个过程会涉及数据结构与算法双端队列 哈希表Python 异步/多线程模拟数据流Docker Docker ComposeGit hooks 自动化代码检查GitHub Actions 持续集成代码已上传 GitHub 仓库示例链接你也可以跟着文章一步步搭建。一、算法选型为什么是滑动窗口假设我们要统计“最近 10 秒内某个用户 ID 的请求次数”。最简单的办法是每来一条请求就存下时间戳然后每次查询时遍历所有时间戳找出在 [now-10, now] 范围内的。复杂度 O(N)窗口越大越慢。滑动窗口的优化思路是只保留窗口内的有效数据并且利用双端队列维持时间顺序使插入和删除都能达到 O(1)。下图示意了窗口随着时间向前移动的过程text时间轴 → [old] [old] [valid] [valid] [new] [new] ↑ 窗口左边界当前时间 - 窗口大小 ↑ 当前时间每次新数据到来时清理队头已经超出窗口的时间戳在队尾插入新数据更新计数若需要按 key 统计则配合字典二、核心实现滑动窗口计数器我们实现一个泛用型SlidingWindowCounter支持按任意 key 分组统计比如 user_id、ip 地址。底层使用defaultdict(deque)。python# sliding_window.py from collections import defaultdict, deque import time from typing import Optional class SlidingWindowCounter: 滑动窗口计数器支持多 key 和自动过期清理 def __init__(self, window_size_sec: float): :param window_size_sec: 窗口大小秒例如 10.0 表示最近10秒 self.window_size window_size_sec # key - deque of (timestamp, count增量为了简单每个事件单独存时间戳) self._buckets defaultdict(deque) # key: deque[timestamp] self._counts defaultdict(int) # 缓存当前窗口内的总数 def add(self, key: str, timestamp: Optional[float] None): 为某个 key 添加一次事件 if timestamp is None: timestamp time.time() # 先清理旧数据 self._evict(key, timestamp) # 插入新事件 self._buckets[key].append(timestamp) self._counts[key] 1 def get_count(self, key: str, now: Optional[float] None) - int: 获取当前 key 在窗口内的总次数 if now is None: now time.time() self._evict(key, now) return self._counts.get(key, 0) def _evict(self, key: str, current_time: float): 移除超出窗口的时间戳 dq self._buckets.get(key) if not dq: return boundary current_time - self.window_size while dq and dq[0] boundary: dq.popleft() self._counts[key] - 1 # 如果队列空删除 key 以释放内存 if not dq: del self._buckets[key] if key in self._counts: del self._counts[key] def get_all_counts(self, now: Optional[float] None) - dict: 返回所有 key 在当前窗口内的统计结果 if now is None: now time.time() # 先全局清理一次为了性能也可懒清理 for key in list(self._buckets.keys()): self._evict(key, now) return dict(self._counts)测试一下python# test_sliding_window.py import time counter SlidingWindowCounter(window_size_sec3.0) counter.add(user_1) time.sleep(1) counter.add(user_1) time.sleep(1) counter.add(user_2) print(counter.get_count(user_1)) # 预期 2 (最近3秒内) time.sleep(1.5) print(counter.get_count(user_1)) # 预期 0 或 1因为第1秒的事件已经超出3秒窗口输出符合预期。这个实现非常轻量并且支持任意多 key。三、模拟实时数据流有了计数器我们还需要一个“数据源”来不断产生事件。这里用 Python 的threading和queue模拟一个多生产者单消费者的流处理管道。python# stream_simulator.py import threading import queue import time import random from sliding_window import SlidingWindowCounter # 全局计数器 counter SlidingWindowCounter(window_size_sec5.0) # 事件队列 event_queue queue.Queue() def producer(producer_id: int, stop_event: threading.Event): 模拟不同来源的事件生成器 while not stop_event.is_set(): # 随机生成 user_id (1~10) user_id fuser_{random.randint(1,10)} event_queue.put((click, user_id)) time.sleep(random.uniform(0.1, 0.5)) def consumer(stop_event: threading.Event): 消费事件更新滑动窗口并打印统计 last_report_time time.time() while not stop_event.is_set(): try: event event_queue.get(timeout0.5) event_type, user_id event counter.add(user_id) # 每秒输出一次统计结果 now time.time() if now - last_report_time 1.0: stats counter.get_all_counts() print(f\n[{time.strftime(%H:%M:%S)}] 窗口内各用户点击次数:) for uid, cnt in sorted(stats.items(), keylambda x: -x[1])[:5]: print(f {uid}: {cnt}) last_report_time now except queue.Empty: continue if __name__ __main__: stop threading.Event() producers [] for i in range(3): t threading.Thread(targetproducer, args(i, stop)) t.start() producers.append(t) consumer_thread threading.Thread(targetconsumer, args(stop,)) consumer_thread.start() try: # 运行30秒 time.sleep(30) finally: stop.set() for t in producers: t.join() consumer_thread.join()运行这段代码你会看到终端每秒打印出最近 5 秒内各个用户的点击次数排名——一个简易的实时热门用户看板就诞生了。四、用 Docker 封装与编排真实生产环境不可能手动python stream_simulator.py我们需要容器化。4.1 编写 Dockerfiledockerfile# Dockerfile FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [python, stream_simulator.py]requirements.txt仅需一行其实标准库就够但为了扩展性text# 目前无依赖保留文件方便以后4.2 编写 docker-compose.yml为了方便扩展多个生产者或消费者我们用 compose 启动一个独立的消费者 多个生产者但上面的代码已经在一个进程内完成了。为了展示更真实的流处理架构我们可以拆成两个服务producer和consumer通过 Redis 或 Kafka 通信。但为了简洁这里只演示单服务但用 compose 设置资源限制和环境变量。yamlversion: 3.8 services: stream-processor: build: . environment: - WINDOW_SIZE_SEC10 - REPORT_INTERVAL_SEC2 restart: unless-stopped logging: driver: json-file options: max-size: 10m修改stream_simulator.py读取环境变量使窗口大小可配置pythonimport os WINDOW_SIZE float(os.getenv(WINDOW_SIZE_SEC, 5.0)) counter SlidingWindowCounter(window_size_secWINDOW_SIZE)4.3 构建并运行bashdocker build -t sliding-window-demo . docker run --rm sliding-window-demo # 或用 compose docker-compose up看到日志滚动说明容器化成功。五、Git 与 Git Hooks让代码更健壮在团队协作中每次提交前自动运行代码格式化、静态检查是非常好的实践。我们用pre-commit框架来管理 Git hooks。5.1 安装 pre-commitbashpip install pre-commit5.2 创建.pre-commit-config.yamlyamlrepos: - repo: https://github.com/psf/black rev: 23.10.0 hooks: - id: black language_version: python3 - repo: https://github.com/pycqa/flake8 rev: 6.1.0 hooks: - id: flake8 args: [--max-line-length88] - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.4.0 hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml5.3 安装 hooksbashpre-commit install现在每次git commit时black 会自动格式化代码flake8 会检查语法风格。不符合规范的提交会被阻止直到你修复。5.4 关联远程仓库bashgit init git add . git commit -m feat: sliding window counter with stream simulator git remote add origin https://github.com/yourname/sliding-window-stream.git git push -u origin main六、CI/CDGitHub Actions 自动测试与镜像构建提交代码后我们希望 GitHub 自动运行单元测试并通过 Docker Hub 或 GitHub Container Registry 构建镜像。下面是一个完整的 GitHub Actions 工作流yaml# .github/workflows/ci.yml name: CI/CD Pipeline on: push: branches: [ main ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv3 - name: Set up Python uses: actions/setup-pythonv4 with: python-version: 3.10 - name: Install dependencies run: pip install pytest - name: Run tests run: pytest tests/ # 需要写几个单元测试 build-and-push: needs: test if: github.event_name push github.ref refs/heads/main runs-on: ubuntu-latest steps: - uses: actions/checkoutv3 - name: Log in to Docker Hub uses: docker/login-actionv2 with: username: ${{ secrets.DOCKER_USERNAME }} password: ${{ secrets.DOCKER_PASSWORD }} - name: Build and push Docker image uses: docker/build-push-actionv4 with: context: . push: true tags: yourusername/sliding-window-demo:latest别忘了在 GitHub 仓库的 Settings - Secrets 中添加DOCKER_USERNAME和DOCKER_PASSWORD。补充单元测试创建tests/test_sliding_window.pypythonimport time from sliding_window import SlidingWindowCounter def test_basic_add_and_get(): c SlidingWindowCounter(2.0) c.add(a) assert c.get_count(a) 1 time.sleep(1) c.add(a) assert c.get_count(a) 2 time.sleep(1.5) assert c.get_count(a) 0 def test_multiple_keys(): c SlidingWindowCounter(1.0) c.add(x) c.add(y) c.add(x) assert c.get_count(x) 2 assert c.get_count(y) 1现在每次git pushGitHub 都会自动运行测试通过后构建镜像并推送到 Docker Hub。七、更进一步接入大数据生态如果你觉得上面的实现太“玩具”可以很容易地替换消息队列和存储用Kafka作为事件总线生产者推送消息到 topic消费者使用kafka-python消费。用Redis存储滑动窗口Redis 的ZSET天然适合按时间戳排序使用ZREMRANGEBYSCORE清理过期数据。用Flink或Spark Structured Streaming做分布式滑动窗口——但那就背离了我们“轻量级”的初衷。八、总结与思考在这篇文章中我们完成了一整套从算法到部署的实时流处理小项目手写滑动窗口算法时间复杂度 O(1) 维护窗口。多线程模拟数据流展示实时统计效果。Docker 容器化并支持环境变量配置。Git hooks强制代码规范。GitHub Actions自动化测试与镜像构建。这套模式完全可以应用在实际的小型监控、限流、实时看板等场景。更重要的是你学会了如何将数据结构与算法落地成工程并用现代开发工具链Git、Docker、CI/CD让它变得可靠、可交付。如果你喜欢这种“从零开始造轮子再装到生产流水线”的玩法欢迎在评论区交流你的实时数据处理经验。别忘了给仓库点个 ⭐️