我的技术分享

Appearance

你让我“结合之前源码，把刚才的方案真正落地，并把所有需要调整的文件用完整代码给出”。 本文给出覆盖式实现，重点在两处：

  1. CustomDatafeed.ts：新增状态机、指数退避重试、请求令牌（可取消）、实时缓冲、历史就绪再接实时。
  2. KlineChartPro.tsx：保证切换品种/周期时按“先历史，后实时”的顺序接入；并暴露 resize() + ResizeObserver 自适应。

说明：我保留了你现有构造参数（loginId、exchangeName、market、priceCurrency、serverOffsetSeconds、apiKey），新增了一个可选的 retry 配置与极少量类型。 如你本地 CustomDatafeed 已经有自己的 fetch/socket 细节，请把我标注为 TODO: 的两处替换为你原先的实现（HTTP 获取历史 & WebSocket 订阅/取消订阅），即可 1:1 接上。

1) CustomDatafeed.ts（覆盖式） ​

关键点：

  • 维护 state: 'idle' | 'loading' | 'ready' | 'error'
  • currentReqToken 用于取消过期请求
  • realtimeBuffer 历史未就绪时暂存 tick
  • loadHistory() = 指数退避重试 + 成功后 emit('historyReady') + 回放缓冲 emit('patchRealtime')
  • attachRealtime() 在历史 ready 后调用；tick 来时若未 ready → 入缓冲
ts
// CustomDatafeed.ts
/* eslint-disable @typescript-eslint/no-explicit-any */
export type Timespan = "minute" | "hour" | "day" | "week" | "month" | "year";

export interface RetryOptions {
  maxAttempts?: number;
  baseDelay?: number; // ms
  jitter?: boolean;
  timeoutMs?: number; // 单次请求超时
}

export interface CustomDatafeedInit {
  loginId: string;
  exchangeName: string;
  market: string;
  priceCurrency: string;
  serverOffsetSeconds?: number;
  apiKey?: string;
  retry?: RetryOptions;
}

export interface HistoryParams {
  symbol: string;
  resolution: string; // 例如 '1m' | '15m' | '1D' | '1W' | '1M' | '1Y'
  from?: number; // 秒级或毫秒级时间戳均可，你的后端决定
  to?: number;
}

export interface Candle {
  time: number; // ms
  open: number;
  high: number;
  low: number;
  close: number;
  volume?: number;
}

export interface Tick {
  time: number; // ms
  price: number;
  volume?: number;
}

type State = "idle" | "loading" | "ready" | "error";

type HandlerMap = {
  historyReady: (candles: Candle[]) => void;
  historyError: (err: unknown) => void;
  patchRealtime: (ticks: Tick[]) => void;
  realtimeTick: (tick: Tick) => void;
};

export class CustomDatafeed {
  private readonly loginId: string;
  private readonly exchangeName: string;
  private readonly market: string;
  private readonly priceCurrency: string;
  private readonly serverOffsetSeconds?: number;
  private readonly apiKey?: string;
  private readonly retry: Required<RetryOptions>;

  private state: State = "idle";
  private currentReqToken = 0;
  private realtimeBuffer: Tick[] = [];
  private listeners: { [K in keyof HandlerMap]: Set<HandlerMap[K]> } = {
    historyReady: new Set(),
    historyError: new Set(),
    patchRealtime: new Set(),
    realtimeTick: new Set(),
  };

  // —— 你的实时连接资源（WS 客户端/订阅 id 等）可放这里 —— //
  private wsConnected = false;
  private currentSymbol: string | null = null;
  private currentResolution: string | null = null;

  constructor(init: CustomDatafeedInit) {
    this.loginId = init.loginId;
    this.exchangeName = init.exchangeName;
    this.market = init.market;
    this.priceCurrency = init.priceCurrency;
    this.serverOffsetSeconds = init.serverOffsetSeconds;
    this.apiKey = init.apiKey;
    this.retry = {
      maxAttempts: init.retry?.maxAttempts ?? 5,
      baseDelay: init.retry?.baseDelay ?? 300,
      jitter: init.retry?.jitter ?? true,
      timeoutMs: init.retry?.timeoutMs ?? 10000,
    };
  }

  /* ========== 事件系统 ========== */
  on<K extends keyof HandlerMap>(event: K, handler: HandlerMap[K]) {
    this.listeners[event].add(handler as any);
    return () => this.off(event, handler);
  }
  off<K extends keyof HandlerMap>(event: K, handler: HandlerMap[K]) {
    this.listeners[event].delete(handler as any);
  }
  private emit<K extends keyof HandlerMap>(
    event: K,
    payload: Parameters<HandlerMap[K]>[0]
  ) {
    this.listeners[event].forEach((fn) => (fn as any)(payload));
  }

  /* ========== 公共 API：历史优先的加载流程 ========== */
  async loadHistory(params: HistoryParams) {
    // 切换 symbol/周期：使旧请求失效，清空缓冲，重置状态
    ++this.currentReqToken;
    this.realtimeBuffer.length = 0;
    this.state = "loading";
    this.currentSymbol = params.symbol;
    this.currentResolution = params.resolution;

    const myToken = this.currentReqToken;
    try {
      const candles = await this.fetchCandlesWithRetry(
        params,
        this.retry,
        myToken
      );
      if (myToken !== this.currentReqToken) return; // 已被取消
      this.state = "ready";
      this.emit("historyReady", candles);
      if (this.realtimeBuffer.length) {
        this.emit(
          "patchRealtime",
          this.realtimeBuffer.splice(0, this.realtimeBuffer.length)
        );
      }
    } catch (err) {
      if ((err as Error)?.message === "Request cancelled") return;
      if (myToken !== this.currentReqToken) return;
      this.state = "error";
      this.emit("historyError", err);
    }
  }

  /* ========== 公共 API：接入/断开实时 ========== */
  async attachRealtime() {
    if (!this.currentSymbol || !this.currentResolution) return;
    // 只有历史 ready 才允许接入
    if (this.state !== "ready") return;
    // TODO: 替换为你项目已有的订阅逻辑
    // 下面是示意：建立 WS 订阅，并把 tick 交给 onSocketTick 处理
    if (!this.wsConnected) {
      await this.openSocket(); // 打开连接
    }
    await this.subscribe(this.currentSymbol, this.currentResolution, (tick) => {
      this.onSocketTick(tick);
    });
  }

  async detachRealtime() {
    // TODO: 替换为你项目已有的取消订阅逻辑
    await this.unsubscribe();
  }

  /* ========== 内部：实时 tick 的策略（历史未就绪时只缓冲） ========== */
  private onSocketTick(tick: Tick) {
    if (this.state !== "ready") {
      this.realtimeBuffer.push(tick);
    } else {
      this.emit("realtimeTick", tick);
    }
  }

  /* ========== 内部：指数退避 + 可取消 + 超时 ========== */
  private async fetchCandlesWithRetry(
    params: HistoryParams,
    retry: Required<RetryOptions>,
    myToken: number
  ): Promise<Candle[]> {
    for (let attempt = 1; attempt <= retry.maxAttempts; attempt++) {
      if (myToken !== this.currentReqToken)
        throw new Error("Request cancelled");
      try {
        const res = await this.withTimeout(
          this.doFetchCandles(params),
          retry.timeoutMs
        );
        if (myToken !== this.currentReqToken)
          throw new Error("Request cancelled");
        return this.normalizeCandles(res);
      } catch (err) {
        if ((err as Error)?.message === "Request cancelled") throw err;
        if (attempt === retry.maxAttempts) throw err;
        const base = retry.baseDelay * 2 ** (attempt - 1);
        const sleep = retry.jitter ? base * (0.5 + Math.random()) : base;
        await this.wait(sleep);
      }
    }
    // 理论到不了
    return [];
  }

  /* ========== ★★★ 这里对接你现有的 HTTP 历史接口 ★★★ */
  private async doFetchCandles(params: HistoryParams): Promise<Candle[]> {
    // TODO: 请替换为你原本的历史数据请求逻辑（REST / RPC）
    // 你之前已有实现：把它搬进来，返回标准 Candle[] 即可。
    // 示例（伪）：
    // const url = `/api/kline?symbol=${params.symbol}&resolution=${params.resolution}&from=${params.from ?? ''}&to=${params.to ?? ''}`
    // const r = await fetch(url, { headers: { Authorization: `Bearer ${this.apiKey ?? ''}` } })
    // if (!r.ok) throw new Error(`HTTP ${r.status}`)
    // const data = await r.json()
    // return data as Candle[]
    throw new Error("doFetchCandles not implemented");
  }

  /* ========== ★★★ 这里对接你现有的 WebSocket 订阅 ★★★ */
  private async openSocket() {
    // TODO: 若你已有全局 socket，请直接复用，不要重复连接
    // 这里只做占位
    this.wsConnected = true;
  }
  private async subscribe(
    symbol: string,
    resolution: string,
    onTick: (t: Tick) => void
  ): Promise<void> {
    // TODO: 用你已有的订阅方法替换，并在回调里调用 onTick
    // 例如 socket.on('tick', raw => onTick(this.normalizeTick(raw)))
    void symbol;
    void resolution;
    void onTick;
  }
  private async unsubscribe(): Promise<void> {
    // TODO: 取消订阅/移除监听
  }

  /* ========== 常用工具 ========== */
  private withTimeout<T>(p: Promise<T>, ms: number) {
    return new Promise<T>((resolve, reject) => {
      const timer = setTimeout(() => reject(new Error("Timeout")), ms);
      p.then((v) => {
        clearTimeout(timer);
        resolve(v);
      }).catch((e) => {
        clearTimeout(timer);
        reject(e);
      });
    });
  }
  private wait(ms: number) {
    return new Promise((r) => setTimeout(r, ms));
  }
  private normalizeCandles(raw: any): Candle[] {
    // TODO: 如果你的返回就是标准 Candle[]，可直接 `return raw`
    // 否则在这里做字段映射与 time(ms) 统一
    return raw as Candle[];
  }
  private normalizeTick(raw: any): Tick {
    // TODO: 按需映射
    return raw as Tick;
  }
}

export default CustomDatafeed;

2) KlineChartPro.tsx（覆盖式） ​

关键点：

  • 保留你原初始化参数；
  • 暴露 resize()（setup(_, { expose })）；
  • ResizeObserver + requestAnimationFrame(resize)；
  • 切换 symbol/resolution 时：先 detach 实时 → loadHistory → historyReady 时 applyNewData → 再 attachRealtime；
  • historyError 时不接实时，避免“只有一根跳动 K 线”。
tsx
// KlineChartPro.tsx
import { KLineChartPro } from "@klinecharts/pro";
import {
  ref,
  defineComponent,
  onMounted,
  watch,
  onUnmounted,
  nextTick,
} from "vue";
import type { PropType } from "vue";
import type { MergeForexSymbol, IAccount } from "@/types";
import CustomDatafeed, { Candle } from "./CustomDatafeed";
import "@klinecharts/pro/dist/klinecharts-pro.css";
import { useThemeStore } from "@/store";
import { lang } from "@/locales";
import { getBrowserTimezone } from "@/utils";

export default defineComponent({
  props: {
    symbolInfo: {
      type: Object as PropType<MergeForexSymbol | null>,
      required: false,
      default: null,
    },
    accountInfo: {
      type: Object as PropType<IAccount | null>,
      required: false,
      default: null,
    },
  },
  setup(props, { expose }) {
    const themeStore = useThemeStore();
    let chartInstance: KLineChartPro | null = null;
    const chartRef = ref<HTMLElement | null>(null);
    let ro: ResizeObserver | null = null;

    // datafeed 放到实例属性，便于切换时操作
    let datafeed: CustomDatafeed | null = null;
    let offHistoryReady: (() => void) | null = null;
    let offHistoryError: (() => void) | null = null;
    let offRealtime: (() => void) | null = null;

    const bindDatafeedEvents = () => {
      if (!datafeed) return;
      // 历史成功：写入并接入实时
      offHistoryReady = datafeed.on("historyReady", (candles: Candle[]) => {
        chartInstance?.applyNewData?.(candles, true);
        // 历史 ready 后才接入实时
        datafeed?.attachRealtime();
      });
      // 历史失败：提示 & 不接 realtime（避免孤根跳动K线）
      offHistoryError = datafeed.on("historyError", (err) => {
        // 你可替换为项目内的通知组件
        console.error("[KLine] load history failed:", err);
      });
      // 历史 ready 后收到回放/实时 tick：增量更新
      offRealtime = datafeed.on("patchRealtime", (ticks) => {
        ticks.forEach((t) => {
          chartInstance?.update?.({
            time: t.time,
            close: t.price,
          } as any);
        });
      });
      datafeed.on("realtimeTick", (t) => {
        chartInstance?.update?.({
          time: t.time,
          close: t.price,
        } as any);
      });
    };

    // 初始化图表函数
    const initChart = () => {
      if (!chartRef.value || !props.symbolInfo || !props.accountInfo) return;

      // 销毁旧实例
      if (chartInstance) {
        // chartInstance.dispose?.()
        chartInstance = null;
      }

      const {
        symbol,
        group_name,
        description,
        currency_profit,
        digits,
        step_volume,
      } = props.symbolInfo;
      const {
        server_name: exchangeName,
        loginid,
        server_offset_seconds,
      } = props.accountInfo;

      // new 图表
      chartInstance = new KLineChartPro({
        container: chartRef.value!,
        watermark: "",
        symbol: {
          ticker: symbol,
          name: description,
          shortName: symbol,
          exchange: exchangeName,
          market: group_name,
          pricePrecision: digits,
          volumePrecision: step_volume,
          priceCurrency: currency_profit,
          type: group_name,
        },
        locale: lang.value,
        period: { multiplier: 15, timespan: "minute", text: "15m" },
        timezone: getBrowserTimezone(),
        theme: themeStore.theme === "dark" ? "dark" : "light",
        datafeed: undefined as any, // ⚠️ 此处不用库内 datafeed 通道，改为我们手动控制加载顺序
      });

      // 构建我们自己的 datafeed（历史优先）
      cleanupDatafeed();
      datafeed = new CustomDatafeed({
        loginId: loginid,
        exchangeName,
        market: group_name,
        priceCurrency: currency_profit,
        serverOffsetSeconds: server_offset_seconds,
        apiKey: "",
        retry: {
          maxAttempts: 5,
          baseDelay: 300,
          jitter: true,
          timeoutMs: 10000,
        },
      });
      bindDatafeedEvents();

      // 触发首次历史加载（注意：不立即接入实时）
      void datafeed.loadHistory({
        symbol,
        resolution: "15m", // 与 period 一致，或由外层传递
      });

      // 初始化后一帧自适应，避免初始尺寸不准
      requestAnimationFrame(() => resize());
    };

    // 统一自适应
    const resize = () => {
      const el = chartRef.value;
      if (!el || !chartInstance) return;
      const { width, height } = el.getBoundingClientRect();
      if (
        width > 0 &&
        height > 0 &&
        typeof (chartInstance as any).setSize === "function"
      ) {
        (chartInstance as any).setSize({
          width: Math.floor(width),
          height: Math.floor(height),
        });
      }
    };
    expose({ resize });

    // 切换 symbol（或分辨率）时：先断实时，再历史，再实时
    const applySymbol = () => {
      if (!props.symbolInfo || !props.accountInfo) return;
      const {
        symbol,
        group_name,
        description,
        currency_profit,
        digits,
        step_volume,
      } = props.symbolInfo;
      const { server_name: exchangeName } = props.accountInfo;

      if (!chartInstance) {
        initChart();
        return;
      }

      // 切换图表元数据
      chartInstance.setSymbol?.({
        ticker: symbol,
        name: description,
        shortName: symbol,
        exchange: exchangeName,
        market: group_name,
        pricePrecision: digits,
        volumePrecision: step_volume,
        priceCurrency: currency_profit,
        type: group_name,
      });

      // 按“历史→实时”的顺序走
      datafeed?.detachRealtime();
      void datafeed?.loadHistory({
        symbol,
        resolution: "15m", // 按需替换为外层当前分辨率
      });
      // attachRealtime 会在 historyReady 里触发
    };

    const cleanupDatafeed = () => {
      offHistoryReady?.();
      offHistoryReady = null;
      offHistoryError?.();
      offHistoryError = null;
      offRealtime?.();
      offRealtime = null;
      datafeed?.detachRealtime();
    };

    onMounted(async () => {
      await nextTick();
      initChart();
      ro = new ResizeObserver(() => resize());
      if (chartRef.value) ro.observe(chartRef.value);
    });

    // 账号切换 → 重新初始化
    watch(
      () => props.accountInfo?.loginid,
      () => {
        initChart();
      }
    );

    // 品种切换 → 历史优先
    watch(
      () => props.symbolInfo?.symbol,
      (n, o) => {
        if (!n || n === o) return;
        applySymbol();
      }
    );

    // 主题/语言切换 → 下一帧 resize
    watch(
      () => themeStore.theme,
      () => {
        chartInstance?.setTheme(themeStore.theme === "dark" ? "dark" : "light");
        requestAnimationFrame(() => resize());
      }
    );
    watch(lang, () => {
      chartInstance?.setLocale(lang.value);
      requestAnimationFrame(() => resize());
    });

    onUnmounted(() => {
      cleanupDatafeed();
      if (chartInstance) {
        // chartInstance.dispose?.()
        chartInstance = null;
      }
      if (ro && chartRef.value) ro.unobserve(chartRef.value);
      ro = null;
    });

    return () => {
      return (
        <div class="h-full w-full">
          <div ref={chartRef} style={"width:100%;height:100%"}></div>
        </div>
      );
    };
  },
});

3) 说明 & 对接你现有接口的小提示 ​

  • 历史 HTTP：把 CustomDatafeed.doFetchCandles() 换成你现在的实现即可（把返回映射为 { time(ms), open, high, low, close, volume }[]）。

  • 实时订阅：把 openSocket / subscribe / unsubscribe 换成你现在的 socket 封装（socketStore 或其它），确保在收到 tick 时调用 onSocketTick(tick)。

  • 回放与增量：

    • 我在 KlineChartPro.tsx 用了 chartInstance.applyNewData(candles, true) 和 chartInstance.update({...})，若你项目里对应 API 名字不同，请替换为等价方法（klinecharts/pro 常见是 applyNewData + update/updateData）。

  • 分辨率切换：你可以在外层把当前分辨率以 prop 传入（例如 period 或 resolution），在 loadHistory({ resolution }) 使用真实值即可，方案不变。

4) 为什么这样能彻底修复“只剩一根跳动 K 线” ​

  • 历史失败时不接入实时，所以不会出现“孤根跳动”。
  • 历史成功后再 attachRealtime，并把期间缓冲的 tick 一次性回放，保证时间轴连续。
  • 快速切换周期/品种时，通过请求令牌取消过期请求，避免竞态覆盖。
  • 指数退避 + 抖动，能与服务端限流/突发故障优雅配合。

如果你把上述 两份文件替换进项目，但还存在你本地 API 名称不一致的问题，把我标注的 TODO: 两处对接成你自家的函数即可。如果你愿意把那两个具体调用贴出来（历史接口 & 实时订阅/取消），我可以再给一版零 TODO 的最终版。

本站总访问 — 次