经过九讲的开发我们拥有了一套完整的分布式任务调度系统。最后一讲我们将把它部署到生产环境并进行实战演练。从容器化到Kubernetes编排从压测到调优完成系统的最后一公里。一、部署架构总览1.1 生产部署拓扑┌─────────────────────────────────────────────────────────────┐ │ Kubernetes 集群 │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Ingress │ │ Ingress │ │ Ingress │ │ │ │ Controller │ │ Controller │ │ Controller │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ │ │ Scheduler │ │ Scheduler │ │ Scheduler │ │ │ │ (Leader) │ │ (Follower) │ │ (Follower) │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ │ │ Worker │ │ Worker │ │ Worker │ │ │ │ Pool │ │ Pool │ │ Pool │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ PostgreSQL │ │ etcd │ │ Prometheus │ │ │ │ (主从) │ │ (集群) │ │ Grafana │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────┘1.2 服务组件清单┌─────────────────────────────────────────────────────────────┐ │ 组件 镜像 端口 副本数 │ ├─────────────────────────────────────────────────────────────┤ │ scheduler-api scheduler:latest 8080 3 │ │ scheduler-worker worker:latest 9001 5 │ │ postgres postgres:15 5432 2 (主从) │ │ etcd etcd:3.5 2379 3 │ │ prometheus prometheus:latest 9090 1 │ │ grafana grafana:latest 3000 1 │ └─────────────────────────────────────────────────────────────┘二、Docker容器化2.1 Dockerfile# Dockerfile FROM python:3.11-slim AS builder WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ libpq-dev \ rm -rf /var/lib/apt/lists/* # 安装Python依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 第二阶段运行镜像 FROM python:3.11-slim WORKDIR /app # 复制依赖 COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY --frombuilder /usr/local/bin /usr/local/bin # 复制应用代码 COPY . . # 健康检查 HEALTHCHECK --interval30s --timeout3s --start-period5s --retries3 \ CMD python -c import urllib.request; urllib.request.urlopen(http://localhost:8080/health) EXPOSE 8080 CMD [python, main.py]2.2 docker-compose.yml# docker-compose.yml version: 3.8 services: # etcd集群 etcd: image: bitnami/etcd:3.5 environment: - ALLOW_NONE_AUTHENTICATIONyes - ETCD_NAMEetcd0 - ETCD_INITIAL_CLUSTERetcd0http://etcd:2380 - ETCD_INITIAL_ADVERTISE_PEER_URLShttp://etcd:2380 - ETCD_ADVERTISE_CLIENT_URLShttp://etcd:2379 - ETCD_LISTEN_CLIENT_URLShttp://0.0.0.0:2379 - ETCD_LISTEN_PEER_URLShttp://0.0.0.0:2380 ports: - 2379:2379 volumes: - etcd_data:/bitnami/etcd/data networks: - scheduler-net # PostgreSQL postgres: image: postgres:15 environment: POSTGRES_DB: scheduler POSTGRES_USER: scheduler POSTGRES_PASSWORD: scheduler_pass ports: - 5432:5432 volumes: - postgres_data:/var/lib/postgresql/data - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql networks: - scheduler-net healthcheck: test: [CMD-SHELL, pg_isready -U scheduler] interval: 10s timeout: 5s retries: 5 # 调度器3副本 scheduler: build: . image: scheduler:latest command: python -m scheduler.server --port8080 environment: - NODE_TYPEscheduler - ETCD_HOSTetcd - ETCD_PORT2379 - DB_DSNpostgresql://scheduler:scheduler_passpostgres:5432/scheduler - REDIS_URLredis://redis:6379/0 - LOG_LEVELINFO ports: - 8080 depends_on: postgres: condition: service_healthy etcd: condition: service_started deploy: replicas: 3 resources: limits: cpus: 1 memory: 512M networks: - scheduler-net # Worker节点5副本 worker: build: . image: scheduler:latest command: python -m scheduler.worker --port9001 environment: - NODE_TYPEworker - WORKER_IDworker-{{.Task.Slot}} - ETCD_HOSTetcd - ETCD_PORT2379 - MAX_CONCURRENCY10 - LOG_LEVELINFO ports: - 9001 depends_on: - scheduler deploy: replicas: 5 resources: limits: cpus: 2 memory: 1G networks: - scheduler-net # Prometheus监控 prometheus: image: prom/prometheus:latest volumes: - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml - prometheus_data:/prometheus ports: - 9090:9090 networks: - scheduler-net # Grafana可视化 grafana: image: grafana/grafana:latest environment: - GF_SECURITY_ADMIN_PASSWORDadmin - GF_INSTALL_PLUGINSgrafana-piechart-panel volumes: - ./grafana/dashboards:/etc/grafana/provisioning/dashboards - ./grafana/datasources:/etc/grafana/provisioning/datasources - grafana_data:/var/lib/grafana ports: - 3000:3000 depends_on: - prometheus networks: - scheduler-net # Redis缓存 redis: image: redis:7-alpine ports: - 6379:6379 volumes: - redis_data:/data networks: - scheduler-net volumes: etcd_data: postgres_data: prometheus_data: grafana_data: redis_data: networks: scheduler-net: driver: bridge2.3 配置文件# config/production.yaml server: host: 0.0.0.0 port: 8080 workers: 10 database: dsn: ${DB_DSN} pool_min: 5 pool_max: 20 pool_timeout: 30 etcd: hosts: [etcd:2379] timeout: 5 ttl: 10 redis: url: ${REDIS_URL} pool_size: 10 scheduler: tick_ms: 100 max_queue_size: 10000 dispatch_strategy: least_loaded worker: max_concurrency: 10 heartbeat_interval: 5 task_timeout_ms: 30000 monitoring: metrics_port: 9100 tracing_enabled: true tracing_endpoint: http://jaeger:14268/api/traces logging: level: ${LOG_LEVEL:-INFO} format: json output: stdout三、Kubernetes部署3.1 Helm Chart# helm/scheduler/Chart.yaml apiVersion: v2 name: distributed-scheduler description: A distributed task scheduler version: 1.0.0 appVersion: 1.0.03.2 Values配置# helm/scheduler/values.yaml replicaCount: 3 image: repository: scheduler tag: latest pullPolicy: Always config: logLevel: INFO maxConcurrency: 10 taskTimeoutMs: 30000 resources: scheduler: limits: cpu: 1 memory: 512Mi requests: cpu: 500m memory: 256Mi worker: limits: cpu: 2 memory: 1Gi requests: cpu: 1 memory: 512Mi persistence: enabled: true storageClass: standard size: 10Gi ingress: enabled: true host: scheduler.example.com tls: true monitoring: prometheus: enabled: true scrapeInterval: 15s grafana: enabled: true adminPassword: admin3.3 Deployment模板# helm/scheduler/templates/scheduler-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: {{ .Values.name }}-scheduler labels: app: scheduler spec: replicas: {{ .Values.replicaCount }} selector: matchLabels: app: scheduler template: metadata: labels: app: scheduler annotations: prometheus.io/scrape: true prometheus.io/port: 9100 spec: containers: - name: scheduler image: {{ .Values.image.repository }}:{{ .Values.image.tag }} command: [python, -m, scheduler.server] args: [--port8080] ports: - containerPort: 8080 name: http - containerPort: 9100 name: metrics env: - name: NODE_TYPE value: scheduler - name: ETCD_HOST value: etcd-cluster - name: DB_DSN valueFrom: secretKeyRef: name: db-secret key: dsn - name: LOG_LEVEL value: {{ .Values.config.logLevel }} resources: {{- toYaml .Values.resources.scheduler | nindent 10 }} livenessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: 8080 initialDelaySeconds: 5 periodSeconds: 5 --- apiVersion: v1 kind: Service metadata: name: scheduler-service spec: selector: app: scheduler ports: - port: 8080 targetPort: 8080 name: http - port: 9100 targetPort: 9100 name: metrics type: ClusterIP3.4 Worker StatefulSet# helm/scheduler/templates/worker-statefulset.yaml apiVersion: apps/v1 kind: StatefulSet metadata: name: {{ .Values.name }}-worker spec: serviceName: worker-headless replicas: 5 selector: matchLabels: app: worker template: metadata: labels: app: worker spec: containers: - name: worker image: {{ .Values.image.repository }}:{{ .Values.image.tag }} command: [python, -m, scheduler.worker] args: [--port9001] ports: - containerPort: 9001 name: grpc env: - name: NODE_TYPE value: worker - name: WORKER_ID valueFrom: fieldRef: fieldPath: metadata.name - name: ETCD_HOST value: etcd-cluster - name: MAX_CONCURRENCY value: {{ .Values.config.maxConcurrency }} resources: {{- toYaml .Values.resources.worker | nindent 12 }} --- apiVersion: v1 kind: Service metadata: name: worker-headless spec: clusterIP: None selector: app: worker ports: - port: 9001 targetPort: 90013.5 ConfigMap与Secret# helm/scheduler/templates/configmap.yaml apiVersion: v1 kind: ConfigMap metadata: name: scheduler-config data: config.yaml: | server: host: 0.0.0.0 port: 8080 scheduler: tick_ms: 100 max_queue_size: 10000 monitoring: metrics_port: 9100 --- apiVersion: v1 kind: Secret metadata: name: db-secret type: Opaque stringData: dsn: postgresql://scheduler:scheduler_passpostgres:5432/scheduler四、CI/CD流水线4.1 GitHub Actions# .github/workflows/deploy.yml name: Deploy Scheduler on: push: branches: [main] pull_request: branches: [main] env: REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv3 - name: Set up Python uses: actions/setup-pythonv4 with: python-version: 3.11 - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest pytest-asyncio - name: Run tests run: | pytest tests/ -v --cov./ --cov-reportxml - name: Upload coverage uses: codecov/codecov-actionv3 build-and-push: needs: test runs-on: ubuntu-latest if: github.event_name push steps: - uses: actions/checkoutv3 - name: Set up Docker Buildx uses: docker/setup-buildx-actionv2 - name: Login to Container Registry uses: docker/login-actionv2 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Extract metadata id: meta uses: docker/metadata-actionv4 with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} tags: | typesha,formatlong typeref,eventbranch typesemver,pattern{{version}} - name: Build and push uses: docker/build-push-actionv4 with: context: . push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} cache-from: typegha cache-to: typegha,modemax deploy: needs: build-and-push runs-on: ubuntu-latest if: github.ref refs/heads/main steps: - name: Configure kubectl uses: azure/setup-kubectlv3 with: version: latest - name: Set up Kubeconfig run: | mkdir -p $HOME/.kube echo ${{ secrets.KUBECONFIG }} $HOME/.kube/config - name: Deploy to Kubernetes run: | helm upgrade --install scheduler ./helm/scheduler \ --set image.tag${{ github.sha }} \ --namespace scheduler-system \ --create-namespace \ --wait \ --timeout 10m - name: Verify deployment run: | kubectl rollout status deployment/scheduler-scheduler \ -n scheduler-system --timeout5m kubectl rollout status statefulset/scheduler-worker \ -n scheduler-system --timeout5m五、压测与调优5.1 压测脚本# scripts/load_test.py 负载测试脚本 模拟大量任务提交和执行测试系统性能。 import asyncio import aiohttp import time import random import statistics from dataclasses import dataclass from typing import List dataclass class LoadTestConfig: 压测配置 scheduler_url: str http://localhost:8080 total_tasks: int 10000 concurrency: int 50 task_types: List[str] None def __post_init__(self): if self.task_types is None: self.task_types [sleep, echo, compute] class LoadTester: 负载测试器 def __init__(self, config: LoadTestConfig): self.config config self.results [] self.errors 0 async def submit_task(self, session: aiohttp.ClientSession, task_id: int) - float: 提交单个任务 task_type random.choice(self.config.task_types) payload { name: fload-test-{task_id}, task_type: task_type, handler: fhandlers.{task_type}, params: { duration: random.uniform(0.1, 0.5), task_id: task_id }, priority: random.choice([LOW, MEDIUM, HIGH, CRITICAL]), schedule_time: time.time() random.uniform(0, 0.1) } start time.perf_counter() try: async with session.post( f{self.config.scheduler_url}/api/tasks, jsonpayload, timeoutaiohttp.ClientTimeout(total5) ) as resp: if resp.status 200: latency time.perf_counter() - start return latency else: self.errors 1 return None except Exception as e: self.errors 1 return None async def run(self): 运行压测 print(fStarting load test: {self.config.total_tasks} tasks, f{self.config.concurrency} concurrency) connector aiohttp.TCPConnector(limitself.config.concurrency) async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for i in range(self.config.total_tasks): tasks.append(self.submit_task(session, i)) # 分批执行 batch_size self.config.concurrency * 10 for i in range(0, len(tasks), batch_size): batch tasks[i:i batch_size] results await asyncio.gather(*batch) self.results.extend([r for r in results if r is not None]) progress (i len(batch)) / len(tasks) * 100 print(f\rProgress: {progress:.1f}% f({len(self.results)} succeeded, f{self.errors} errors), end) print() self.print_report() def print_report(self): 打印报告 if not self.results: print(No successful requests) return latencies sorted(self.results) n len(latencies) print(\n * 70) print( Load Test Report) print( * 70) print(fTotal tasks: {self.config.total_tasks}) print(fSuccessful: {len(self.results)}) print(fErrors: {self.errors}) print(fSuccess rate: {len(self.results)/self.config.total_tasks*100:.1f}%) print(f\nLatency (seconds):) print(f Average: {statistics.mean(latencies)*1000:.2f} ms) print(f P50: {latencies[n//2]*1000:.2f} ms) print(f P90: {latencies[int(n*0.9)]*1000:.2f} ms) print(f P99: {latencies[int(n*0.99)]*1000:.2f} ms) print(f Min: {latencies[0]*1000:.2f} ms) print(f Max: {latencies[-1]*1000:.2f} ms) print(f\nThroughput: {len(self.results)/(latencies[-1]):.0f} req/s) async def main(): config LoadTestConfig( scheduler_urlhttp://localhost:8080, total_tasks5000, concurrency50 ) tester LoadTester(config) await tester.run() if __name__ __main__: asyncio.run(main())5.2 调优指南# 生产调优指南 ## 1. 操作系统调优 ### 1.1 内核参数bash/etc/sysctl.confnet.core.somaxconn 65535net.ipv4.tcp_max_syn_backlog 65535net.ipv4.ip_local_port_range 1024 65000fs.file-max 1000000应用sysctl -p### 1.2 ulimit设置bash/etc/security/limits.confsoft nofile 1000000hard nofile 1000000soft nproc 100000hard nproc 100000## 2. Python调优 ### 2.1 GIL优化python对于CPU密集型任务使用多进程from multiprocessing import Pool或者使用异步线程池import asynciofrom concurrent.futures import ThreadPoolExecutorexecutor ThreadPoolExecutor(max_workers10)loop asyncio.get_event_loop()result await loop.run_in_executor(executor, cpu_intensive_func)### 2.2 内存优化python使用slots减少内存占用class Task:slots [task_id, name, status]使用array代替list存储大量数字from array import arraynumbers array(i, [1, 2, 3, 4, 5])## 3. 数据库调优 ### 3.1 PostgreSQL配置inipostgresql.confmax_connections 200shared_buffers 1GBeffective_cache_size 3GBwork_mem 32MBmaintenance_work_mem 128MBrandom_page_cost 1.1effective_io_concurrency 200wal_buffers 16MB### 3.2 查询优化sql-- 创建复合索引CREATE INDEX idx_tasks_status_schedule ON tasks(status, schedule_time);CREATE INDEX idx_tasks_priority_created ON tasks(priority, created_at DESC);-- 使用覆盖索引CREATE INDEX idx_tasks_covering ON tasks(status, schedule_time) INCLUDE (name, task_type);## 4. 应用调优 ### 4.1 连接池配置pythonpool_config PoolConfig(min_size10, # 最小连接数max_size50, # 最大连接数max_idle_time300, # 最大空闲时间acquire_timeout5 # 获取超时)### 4.2 缓存策略python热点数据缓存cache TwoLevelCache(local_size10000, # 本地缓存10K条local_ttl60, # 本地缓存1分钟remote_ttl600 # Redis缓存10分钟)### 4.3 批量处理python批量写入数据库batch_size 100 # 每批100条flush_interval 0.1 # 100ms刷新一次## 5. Kubernetes调优 ### 5.1 HPA配置yamlapiVersion: autoscaling/v2kind: HorizontalPodAutoscalermetadata:name: scheduler-hpaspec:scaleTargetRef:apiVersion: apps/v1kind: Deploymentname: schedulerminReplicas: 3maxReplicas: 10metrics:type: Resourceresource:name: cputarget:type: UtilizationaverageUtilization: 70type: Podspods:metric:name: scheduler_queue_depthtarget:type: AverageValueaverageValue: 1000### 5.2 PDB配置yamlapiVersion: policy/v1kind: PodDisruptionBudgetmetadata:name: scheduler-pdbspec:minAvailable: 2selector:matchLabels:app: scheduler六、运维手册6.1 日常运维命令#!/bin/bash # scripts/ops.sh # 调度系统运维脚本 # 查看集群状态 alias sched-statuskubectl get all -n scheduler-system alias sched-logskubectl logs -n scheduler-system -l appscheduler --tail100 alias sched-workerskubectl get pods -n scheduler-system -l appworker # 查看Leader function sched-leader() { kubectl exec -n scheduler-system deploy/scheduler-scheduler-0 -- \ python -c from coordination.election import LeaderElection; print(Leader check) } # 滚动更新 function sched-update() { kubectl set image deployment/scheduler-scheduler \ scheduler$1 -n scheduler-system --record kubectl rollout status deployment/scheduler-scheduler -n scheduler-system } # 扩缩容 function sched-scale() { kubectl scale deployment/scheduler-scheduler --replicas$1 -n scheduler-system kubectl scale statefulset/scheduler-worker --replicas$2 -n scheduler-system } # 查看监控 function sched-metrics() { kubectl port-forward -n scheduler-system svc/prometheus 9090:9090 kubectl port-forward -n scheduler-system svc/grafana 3000:3000 echo Prometheus: http://localhost:9090 echo Grafana: http://localhost:3000 (admin/admin) } # 备份数据库 function sched-backup() { kubectl exec -n scheduler-system deploy/postgres -- \ pg_dump -U scheduler scheduler backup_$(date %Y%m%d_%H%M%S).sql } # 查看任务队列 function sched-queue() { kubectl exec -n scheduler-system deploy/scheduler-scheduler-0 -- \ curl -s localhost:8080/api/stats | python -m json.tool }6.2 告警规则# prometheus/alerts.yml groups: - name: scheduler-alerts rules: - alert: SchedulerDown expr: up{jobscheduler} 1 for: 1m labels: severity: critical annotations: summary: Scheduler instance down - alert: HighTaskFailureRate expr: rate(scheduler_tasks_failed_total[5m]) / rate(scheduler_tasks_total[5m]) 0.1 for: 5m labels: severity: warning annotations: summary: High task failure rate - alert: QueueDepthHigh expr: scheduler_queue_depth 10000 for: 2m labels: severity: warning annotations: summary: Task queue depth exceeds 10000 - alert: WorkerOffline expr: scheduler_workers_active 3 for: 1m labels: severity: critical annotations: summary: Less than 3 workers active - alert: HighLatency expr: histogram_quantile(0.99, rate(scheduler_task_duration_seconds_bucket[5m])) 10 for: 5m labels: severity: warning annotations: summary: P99 task latency exceeds 10s七、实战演练7.1 端到端测试# scripts/e2e_test.py 端到端测试 验证整个系统的功能完整性。 import asyncio import aiohttp import time import json class E2ETest: 端到端测试 def __init__(self, base_url: str http://localhost:8080): self.base_url base_url self.session None async def __aenter__(self): self.session aiohttp.ClientSession() return self async def __aexit__(self, *args): await self.session.close() async def test_health(self): 测试健康检查 async with self.session.get(f{self.base_url}/health) as resp: assert resp.status 200 data await resp.json() assert data[status] healthy print(✅ Health check passed) async def test_submit_task(self): 测试提交任务 payload { name: e2e-test-task, task_type: sleep, handler: handlers.sleep, params: {duration: 0.5}, priority: HIGH, schedule_time: time.time() } async with self.session.post( f{self.base_url}/api/tasks, jsonpayload ) as resp: assert resp.status 200 data await resp.json() assert task_id in data print(f✅ Task submitted: {data[task_id]}) return data[task_id] async def test_get_task(self, task_id: str): 测试获取任务 async with self.session.get( f{self.base_url}/api/tasks/{task_id} ) as resp: assert resp.status 200 data await resp.json() assert data[task_id] task_id print(f✅ Task retrieved: {data[name]} ({data[status]})) async def test_list_tasks(self): 测试任务列表 async with self.session.get( f{self.base_url}/api/tasks?limit10 ) as resp: assert resp.status 200 data await resp.json() assert tasks in data print(f✅ Tasks listed: {len(data[tasks])} tasks) async def test_workflow(self): 测试工作流 workflow { name: e2e-workflow, nodes: [ {node_id: extract, name: Extract, type: task}, {node_id: transform, name: Transform, type: task}, {node_id: load, name: Load, type: task} ], edges: [ {source: extract, target: transform}, {source: transform, target: load} ] } async with self.session.post( f{self.base_url}/api/workflows, jsonworkflow ) as resp: assert resp.status 200 data await resp.json() print(f✅ Workflow created: {data[workflow_id]}) async def test_stats(self): 测试统计 async with self.session.get(f{self.base_url}/api/stats) as resp: assert resp.status 200 data await resp.json() print(f✅ Stats retrieved: {json.dumps(data, indent2)}) async def run_all(self): 运行所有测试 print( * 65) print( E2E Test Suite) print( * 65) await self.test_health() task_id await self.test_submit_task() await asyncio.sleep(1) await self.test_get_task(task_id) await self.test_list_tasks() await self.test_workflow() await self.test_stats() print(\n * 65) print(✅ All E2E tests passed!) print( * 65) async def main(): async with E2ETest() as tester: await tester.run_all() if __name__ __main__: asyncio.run(main())八、总结8.1 十讲回顾┌─────────────────────────────────────────────────────────────┐ │ 分布式任务调度系统 · 十讲回顾 │ ├──────┬──────────────────────────────┬───────────────────────┤ │ 讲次 │ 主题 │ 核心产出 │ ├──────┼──────────────────────────────┼───────────────────────┤ │ 01 │ 系统设计与任务模型 │ Task/Job模型 │ │ 02 │ 调度引擎 │ 时间轮优先级队列 │ │ 03 │ 分布式协调 │ 选举锁服务发现 │ │ 04 │ 任务分发与执行 │ Worker分发器 │ │ 05 │ 故障转移与高可用 │ 心跳重试熔断 │ │ 06 │ 持久化与历史 │ 数据库归档 │ │ 07 │ 监控与告警 │ 指标追踪告警 │ │ 08 │ 高级特性 │ CronDAG分片 │ │ 09 │ 性能优化 │ 连接池缓存批处理 │ │ 10 │ 部署与实战 │ DockerK8sCI/CD │ └──────┴──────────────────────────────┴───────────────────────┘8.2 项目结构distributed-scheduler/ ├── scheduler/ # 调度器核心 │ ├── engine.py # 调度引擎 │ ├── dispatcher.py # 任务分发 │ └── orchestrator.py # 编排器 ├── worker/ # 执行器 │ ├── executor.py # 任务执行 │ └── handlers.py # 处理器 ├── coordination/ # 分布式协调 │ ├── election.py # Leader选举 │ ├── lock.py # 分布式锁 │ ├── registry.py # 服务注册 │ └── hash_ring.py # 一致性哈希 ├── storage/ # 持久化 │ ├── models.py # ORM模型 │ ├── repository.py # 数据仓库 │ └── archiver.py # 归档管理 ├── monitor/ # 监控 │ ├── metrics.py # 指标采集 │ ├── tracing.py # 链路追踪 │ ├── alerting.py # 告警引擎 │ └── rules.py # 告警规则 ├── fault_tolerance/ # 高可用 │ ├── heartbeat.py # 心跳检测 │ ├── retry.py # 重试管理 │ ├── failover.py # 故障转移 │ └── circuit_breaker.py # 断路器 ├── advanced/ # 高级特性 │ ├── cron.py # Cron调度 │ ├── dag_workflow.py # DAG工作流 │ └── sharding.py # 动态分片 ├── optimization/ # 性能优化 │ ├── connection_pool.py # 连接池 │ ├── batch_processor.py # 批处理 │ ├── cache.py # 缓存 │ └── object_pool.py # 对象池 ├── config/ # 配置 ├── scripts/ # 脚本 ├── helm/ # K8s部署 ├── tests/ # 测试 └── docs/ # 文档8.3 下一步方向未来演进方向 ┌─────────────────────────────────────────────────────────────┐ │ 1. 多租户支持 │ │ - 租户隔离 │ │ - 配额管理 │ │ - 计费系统 │ │ │ │ 2. 智能调度 │ │ - 基于机器学习的预测 │ │ - 自动扩缩容 │ │ - 成本优化调度 │ │ │ │ 3. 事件驱动 │ │ - Kafka集成 │ │ - Webhook支持 │ │ - 事件溯源 │ │ │ │ 4. 跨数据中心 │ │ - 异地多活 │ │ - 数据同步 │ │ - 流量调度 │ └─────────────────────────────────────────────────────────────┘感谢你跟随这十讲的旅程从零开始我们构建了一个完整的分布式任务调度系统。希望这个项目能成为你学习和工作的有力工具。如果你有任何问题或建议欢迎在GitHub上提Issue或PR。祝编码愉快开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。 第10讲：部署与实战
经过九讲的开发我们拥有了一套完整的分布式任务调度系统。最后一讲我们将把它部署到生产环境并进行实战演练。从容器化到Kubernetes编排从压测到调优完成系统的最后一公里。一、部署架构总览1.1 生产部署拓扑┌─────────────────────────────────────────────────────────────┐ │ Kubernetes 集群 │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Ingress │ │ Ingress │ │ Ingress │ │ │ │ Controller │ │ Controller │ │ Controller │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ │ │ Scheduler │ │ Scheduler │ │ Scheduler │ │ │ │ (Leader) │ │ (Follower) │ │ (Follower) │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ │ │ Worker │ │ Worker │ │ Worker │ │ │ │ Pool │ │ Pool │ │ Pool │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ PostgreSQL │ │ etcd │ │ Prometheus │ │ │ │ (主从) │ │ (集群) │ │ Grafana │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ └─────────────────────────────────────────────────────────────┘1.2 服务组件清单┌─────────────────────────────────────────────────────────────┐ │ 组件 镜像 端口 副本数 │ ├─────────────────────────────────────────────────────────────┤ │ scheduler-api scheduler:latest 8080 3 │ │ scheduler-worker worker:latest 9001 5 │ │ postgres postgres:15 5432 2 (主从) │ │ etcd etcd:3.5 2379 3 │ │ prometheus prometheus:latest 9090 1 │ │ grafana grafana:latest 3000 1 │ └─────────────────────────────────────────────────────────────┘二、Docker容器化2.1 Dockerfile# Dockerfile FROM python:3.11-slim AS builder WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ libpq-dev \ rm -rf /var/lib/apt/lists/* # 安装Python依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 第二阶段运行镜像 FROM python:3.11-slim WORKDIR /app # 复制依赖 COPY --frombuilder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages COPY --frombuilder /usr/local/bin /usr/local/bin # 复制应用代码 COPY . . # 健康检查 HEALTHCHECK --interval30s --timeout3s --start-period5s --retries3 \ CMD python -c import urllib.request; urllib.request.urlopen(http://localhost:8080/health) EXPOSE 8080 CMD [python, main.py]2.2 docker-compose.yml# docker-compose.yml version: 3.8 services: # etcd集群 etcd: image: bitnami/etcd:3.5 environment: - ALLOW_NONE_AUTHENTICATIONyes - ETCD_NAMEetcd0 - ETCD_INITIAL_CLUSTERetcd0http://etcd:2380 - ETCD_INITIAL_ADVERTISE_PEER_URLShttp://etcd:2380 - ETCD_ADVERTISE_CLIENT_URLShttp://etcd:2379 - ETCD_LISTEN_CLIENT_URLShttp://0.0.0.0:2379 - ETCD_LISTEN_PEER_URLShttp://0.0.0.0:2380 ports: - 2379:2379 volumes: - etcd_data:/bitnami/etcd/data networks: - scheduler-net # PostgreSQL postgres: image: postgres:15 environment: POSTGRES_DB: scheduler POSTGRES_USER: scheduler POSTGRES_PASSWORD: scheduler_pass ports: - 5432:5432 volumes: - postgres_data:/var/lib/postgresql/data - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql networks: - scheduler-net healthcheck: test: [CMD-SHELL, pg_isready -U scheduler] interval: 10s timeout: 5s retries: 5 # 调度器3副本 scheduler: build: . image: scheduler:latest command: python -m scheduler.server --port8080 environment: - NODE_TYPEscheduler - ETCD_HOSTetcd - ETCD_PORT2379 - DB_DSNpostgresql://scheduler:scheduler_passpostgres:5432/scheduler - REDIS_URLredis://redis:6379/0 - LOG_LEVELINFO ports: - 8080 depends_on: postgres: condition: service_healthy etcd: condition: service_started deploy: replicas: 3 resources: limits: cpus: 1 memory: 512M networks: - scheduler-net # Worker节点5副本 worker: build: . image: scheduler:latest command: python -m scheduler.worker --port9001 environment: - NODE_TYPEworker - WORKER_IDworker-{{.Task.Slot}} - ETCD_HOSTetcd - ETCD_PORT2379 - MAX_CONCURRENCY10 - LOG_LEVELINFO ports: - 9001 depends_on: - scheduler deploy: replicas: 5 resources: limits: cpus: 2 memory: 1G networks: - scheduler-net # Prometheus监控 prometheus: image: prom/prometheus:latest volumes: - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml - prometheus_data:/prometheus ports: - 9090:9090 networks: - scheduler-net # Grafana可视化 grafana: image: grafana/grafana:latest environment: - GF_SECURITY_ADMIN_PASSWORDadmin - GF_INSTALL_PLUGINSgrafana-piechart-panel volumes: - ./grafana/dashboards:/etc/grafana/provisioning/dashboards - ./grafana/datasources:/etc/grafana/provisioning/datasources - grafana_data:/var/lib/grafana ports: - 3000:3000 depends_on: - prometheus networks: - scheduler-net # Redis缓存 redis: image: redis:7-alpine ports: - 6379:6379 volumes: - redis_data:/data networks: - scheduler-net volumes: etcd_data: postgres_data: prometheus_data: grafana_data: redis_data: networks: scheduler-net: driver: bridge2.3 配置文件# config/production.yaml server: host: 0.0.0.0 port: 8080 workers: 10 database: dsn: ${DB_DSN} pool_min: 5 pool_max: 20 pool_timeout: 30 etcd: hosts: [etcd:2379] timeout: 5 ttl: 10 redis: url: ${REDIS_URL} pool_size: 10 scheduler: tick_ms: 100 max_queue_size: 10000 dispatch_strategy: least_loaded worker: max_concurrency: 10 heartbeat_interval: 5 task_timeout_ms: 30000 monitoring: metrics_port: 9100 tracing_enabled: true tracing_endpoint: http://jaeger:14268/api/traces logging: level: ${LOG_LEVEL:-INFO} format: json output: stdout三、Kubernetes部署3.1 Helm Chart# helm/scheduler/Chart.yaml apiVersion: v2 name: distributed-scheduler description: A distributed task scheduler version: 1.0.0 appVersion: 1.0.03.2 Values配置# helm/scheduler/values.yaml replicaCount: 3 image: repository: scheduler tag: latest pullPolicy: Always config: logLevel: INFO maxConcurrency: 10 taskTimeoutMs: 30000 resources: scheduler: limits: cpu: 1 memory: 512Mi requests: cpu: 500m memory: 256Mi worker: limits: cpu: 2 memory: 1Gi requests: cpu: 1 memory: 512Mi persistence: enabled: true storageClass: standard size: 10Gi ingress: enabled: true host: scheduler.example.com tls: true monitoring: prometheus: enabled: true scrapeInterval: 15s grafana: enabled: true adminPassword: admin3.3 Deployment模板# helm/scheduler/templates/scheduler-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: {{ .Values.name }}-scheduler labels: app: scheduler spec: replicas: {{ .Values.replicaCount }} selector: matchLabels: app: scheduler template: metadata: labels: app: scheduler annotations: prometheus.io/scrape: true prometheus.io/port: 9100 spec: containers: - name: scheduler image: {{ .Values.image.repository }}:{{ .Values.image.tag }} command: [python, -m, scheduler.server] args: [--port8080] ports: - containerPort: 8080 name: http - containerPort: 9100 name: metrics env: - name: NODE_TYPE value: scheduler - name: ETCD_HOST value: etcd-cluster - name: DB_DSN valueFrom: secretKeyRef: name: db-secret key: dsn - name: LOG_LEVEL value: {{ .Values.config.logLevel }} resources: {{- toYaml .Values.resources.scheduler | nindent 10 }} livenessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: 8080 initialDelaySeconds: 5 periodSeconds: 5 --- apiVersion: v1 kind: Service metadata: name: scheduler-service spec: selector: app: scheduler ports: - port: 8080 targetPort: 8080 name: http - port: 9100 targetPort: 9100 name: metrics type: ClusterIP3.4 Worker StatefulSet# helm/scheduler/templates/worker-statefulset.yaml apiVersion: apps/v1 kind: StatefulSet metadata: name: {{ .Values.name }}-worker spec: serviceName: worker-headless replicas: 5 selector: matchLabels: app: worker template: metadata: labels: app: worker spec: containers: - name: worker image: {{ .Values.image.repository }}:{{ .Values.image.tag }} command: [python, -m, scheduler.worker] args: [--port9001] ports: - containerPort: 9001 name: grpc env: - name: NODE_TYPE value: worker - name: WORKER_ID valueFrom: fieldRef: fieldPath: metadata.name - name: ETCD_HOST value: etcd-cluster - name: MAX_CONCURRENCY value: {{ .Values.config.maxConcurrency }} resources: {{- toYaml .Values.resources.worker | nindent 12 }} --- apiVersion: v1 kind: Service metadata: name: worker-headless spec: clusterIP: None selector: app: worker ports: - port: 9001 targetPort: 90013.5 ConfigMap与Secret# helm/scheduler/templates/configmap.yaml apiVersion: v1 kind: ConfigMap metadata: name: scheduler-config data: config.yaml: | server: host: 0.0.0.0 port: 8080 scheduler: tick_ms: 100 max_queue_size: 10000 monitoring: metrics_port: 9100 --- apiVersion: v1 kind: Secret metadata: name: db-secret type: Opaque stringData: dsn: postgresql://scheduler:scheduler_passpostgres:5432/scheduler四、CI/CD流水线4.1 GitHub Actions# .github/workflows/deploy.yml name: Deploy Scheduler on: push: branches: [main] pull_request: branches: [main] env: REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkoutv3 - name: Set up Python uses: actions/setup-pythonv4 with: python-version: 3.11 - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest pytest-asyncio - name: Run tests run: | pytest tests/ -v --cov./ --cov-reportxml - name: Upload coverage uses: codecov/codecov-actionv3 build-and-push: needs: test runs-on: ubuntu-latest if: github.event_name push steps: - uses: actions/checkoutv3 - name: Set up Docker Buildx uses: docker/setup-buildx-actionv2 - name: Login to Container Registry uses: docker/login-actionv2 with: registry: ${{ env.REGISTRY }} username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - name: Extract metadata id: meta uses: docker/metadata-actionv4 with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} tags: | typesha,formatlong typeref,eventbranch typesemver,pattern{{version}} - name: Build and push uses: docker/build-push-actionv4 with: context: . push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} cache-from: typegha cache-to: typegha,modemax deploy: needs: build-and-push runs-on: ubuntu-latest if: github.ref refs/heads/main steps: - name: Configure kubectl uses: azure/setup-kubectlv3 with: version: latest - name: Set up Kubeconfig run: | mkdir -p $HOME/.kube echo ${{ secrets.KUBECONFIG }} $HOME/.kube/config - name: Deploy to Kubernetes run: | helm upgrade --install scheduler ./helm/scheduler \ --set image.tag${{ github.sha }} \ --namespace scheduler-system \ --create-namespace \ --wait \ --timeout 10m - name: Verify deployment run: | kubectl rollout status deployment/scheduler-scheduler \ -n scheduler-system --timeout5m kubectl rollout status statefulset/scheduler-worker \ -n scheduler-system --timeout5m五、压测与调优5.1 压测脚本# scripts/load_test.py 负载测试脚本 模拟大量任务提交和执行测试系统性能。 import asyncio import aiohttp import time import random import statistics from dataclasses import dataclass from typing import List dataclass class LoadTestConfig: 压测配置 scheduler_url: str http://localhost:8080 total_tasks: int 10000 concurrency: int 50 task_types: List[str] None def __post_init__(self): if self.task_types is None: self.task_types [sleep, echo, compute] class LoadTester: 负载测试器 def __init__(self, config: LoadTestConfig): self.config config self.results [] self.errors 0 async def submit_task(self, session: aiohttp.ClientSession, task_id: int) - float: 提交单个任务 task_type random.choice(self.config.task_types) payload { name: fload-test-{task_id}, task_type: task_type, handler: fhandlers.{task_type}, params: { duration: random.uniform(0.1, 0.5), task_id: task_id }, priority: random.choice([LOW, MEDIUM, HIGH, CRITICAL]), schedule_time: time.time() random.uniform(0, 0.1) } start time.perf_counter() try: async with session.post( f{self.config.scheduler_url}/api/tasks, jsonpayload, timeoutaiohttp.ClientTimeout(total5) ) as resp: if resp.status 200: latency time.perf_counter() - start return latency else: self.errors 1 return None except Exception as e: self.errors 1 return None async def run(self): 运行压测 print(fStarting load test: {self.config.total_tasks} tasks, f{self.config.concurrency} concurrency) connector aiohttp.TCPConnector(limitself.config.concurrency) async with aiohttp.ClientSession(connectorconnector) as session: tasks [] for i in range(self.config.total_tasks): tasks.append(self.submit_task(session, i)) # 分批执行 batch_size self.config.concurrency * 10 for i in range(0, len(tasks), batch_size): batch tasks[i:i batch_size] results await asyncio.gather(*batch) self.results.extend([r for r in results if r is not None]) progress (i len(batch)) / len(tasks) * 100 print(f\rProgress: {progress:.1f}% f({len(self.results)} succeeded, f{self.errors} errors), end) print() self.print_report() def print_report(self): 打印报告 if not self.results: print(No successful requests) return latencies sorted(self.results) n len(latencies) print(\n * 70) print( Load Test Report) print( * 70) print(fTotal tasks: {self.config.total_tasks}) print(fSuccessful: {len(self.results)}) print(fErrors: {self.errors}) print(fSuccess rate: {len(self.results)/self.config.total_tasks*100:.1f}%) print(f\nLatency (seconds):) print(f Average: {statistics.mean(latencies)*1000:.2f} ms) print(f P50: {latencies[n//2]*1000:.2f} ms) print(f P90: {latencies[int(n*0.9)]*1000:.2f} ms) print(f P99: {latencies[int(n*0.99)]*1000:.2f} ms) print(f Min: {latencies[0]*1000:.2f} ms) print(f Max: {latencies[-1]*1000:.2f} ms) print(f\nThroughput: {len(self.results)/(latencies[-1]):.0f} req/s) async def main(): config LoadTestConfig( scheduler_urlhttp://localhost:8080, total_tasks5000, concurrency50 ) tester LoadTester(config) await tester.run() if __name__ __main__: asyncio.run(main())5.2 调优指南# 生产调优指南 ## 1. 操作系统调优 ### 1.1 内核参数bash/etc/sysctl.confnet.core.somaxconn 65535net.ipv4.tcp_max_syn_backlog 65535net.ipv4.ip_local_port_range 1024 65000fs.file-max 1000000应用sysctl -p### 1.2 ulimit设置bash/etc/security/limits.confsoft nofile 1000000hard nofile 1000000soft nproc 100000hard nproc 100000## 2. Python调优 ### 2.1 GIL优化python对于CPU密集型任务使用多进程from multiprocessing import Pool或者使用异步线程池import asynciofrom concurrent.futures import ThreadPoolExecutorexecutor ThreadPoolExecutor(max_workers10)loop asyncio.get_event_loop()result await loop.run_in_executor(executor, cpu_intensive_func)### 2.2 内存优化python使用slots减少内存占用class Task:slots [task_id, name, status]使用array代替list存储大量数字from array import arraynumbers array(i, [1, 2, 3, 4, 5])## 3. 数据库调优 ### 3.1 PostgreSQL配置inipostgresql.confmax_connections 200shared_buffers 1GBeffective_cache_size 3GBwork_mem 32MBmaintenance_work_mem 128MBrandom_page_cost 1.1effective_io_concurrency 200wal_buffers 16MB### 3.2 查询优化sql-- 创建复合索引CREATE INDEX idx_tasks_status_schedule ON tasks(status, schedule_time);CREATE INDEX idx_tasks_priority_created ON tasks(priority, created_at DESC);-- 使用覆盖索引CREATE INDEX idx_tasks_covering ON tasks(status, schedule_time) INCLUDE (name, task_type);## 4. 应用调优 ### 4.1 连接池配置pythonpool_config PoolConfig(min_size10, # 最小连接数max_size50, # 最大连接数max_idle_time300, # 最大空闲时间acquire_timeout5 # 获取超时)### 4.2 缓存策略python热点数据缓存cache TwoLevelCache(local_size10000, # 本地缓存10K条local_ttl60, # 本地缓存1分钟remote_ttl600 # Redis缓存10分钟)### 4.3 批量处理python批量写入数据库batch_size 100 # 每批100条flush_interval 0.1 # 100ms刷新一次## 5. Kubernetes调优 ### 5.1 HPA配置yamlapiVersion: autoscaling/v2kind: HorizontalPodAutoscalermetadata:name: scheduler-hpaspec:scaleTargetRef:apiVersion: apps/v1kind: Deploymentname: schedulerminReplicas: 3maxReplicas: 10metrics:type: Resourceresource:name: cputarget:type: UtilizationaverageUtilization: 70type: Podspods:metric:name: scheduler_queue_depthtarget:type: AverageValueaverageValue: 1000### 5.2 PDB配置yamlapiVersion: policy/v1kind: PodDisruptionBudgetmetadata:name: scheduler-pdbspec:minAvailable: 2selector:matchLabels:app: scheduler六、运维手册6.1 日常运维命令#!/bin/bash # scripts/ops.sh # 调度系统运维脚本 # 查看集群状态 alias sched-statuskubectl get all -n scheduler-system alias sched-logskubectl logs -n scheduler-system -l appscheduler --tail100 alias sched-workerskubectl get pods -n scheduler-system -l appworker # 查看Leader function sched-leader() { kubectl exec -n scheduler-system deploy/scheduler-scheduler-0 -- \ python -c from coordination.election import LeaderElection; print(Leader check) } # 滚动更新 function sched-update() { kubectl set image deployment/scheduler-scheduler \ scheduler$1 -n scheduler-system --record kubectl rollout status deployment/scheduler-scheduler -n scheduler-system } # 扩缩容 function sched-scale() { kubectl scale deployment/scheduler-scheduler --replicas$1 -n scheduler-system kubectl scale statefulset/scheduler-worker --replicas$2 -n scheduler-system } # 查看监控 function sched-metrics() { kubectl port-forward -n scheduler-system svc/prometheus 9090:9090 kubectl port-forward -n scheduler-system svc/grafana 3000:3000 echo Prometheus: http://localhost:9090 echo Grafana: http://localhost:3000 (admin/admin) } # 备份数据库 function sched-backup() { kubectl exec -n scheduler-system deploy/postgres -- \ pg_dump -U scheduler scheduler backup_$(date %Y%m%d_%H%M%S).sql } # 查看任务队列 function sched-queue() { kubectl exec -n scheduler-system deploy/scheduler-scheduler-0 -- \ curl -s localhost:8080/api/stats | python -m json.tool }6.2 告警规则# prometheus/alerts.yml groups: - name: scheduler-alerts rules: - alert: SchedulerDown expr: up{jobscheduler} 1 for: 1m labels: severity: critical annotations: summary: Scheduler instance down - alert: HighTaskFailureRate expr: rate(scheduler_tasks_failed_total[5m]) / rate(scheduler_tasks_total[5m]) 0.1 for: 5m labels: severity: warning annotations: summary: High task failure rate - alert: QueueDepthHigh expr: scheduler_queue_depth 10000 for: 2m labels: severity: warning annotations: summary: Task queue depth exceeds 10000 - alert: WorkerOffline expr: scheduler_workers_active 3 for: 1m labels: severity: critical annotations: summary: Less than 3 workers active - alert: HighLatency expr: histogram_quantile(0.99, rate(scheduler_task_duration_seconds_bucket[5m])) 10 for: 5m labels: severity: warning annotations: summary: P99 task latency exceeds 10s七、实战演练7.1 端到端测试# scripts/e2e_test.py 端到端测试 验证整个系统的功能完整性。 import asyncio import aiohttp import time import json class E2ETest: 端到端测试 def __init__(self, base_url: str http://localhost:8080): self.base_url base_url self.session None async def __aenter__(self): self.session aiohttp.ClientSession() return self async def __aexit__(self, *args): await self.session.close() async def test_health(self): 测试健康检查 async with self.session.get(f{self.base_url}/health) as resp: assert resp.status 200 data await resp.json() assert data[status] healthy print(✅ Health check passed) async def test_submit_task(self): 测试提交任务 payload { name: e2e-test-task, task_type: sleep, handler: handlers.sleep, params: {duration: 0.5}, priority: HIGH, schedule_time: time.time() } async with self.session.post( f{self.base_url}/api/tasks, jsonpayload ) as resp: assert resp.status 200 data await resp.json() assert task_id in data print(f✅ Task submitted: {data[task_id]}) return data[task_id] async def test_get_task(self, task_id: str): 测试获取任务 async with self.session.get( f{self.base_url}/api/tasks/{task_id} ) as resp: assert resp.status 200 data await resp.json() assert data[task_id] task_id print(f✅ Task retrieved: {data[name]} ({data[status]})) async def test_list_tasks(self): 测试任务列表 async with self.session.get( f{self.base_url}/api/tasks?limit10 ) as resp: assert resp.status 200 data await resp.json() assert tasks in data print(f✅ Tasks listed: {len(data[tasks])} tasks) async def test_workflow(self): 测试工作流 workflow { name: e2e-workflow, nodes: [ {node_id: extract, name: Extract, type: task}, {node_id: transform, name: Transform, type: task}, {node_id: load, name: Load, type: task} ], edges: [ {source: extract, target: transform}, {source: transform, target: load} ] } async with self.session.post( f{self.base_url}/api/workflows, jsonworkflow ) as resp: assert resp.status 200 data await resp.json() print(f✅ Workflow created: {data[workflow_id]}) async def test_stats(self): 测试统计 async with self.session.get(f{self.base_url}/api/stats) as resp: assert resp.status 200 data await resp.json() print(f✅ Stats retrieved: {json.dumps(data, indent2)}) async def run_all(self): 运行所有测试 print( * 65) print( E2E Test Suite) print( * 65) await self.test_health() task_id await self.test_submit_task() await asyncio.sleep(1) await self.test_get_task(task_id) await self.test_list_tasks() await self.test_workflow() await self.test_stats() print(\n * 65) print(✅ All E2E tests passed!) print( * 65) async def main(): async with E2ETest() as tester: await tester.run_all() if __name__ __main__: asyncio.run(main())八、总结8.1 十讲回顾┌─────────────────────────────────────────────────────────────┐ │ 分布式任务调度系统 · 十讲回顾 │ ├──────┬──────────────────────────────┬───────────────────────┤ │ 讲次 │ 主题 │ 核心产出 │ ├──────┼──────────────────────────────┼───────────────────────┤ │ 01 │ 系统设计与任务模型 │ Task/Job模型 │ │ 02 │ 调度引擎 │ 时间轮优先级队列 │ │ 03 │ 分布式协调 │ 选举锁服务发现 │ │ 04 │ 任务分发与执行 │ Worker分发器 │ │ 05 │ 故障转移与高可用 │ 心跳重试熔断 │ │ 06 │ 持久化与历史 │ 数据库归档 │ │ 07 │ 监控与告警 │ 指标追踪告警 │ │ 08 │ 高级特性 │ CronDAG分片 │ │ 09 │ 性能优化 │ 连接池缓存批处理 │ │ 10 │ 部署与实战 │ DockerK8sCI/CD │ └──────┴──────────────────────────────┴───────────────────────┘8.2 项目结构distributed-scheduler/ ├── scheduler/ # 调度器核心 │ ├── engine.py # 调度引擎 │ ├── dispatcher.py # 任务分发 │ └── orchestrator.py # 编排器 ├── worker/ # 执行器 │ ├── executor.py # 任务执行 │ └── handlers.py # 处理器 ├── coordination/ # 分布式协调 │ ├── election.py # Leader选举 │ ├── lock.py # 分布式锁 │ ├── registry.py # 服务注册 │ └── hash_ring.py # 一致性哈希 ├── storage/ # 持久化 │ ├── models.py # ORM模型 │ ├── repository.py # 数据仓库 │ └── archiver.py # 归档管理 ├── monitor/ # 监控 │ ├── metrics.py # 指标采集 │ ├── tracing.py # 链路追踪 │ ├── alerting.py # 告警引擎 │ └── rules.py # 告警规则 ├── fault_tolerance/ # 高可用 │ ├── heartbeat.py # 心跳检测 │ ├── retry.py # 重试管理 │ ├── failover.py # 故障转移 │ └── circuit_breaker.py # 断路器 ├── advanced/ # 高级特性 │ ├── cron.py # Cron调度 │ ├── dag_workflow.py # DAG工作流 │ └── sharding.py # 动态分片 ├── optimization/ # 性能优化 │ ├── connection_pool.py # 连接池 │ ├── batch_processor.py # 批处理 │ ├── cache.py # 缓存 │ └── object_pool.py # 对象池 ├── config/ # 配置 ├── scripts/ # 脚本 ├── helm/ # K8s部署 ├── tests/ # 测试 └── docs/ # 文档8.3 下一步方向未来演进方向 ┌─────────────────────────────────────────────────────────────┐ │ 1. 多租户支持 │ │ - 租户隔离 │ │ - 配额管理 │ │ - 计费系统 │ │ │ │ 2. 智能调度 │ │ - 基于机器学习的预测 │ │ - 自动扩缩容 │ │ - 成本优化调度 │ │ │ │ 3. 事件驱动 │ │ - Kafka集成 │ │ - Webhook支持 │ │ - 事件溯源 │ │ │ │ 4. 跨数据中心 │ │ - 异地多活 │ │ - 数据同步 │ │ - 流量调度 │ └─────────────────────────────────────────────────────────────┘感谢你跟随这十讲的旅程从零开始我们构建了一个完整的分布式任务调度系统。希望这个项目能成为你学习和工作的有力工具。如果你有任何问题或建议欢迎在GitHub上提Issue或PR。祝编码愉快开发之余的小工具推荐处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。