From 554d224be23fc89c8a756d700b0477413818d018 Mon Sep 17 00:00:00 2001 From: eezd Date: Tue, 13 Oct 2026 13:51:00 +0800 Subject: [PATCH] Harden background loops and API edge cases from code review - crawler manager: retry transient errors inside each loop, never let log writes break supervision, run tasks in a TaskGroup so any escaped error restarts the container instead of leaving a half-dead process - market worker: tolerate symbols deleted mid-backfill, retry task selection and symbol queries, run loops in a TaskGroup - market WS: validate subscription messages and close the connection when either the receive or forward side ends - news pipeline: make first insert concurrency-safe with ON CONFLICT - market API: only proxy ticker/1s klines for added symbols, map Binance rate limits to 503 and upstream errors to 502 without blocking - news API: escape LIKE wildcards in search, return 422 for bad cursors - market page: cancel pending reconnect timer on unmount --- backend/src/alphapulse/api/market.py | 92 ++++++++--- backend/src/alphapulse/api/news.py | 16 +- backend/src/alphapulse/crawler/manager.py | 163 +++++++++++++------- backend/src/alphapulse/crawler/pipelines.py | 26 +++- backend/src/alphapulse/market/binance.py | 12 +- backend/src/alphapulse/market/worker.py | 57 ++++--- web/src/pages/market/useMarketSocket.ts | 7 +- 7 files changed, 259 insertions(+), 114 deletions(-) diff --git a/backend/src/alphapulse/api/market.py b/backend/src/alphapulse/api/market.py index 8e03583..bdea270 100644 --- a/backend/src/alphapulse/api/market.py +++ b/backend/src/alphapulse/api/market.py @@ -1,11 +1,15 @@ import asyncio import json +import logging import time +from collections.abc import Awaitable from datetime import datetime from typing import Annotated, Literal +import httpx from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect, status -from pydantic import BaseModel +from pydantic import BaseModel, Field, ValidationError, field_validator +from starlette.websockets import WebSocketState from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -13,7 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from alphapulse.core.auth import require_admin, websocket_admin from alphapulse.core.db import get_session from alphapulse.core.redis import KEY_PREFIX, redis -from alphapulse.market.binance import BinanceClient +from alphapulse.market.binance import BinanceClient, RateLimited from alphapulse.market.channels import kline_channel, ticker_channel from alphapulse.models import Kline, KlineSyncState, MarketSymbol from alphapulse.models.market import INTERVALS @@ -26,7 +30,9 @@ Market = Literal["spot", "um"] SYMBOLS_CACHE_SECONDS = 3600 MAX_KLINES = 1500 -client = BinanceClient() +logger = logging.getLogger("alphapulse.api.market") +# API 请求不等待限流，直接报 503，避免 HTTP 请求被挂起 +client = BinanceClient(wait=False) class IntervalProgress(BaseModel): @@ -62,11 +68,43 @@ class TradableSymbol(BaseModel): volume_precision: int +class Subscription(BaseModel): + market: Market + symbol: str = Field(pattern=r"^[A-Z0-9]{2,30}$") + interval: str + + @field_validator("interval") + @classmethod + def check_interval(cls, value: str) -> str: + if value not in (*INTERVALS, "1s"): + raise ValueError(f"不支持的周期：{value}") + return value + + +async def call_binance[T](request: Awaitable[T]) -> T: + """把币安调用的失败转换成明确的 HTTP 错误，而不是 500""" + try: + return await request + except RateLimited as e: + raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, str(e)) from None + except httpx.HTTPStatusError as e: + raise HTTPException(status.HTTP_502_BAD_GATEWAY, f"币安返回 {e.response.status_code}：{e.response.text[:200]}") from None + except httpx.TransportError as e: + raise HTTPException(status.HTTP_502_BAD_GATEWAY, f"无法连接币安：{e!r}") from None + + +async def ensure_symbol(session: AsyncSession, market: str, symbol: str) -> None: + """只允许查询已添加的交易对，不把任意 symbol 转发给币安""" + exists = await session.scalar(select(MarketSymbol.id).where(MarketSymbol.market == market, MarketSymbol.symbol == symbol)) + if exists is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"未添加交易对 {market} {symbol}") + + async def tradable_symbols(market: str) -> list[dict]: key = f"{KEY_PREFIX}market:symbols:{market}" if cached := await redis.get(key): return json.loads(cached) - symbols = await client.tradable_symbols(market) + symbols = await call_binance(client.tradable_symbols(market)) await redis.set(key, json.dumps(symbols), ex=SYMBOLS_CACHE_SECONDS) return symbols @@ -136,8 +174,9 @@ async def klines( # 1s 不落库，直接取币安最近数据 if market != "spot": raise HTTPException(status.HTTP_400_BAD_REQUEST, "合约没有 1s 周期") + await ensure_symbol(session, market, symbol) params = {"symbol": symbol, "interval": "1s", "limit": limit, **({"endTime": end - 1} if end else {})} - rows = await client.get("spot", "klines", params) + rows = await call_binance(client.get("spot", "klines", params)) return [[r[0], float(r[1]), float(r[2]), float(r[3]), float(r[4]), float(r[5]), float(r[7])] for r in rows] if interval not in INTERVALS: raise HTTPException(status.HTTP_400_BAD_REQUEST, f"不支持的周期：{interval}") @@ -154,8 +193,9 @@ async def klines( @router.get("/ticker") -async def ticker(market: Market, symbol: str) -> dict: - return await client.ticker(market, symbol) +async def ticker(session: Annotated[AsyncSession, Depends(get_session)], market: Market, symbol: str) -> dict: + await ensure_symbol(session, market, symbol) + return await call_binance(client.ticker(market, symbol)) @ws_router.websocket("/ws") @@ -168,6 +208,19 @@ async def market_ws(websocket: WebSocket) -> None: pubsub = redis.pubsub() subscribed = asyncio.Event() + async def receive() -> None: + while True: + raw = await websocket.receive_text() + try: + sub = Subscription.model_validate_json(raw) + except ValidationError as e: + await websocket.send_text(json.dumps({"type": "error", "message": f"订阅参数无效：{e.errors()[0]['msg']}"}, ensure_ascii=False)) + continue + if subscribed.is_set(): + await pubsub.unsubscribe() + await pubsub.subscribe(kline_channel(sub.market, sub.symbol, sub.interval), ticker_channel(sub.market, sub.symbol)) + subscribed.set() + async def forward() -> None: await subscribed.wait() while True: @@ -184,19 +237,18 @@ async def market_ws(websocket: WebSocket) -> None: "data": json.loads(message["data"]), })) - task = asyncio.create_task(forward()) + # 收、发任一方向结束（客户端断开或出错）就整体收尾；非正常断开时主动关闭，让前端重连， + # 避免转发任务已死而连接看起来仍正常 + tasks = [asyncio.create_task(receive()), asyncio.create_task(forward())] try: - while True: - sub = await websocket.receive_json() - if subscribed.is_set(): - await pubsub.unsubscribe() - await pubsub.subscribe( - kline_channel(sub["market"], sub["symbol"], sub["interval"]), - ticker_channel(sub["market"], sub["symbol"]), - ) - subscribed.set() - except WebSocketDisconnect: - pass + done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for task in done: + error = task.exception() + if error is not None and not isinstance(error, WebSocketDisconnect): + logger.warning("行情 WS 连接异常结束：%r", error) finally: - task.cancel() + for task in tasks: + task.cancel() await pubsub.aclose() + if websocket.application_state == WebSocketState.CONNECTED and websocket.client_state == WebSocketState.CONNECTED: + await websocket.close(code=1011) diff --git a/backend/src/alphapulse/api/news.py b/backend/src/alphapulse/api/news.py index b07e068..a083c12 100644 --- a/backend/src/alphapulse/api/news.py +++ b/backend/src/alphapulse/api/news.py @@ -69,15 +69,23 @@ async def list_news( if source: stmt = stmt.where(NewsItem.source_code.in_(source)) if q: - pattern = f"%{q}%" - stmt = stmt.where(or_(NewsItem.title.ilike(pattern), NewsItem.content_text.ilike(pattern))) + # 转义 LIKE 通配符，按字面搜索 + escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + pattern = f"%{escaped}%" + stmt = stmt.where(or_(NewsItem.title.ilike(pattern, escape="\\"), NewsItem.content_text.ilike(pattern, escape="\\"))) if start: stmt = stmt.where(NewsItem.published_at >= start) if end: stmt = stmt.where(NewsItem.published_at < end) if cursor: - published_at, news_id = cursor.rsplit("|", 1) - stmt = stmt.where(tuple_(NewsItem.published_at, NewsItem.id) < (datetime.fromisoformat(published_at), int(news_id))) + try: + published_at, news_id = cursor.rsplit("|", 1) + position = (datetime.fromisoformat(published_at), int(news_id)) + if position[0].tzinfo is None: + raise ValueError("cursor 时间缺少时区") + except ValueError: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, "cursor 格式无效") from None + stmt = stmt.where(tuple_(NewsItem.published_at, NewsItem.id) < position) rows = (await session.execute(stmt.order_by(NewsItem.published_at.desc(), NewsItem.id.desc()).limit(limit))).all() items = [NewsBrief.model_validate({**news.__dict__, "excerpt": excerpt}) for news, excerpt in rows] diff --git a/backend/src/alphapulse/crawler/manager.py b/backend/src/alphapulse/crawler/manager.py index ea92cd6..0f8c9f5 100644 --- a/backend/src/alphapulse/crawler/manager.py +++ b/backend/src/alphapulse/crawler/manager.py @@ -24,6 +24,7 @@ RESTART_DELAY_MAX = 300 # 子进程稳定运行超过该时长后，异常退出的重启延迟重新从最小值开始 STABLE_SECONDS = 60 STOP_TIMEOUT = 15 +CONTROL_RECONNECT_SECONDS = 5 @dataclass @@ -40,8 +41,12 @@ class Worker: async def log(code: str, level: str, message: str) -> None: + """写 stdout 和 crawl_logs；写库失败只记 stdout，不能让守护任务因为日志而中断""" logger.log(logging.getLevelName(level), "[%s] %s", code, message) - await write_logs([(code, None, level, message)]) + try: + await write_logs([(code, None, level, message)]) + except Exception: + logger.exception("[%s] 日志落库失败", code) class Manager: @@ -58,26 +63,22 @@ class Manager: for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, self.stopping.set) - tasks = [asyncio.create_task(self.supervise(w)) for w in self.workers.values()] - tasks += [asyncio.create_task(c) for c in (self.schedule_loop(), self.control_loop(), self.status_loop())] - await self.stopping.wait() - logger.info("收到退出信号，停止全部子进程") - await asyncio.gather(*(self.stop_process(w) for w in self.workers.values())) - for task in tasks: - task.cancel() + # 各循环自行重试可恢复的错误；仍有异常逃出时 TaskGroup 让整个进程退出，由 docker 重启， + # 避免出现"进程活着但调度或守护任务已经死掉"的状态 + async with asyncio.TaskGroup() as tg: + for w in self.workers.values(): + tg.create_task(self.supervise(w)) + loops = [tg.create_task(c) for c in (self.schedule_loop(), self.control_loop(), self.status_loop())] + await self.stopping.wait() + logger.info("收到退出信号，停止全部子进程") + await asyncio.gather(*(self.stop_process(w) for w in self.workers.values())) + for task in loops: + task.cancel() await redis.delete(STATUS_KEY) async def supervise(self, w: Worker) -> None: while not self.stopping.is_set(): - # 上次进程遗留的请求和未收尾的轮次都已失效，清理后从干净状态开始 - await redis.delete(f"{KEY_PREFIX}{w.code}:requests", start_urls_key(w.code)) - async with SessionLocal() as session: - await session.execute( - update(CrawlRun) - .where(CrawlRun.source_code == w.code, CrawlRun.status == "running") - .values(status="interrupted", finished_at=datetime.now(UTC), error="子进程退出时本轮未完成") - ) - await session.commit() + await self.cleanup_before_spawn(w) w.proc = await asyncio.create_subprocess_exec("scrapy", "crawl", w.code) w.started_at = datetime.now(UTC) # 新进程的第一轮立即执行，并做全量回溯扫描 @@ -104,6 +105,20 @@ class Manager: pass w.restart_delay = min(w.restart_delay * 2, RESTART_DELAY_MAX) + async def cleanup_before_spawn(self, w: Worker) -> None: + """上次进程遗留的请求和未收尾的轮次都已失效；清理失败不影响拉起子进程""" + try: + await redis.delete(f"{KEY_PREFIX}{w.code}:requests", start_urls_key(w.code)) + async with SessionLocal() as session: + await session.execute( + update(CrawlRun) + .where(CrawlRun.source_code == w.code, CrawlRun.status == "running") + .values(status="interrupted", finished_at=datetime.now(UTC), error="子进程退出时本轮未完成") + ) + await session.commit() + except Exception: + logger.exception("[%s] 拉起前清理失败，继续启动子进程", w.code) + async def stop_process(self, w: Worker) -> None: if w.proc is None: return @@ -115,38 +130,66 @@ class Manager: w.proc.kill() async def schedule_loop(self) -> None: + unknown: set[str] = set() while True: - async with SessionLocal() as session: - sources = (await session.scalars(select(Source))).all() - now = datetime.now(UTC) - for s in sources: - w = self.workers[s.code] - if not s.enabled or w.proc is None or w.restarting: - continue - if s.paused_until and s.paused_until > now: - w.next_run_at = max(w.next_run_at, s.paused_until) - continue - if now < w.next_run_at: - continue - # 子进程还没取走上一条触发时不再堆积 - if await redis.llen(start_urls_key(s.code)) == 0: - await redis.lpush(start_urls_key(s.code), json.dumps({"trigger": w.trigger, "backfill_hours": s.backfill_hours})) - w.trigger = "schedule" - w.next_run_at = now + timedelta(seconds=s.interval_seconds) + try: + await self.schedule_once(unknown) + except Exception: + logger.exception("调度出错，1s 后重试") await asyncio.sleep(1) + async def schedule_once(self, unknown: set[str]) -> None: + async with SessionLocal() as session: + sources = (await session.scalars(select(Source))).all() + now = datetime.now(UTC) + for s in sources: + w = self.workers.get(s.code) + if w is None: + # 启动后才新增的来源，需要重启管理进程才会接管 + if s.code not in unknown: + unknown.add(s.code) + logger.warning("来源 %s 不在本进程的守护列表中，重启管理进程后生效", s.code) + continue + if not s.enabled or w.proc is None or w.restarting: + continue + if s.paused_until and s.paused_until > now: + w.next_run_at = max(w.next_run_at, s.paused_until) + continue + if now < w.next_run_at: + continue + # 子进程还没取走上一条触发时不再堆积 + if await redis.llen(start_urls_key(s.code)) == 0: + await redis.lpush(start_urls_key(s.code), json.dumps({"trigger": w.trigger, "backfill_hours": s.backfill_hours})) + w.trigger = "schedule" + w.next_run_at = now + timedelta(seconds=s.interval_seconds) + async def control_loop(self) -> None: - async with redis.pubsub() as pubsub: - await pubsub.subscribe(CONTROL_CHANNEL) - async for message in pubsub.listen(): - if message["type"] != "message": - continue - command = json.loads(message["data"]) - if command["action"] != "restart": - logger.warning("未知指令：%s", command) - continue - codes = list(self.workers) if command["source"] == "all" else [command["source"]] - await asyncio.gather(*(self.restart(self.workers[code]) for code in codes)) + while True: + try: + async with redis.pubsub() as pubsub: + await pubsub.subscribe(CONTROL_CHANNEL) + async for message in pubsub.listen(): + if message["type"] == "message": + await self.handle_command(message["data"]) + except Exception: + logger.exception("控制通道断开，%ds 后重新订阅", CONTROL_RECONNECT_SECONDS) + await asyncio.sleep(CONTROL_RECONNECT_SECONDS) + + async def handle_command(self, raw: str) -> None: + """单条指令出错只记日志，不影响后续指令""" + try: + command = json.loads(raw) + if command.get("action") != "restart": + logger.warning("未知指令：%s", raw) + return + source = command.get("source") + codes = list(self.workers) if source == "all" else [source] + if any(code not in self.workers for code in codes): + logger.warning("重启指令中的来源不存在：%s", raw) + return + await asyncio.gather(*(self.restart(self.workers[code]) for code in codes)) + except Exception: + logger.exception("处理指令失败：%s", raw) async def restart(self, w: Worker) -> None: await log(w.code, "INFO", "收到重启指令") @@ -166,20 +209,26 @@ class Manager: async def status_loop(self) -> None: while True: - mapping = { - w.code: json.dumps({ - "state": "restarting" if w.restarting else ("running" if w.proc else "stopped"), - "pid": w.proc.pid if w.proc else None, - "started_at": w.started_at.isoformat() if w.started_at else None, - "next_run_at": w.next_run_at.isoformat(), - }) - for w in self.workers.values() - } - async with redis.pipeline() as pipe: - # 管理进程挂掉后状态自动过期，前端据此显示"管理进程未运行" - await pipe.hset(STATUS_KEY, mapping=mapping).expire(STATUS_KEY, 10).execute() + try: + await self.publish_status() + except Exception: + logger.exception("发布状态失败，2s 后重试") await asyncio.sleep(2) + async def publish_status(self) -> None: + mapping = { + w.code: json.dumps({ + "state": "restarting" if w.restarting else ("running" if w.proc else "stopped"), + "pid": w.proc.pid if w.proc else None, + "started_at": w.started_at.isoformat() if w.started_at else None, + "next_run_at": w.next_run_at.isoformat(), + }) + for w in self.workers.values() + } + async with redis.pipeline() as pipe: + # 管理进程挂掉后状态自动过期，前端据此显示"管理进程未运行" + await pipe.hset(STATUS_KEY, mapping=mapping).expire(STATUS_KEY, 10).execute() + if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") diff --git a/backend/src/alphapulse/crawler/pipelines.py b/backend/src/alphapulse/crawler/pipelines.py index c806baa..20be1ef 100644 --- a/backend/src/alphapulse/crawler/pipelines.py +++ b/backend/src/alphapulse/crawler/pipelines.py @@ -4,6 +4,7 @@ from typing import Any from scrapy.crawler import Crawler from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert from alphapulse.core.db import SessionLocal from alphapulse.core.redis import redis @@ -32,17 +33,26 @@ class NewsPipeline: spider = self.crawler.spider digest = content_hash(item) raw_response_id = item.pop("raw_response_id") + locked = ( + select(NewsItem) + .where(NewsItem.source_code == item["source_code"], NewsItem.external_id == item["external_id"]) + .with_for_update() + ) async with SessionLocal() as session: - news = await session.scalar( - select(NewsItem) - .where(NewsItem.source_code == item["source_code"], NewsItem.external_id == item["external_id"]) - .with_for_update() - ) + news = await session.scalar(locked) + inserted_id = None if news is None: + # 行不存在时 FOR UPDATE 锁不住任何东西；用 ON CONFLICT 处理同一条目被并发首次写入 + inserted_id = await session.scalar( + pg_insert(NewsItem) + .values(**item, content_hash=digest, revision_no=1) + .on_conflict_do_nothing(index_elements=[NewsItem.source_code, NewsItem.external_id]) + .returning(NewsItem.id) + ) + # 插入成功取新行；冲突说明另一处刚写入，按已存在条目走比较/修订流程 + news = await session.scalar(locked) + if inserted_id is not None: kind = "new" - news = NewsItem(**item, content_hash=digest, revision_no=1) - session.add(news) - await session.flush() elif news.content_hash != digest: kind = "revised" for field in UPDATABLE_FIELDS: diff --git a/backend/src/alphapulse/market/binance.py b/backend/src/alphapulse/market/binance.py index c82ea6b..f4751f5 100644 --- a/backend/src/alphapulse/market/binance.py +++ b/backend/src/alphapulse/market/binance.py @@ -27,8 +27,14 @@ def decimals(step: str) -> int: return len(step.rstrip("0").split(".")[1]) if "." in step.rstrip("0") else 0 +class RateLimited(Exception): + """非等待模式下被币安限流""" + + class BinanceClient: - def __init__(self) -> None: + def __init__(self, wait: bool = True) -> None: + # wait=False 用于 API 请求：被限流时立即抛 RateLimited，不让 HTTP 请求挂起等待 + self.wait = wait self.http = httpx.AsyncClient(timeout=20) async def get(self, market: str, path_key: str, params: dict[str, Any] | None = None) -> Any: @@ -45,6 +51,8 @@ class BinanceClient: if resp.status_code in (418, 429): # 418 表示 IP 已被临时封禁，必须按 Retry-After 等待 wait = int(resp.headers.get("Retry-After", "60")) + if not self.wait: + raise RateLimited(f"币安限流 {resp.status_code}，需等待 {wait}s") logger.warning("币安限流 %d，暂停 %ds", resp.status_code, wait) await asyncio.sleep(wait) continue @@ -53,7 +61,7 @@ class BinanceClient: continue resp.raise_for_status() used = int(resp.headers.get("X-MBX-USED-WEIGHT-1M", "0")) - if used > conf["weight_limit"] * WEIGHT_RATIO: + if self.wait and used > conf["weight_limit"] * WEIGHT_RATIO: wait = 61 - time.time() % 60 logger.info("%s 权重已用 %d，等待 %.0fs 到下一分钟", market, used, wait) await asyncio.sleep(wait) diff --git a/backend/src/alphapulse/market/worker.py b/backend/src/alphapulse/market/worker.py index 29f656c..9f447d4 100644 --- a/backend/src/alphapulse/market/worker.py +++ b/backend/src/alphapulse/market/worker.py @@ -100,6 +100,9 @@ async def next_backfill_task(market: str) -> tuple[MarketSymbol, str] | None: async def set_symbol_status(symbol_id: int, status: str, error: str | None = None) -> None: async with SessionLocal() as session: s = await session.get(MarketSymbol, symbol_id) + if s is None: + # 补齐过程中交易对被用户删除 + return s.status = status s.last_error = error await session.commit() @@ -107,15 +110,20 @@ async def set_symbol_status(symbol_id: int, status: str, error: str | None = Non async def backfill_loop(market: str) -> None: while True: - task = await next_backfill_task(market) + try: + task = await next_backfill_task(market) + except Exception: + logger.exception("%s 选取补齐任务出错，稍后重试", market) + await asyncio.sleep(10) + continue if task is None: await asyncio.sleep(5) continue s, interval = task - if s.status == "pending": - await set_symbol_status(s.id, "backfilling") - logger.info("%s %s 开始补齐", market, s.symbol) try: + if s.status == "pending": + await set_symbol_status(s.id, "backfilling") + logger.info("%s %s 开始补齐", market, s.symbol) for _ in range(PAGES_PER_TURN): if await sync_page(market, s.symbol, interval): logger.info("%s %s %s 补齐完成", market, s.symbol, interval) @@ -141,12 +149,16 @@ async def catchup_loop() -> None: except TimeoutError: pass catchup_needed.clear() - async with SessionLocal() as session: - states = (await session.execute( - select(KlineSyncState.market, KlineSyncState.symbol, KlineSyncState.interval) - .join(MarketSymbol, (MarketSymbol.market == KlineSyncState.market) & (MarketSymbol.symbol == KlineSyncState.symbol)) - .where(KlineSyncState.backfill_done, MarketSymbol.status.in_(ACTIVE_STATUSES)) - )).all() + try: + async with SessionLocal() as session: + states = (await session.execute( + select(KlineSyncState.market, KlineSyncState.symbol, KlineSyncState.interval) + .join(MarketSymbol, (MarketSymbol.market == KlineSyncState.market) & (MarketSymbol.symbol == KlineSyncState.symbol)) + .where(KlineSyncState.backfill_done, MarketSymbol.status.in_(ACTIVE_STATUSES)) + )).all() + except Exception: + logger.exception("读取追平列表失败，下次再试") + continue started = time.monotonic() for market, symbol, interval in states: try: @@ -186,13 +198,13 @@ async def handle_message(market: str, raw: str | bytes) -> None: async def stream_loop(market: str) -> None: while True: - symbols = await active_symbols(market) - if not symbols: - await asyncio.sleep(SYMBOL_CHECK_SECONDS) - continue - intervals = INTERVALS + (["1s"] if market == "spot" else []) - streams = [f"{s.lower()}@kline_{i}" for s in symbols for i in intervals] + [f"{s.lower()}@miniTicker" for s in symbols] try: + symbols = await active_symbols(market) + if not symbols: + await asyncio.sleep(SYMBOL_CHECK_SECONDS) + continue + intervals = INTERVALS + (["1s"] if market == "spot" else []) + streams = [f"{s.lower()}@kline_{i}" for s in symbols for i in intervals] + [f"{s.lower()}@miniTicker" for s in symbols] async with websockets.connect(f"{WS[market]}?streams={'/'.join(streams)}", max_size=2**22) as ws: logger.info("%s WS 已连接：%d 个交易对，%d 个流", market, len(symbols), len(streams)) catchup_needed.set() @@ -205,16 +217,17 @@ async def stream_loop(market: str) -> None: logger.info("%s 交易对列表变化，重新订阅", market) break except Exception: - logger.exception("%s WS 断开，5s 后重连", market) + logger.exception("%s WS 循环出错，5s 后重连", market) await asyncio.sleep(5) async def main() -> None: - await asyncio.gather( - *(backfill_loop(m) for m in MARKETS), - *(stream_loop(m) for m in MARKETS), - catchup_loop(), - ) + # 各循环自行重试可恢复的错误；仍有异常逃出时整个进程退出，由 docker 重启 + async with asyncio.TaskGroup() as tg: + for m in MARKETS: + tg.create_task(backfill_loop(m)) + tg.create_task(stream_loop(m)) + tg.create_task(catchup_loop()) if __name__ == "__main__": diff --git a/web/src/pages/market/useMarketSocket.ts b/web/src/pages/market/useMarketSocket.ts index 9bbdd61..7a0988a 100644 --- a/web/src/pages/market/useMarketSocket.ts +++ b/web/src/pages/market/useMarketSocket.ts @@ -36,8 +36,12 @@ export function useMarketSocket(onKline: (k: KlineMessage) => void, onTicker: (t let current: Subscription | undefined let retryMs = 1000 let closed = false + let retryTimer: ReturnType | undefined function connect() { + // 组件已卸载时不再建立连接 + if (closed) + return socket = new WebSocket(`${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/api/market/ws`) socket.onopen = () => { connected.value = true @@ -58,7 +62,7 @@ export function useMarketSocket(onKline: (k: KlineMessage) => void, onTicker: (t connected.value = false if (closed) return - setTimeout(connect, retryMs) + retryTimer = setTimeout(connect, retryMs) retryMs = Math.min(retryMs * 2, RETRY_MAX_MS) } } @@ -72,6 +76,7 @@ export function useMarketSocket(onKline: (k: KlineMessage) => void, onTicker: (t connect() onBeforeUnmount(() => { closed = true + clearTimeout(retryTimer) socket?.close() }) return { connected, subscribe }