Websocket代码示例

WebSocket代码示例，获取实时成交明细、盘口、k线。包括自动重连，心跳检测机制。

package org.example.ws;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.websocket.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.URI;
import java.util.concurrent.*;

@ClientEndpoint
@Slf4j
@Component
public class WebsocketExample {

    // 本地session通道（使用volatile保证多线程可见性）
    private volatile Session session;

    // wss连接地址 business可以为stock、crypto、common；apikey为您的凭证
    private static final String WS_URL = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey";

    // 定时任务池（统一管理，避免创建多个线程池）
    private ScheduledExecutorService scheduledExecutorService;

    @PostConstruct
    public void connectAll() {
        // 初始化线程池（核心线程1个，避免资源浪费）
        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        
        try {
            // 建立WEBSOCKET连接
            connect(WS_URL);
            // 开启自动重连（延迟1秒启动，每10秒检查一次）
            startReconnection(WS_URL);
        } catch (Exception e) {
            log.error("初始化WebSocket连接失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 自动重连机制
     * @param wsUrl WebSocket连接地址
     */
    private void startReconnection(String wsUrl) {
        Runnable reconnectTask = () -> {
            if (session == null || !session.isOpen()) {
                log.warn("WebSocket连接已断开，开始重连...");
                connect(wsUrl);
            }
        };
        // 延迟1秒启动，每10秒执行一次重连检查
        scheduledExecutorService.scheduleAtFixedRate(reconnectTask, 1, 10, TimeUnit.SECONDS);
    }

    /**
     * 建立WEBSOCKET连接具体实现
     * @param wsUrl 连接地址
     */
    private void connect(String wsUrl) {
        try {
            WebSocketContainer container = ContainerProvider.getWebSocketContainer();
            // 关闭旧连接（防止连接泄露）
            if (session != null && session.isOpen()) {
                session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "重新连接"));
            }
            // 建立新连接
            session = container.connectToServer(this, URI.create(wsUrl));
            log.info("WebSocket连接成功，sessionId: {}", session.getId());
        } catch (DeploymentException | IOException e) {
            log.error("WebSocket连接失败: {}", e.getMessage(), e);
            // 连接失败时清空session，触发下次重连
            session = null;
        }
    }

    /**
     * WebSocket连接建立成功后触发
     */
    @OnOpen
    public void onOpen(Session session) {
        // 更新全局session
        this.session = session;
        log.info("Connection opened: {}", session.getId());

        // 异步发送订阅请求（避免阻塞OnOpen线程）
        scheduledExecutorService.submit(() -> {
            try {
                // 1. 订阅实时成交明细（协议号10000）
                sendTradeSubscribe(session);
                TimeUnit.MILLISECONDS.sleep(5000); // 间隔5秒

                // 2. 订阅实时盘口数据（协议号10003）
                sendDepthSubscribe(session);
                TimeUnit.MILLISECONDS.sleep(5000); // 间隔5秒

                // 3. 订阅1分钟K线数据（协议号10006）
                sendKlineSubscribe(session);

                // 4. 启动心跳任务（30秒一次）
                startHeartbeatTask();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // 恢复中断状态
                log.error("发送订阅请求时线程被中断: {}", e.getMessage());
            } catch (IOException e) {
                log.error("发送订阅请求失败: {}", e.getMessage(), e);
            }
        });
    }

    /**
     * 发送实时成交明细订阅请求
     */
    private void sendTradeSubscribe(Session session) throws IOException {
        JSONObject tradeSendObj = new JSONObject();
        tradeSendObj.put("code", 10000); // 订阅成交明细协议号
        tradeSendObj.put("trace", generateTraceId()); // 自定义traceId
        JSONObject data = new JSONObject();
        data.put("codes", "BTCUSDT"); // 订阅BTCUSDT
        tradeSendObj.put("data", data);
        session.getBasicRemote().sendText(tradeSendObj.toJSONString());
        log.info("发送成交明细订阅请求: {}", tradeSendObj);
    }

    /**
     * 发送实时盘口数据订阅请求
     */
    private void sendDepthSubscribe(Session session) throws IOException {
        JSONObject depthSendObj = new JSONObject();
        depthSendObj.put("code", 10003); // 订阅盘口协议号
        depthSendObj.put("trace", generateTraceId());
        JSONObject data = new JSONObject();
        data.put("codes", "BTCUSDT");
        depthSendObj.put("data", data);
        session.getBasicRemote().sendText(depthSendObj.toJSONString());
        log.info("发送盘口数据订阅请求: {}", depthSendObj);
    }

    /**
     * 发送实时K线数据订阅请求
     */
    private void sendKlineSubscribe(Session session) throws IOException {
        JSONObject klineSendObj = new JSONObject();
        klineSendObj.put("code", 10006); // 订阅K线协议号
        klineSendObj.put("trace", generateTraceId());
        
        JSONObject klineData = new JSONObject();
        JSONArray klineDataArray = new JSONArray();
        
        JSONObject kline1minObj = new JSONObject();
        kline1minObj.put("type", 1); // 1分钟K线
        kline1minObj.put("codes", "BTCUSDT");
        klineDataArray.add(kline1minObj);
        klineData.put("arr", klineDataArray);
        
        klineSendObj.put("data", klineData);
        session.getBasicRemote().sendText(klineSendObj.toJSONString());
        log.info("发送K线数据订阅请求: {}", klineSendObj);
    }

    /**
     * 启动心跳保活任务
     */
    private void startHeartbeatTask() {
        Runnable heartbeatTask = () -> {
            try {
                if (session != null && session.isOpen()) {
                    JSONObject pingObj = new JSONObject();
                    pingObj.put("code", 10010); // 心跳协议号
                    pingObj.put("trace", generateTraceId());
                    session.getBasicRemote().sendText(pingObj.toJSONString());
                    log.debug("发送心跳包: {}", pingObj);
                }
            } catch (IOException e) {
                log.error("发送心跳包失败: {}", e.getMessage(), e);
            }
        };
        // 延迟30秒启动，每30秒发送一次心跳
        scheduledExecutorService.scheduleAtFixedRate(heartbeatTask, 30, 30, TimeUnit.SECONDS);
    }

    /**
     * 接收服务端推送的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到服务端消息: {}", message);
        // 这里可以解析消息并处理业务逻辑
        // 示例：解析JSON消息
        try {
            JSONObject msgObj = JSONObject.parseObject(message);
            Integer code = msgObj.getInteger("code");
            String trace = msgObj.getString("trace");
            JSONObject data = msgObj.getJSONObject("data");
            
            // 根据协议号区分消息类型
            switch (code) {
                case 10000: // 成交明细数据
                    handleTradeData(data);
                    break;
                case 10003: // 盘口数据
                    handleDepthData(data);
                    break;
                case 10006: // K线数据
                    handleKlineData(data);
                    break;
                case 10010: // 心跳响应
                    log.debug("收到心跳响应，trace: {}", trace);
                    break;
                default:
                    log.warn("未知协议号的消息: {}", code);
            }
        } catch (Exception e) {
            log.error("解析消息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 处理成交明细数据
     */
    private void handleTradeData(JSONObject data) {
        // 实现业务逻辑，比如入库、推送前端等
        log.info("处理成交明细数据: {}", data);
    }

    /**
     * 处理盘口数据
     */
    private void handleDepthData(JSONObject data) {
        // 实现业务逻辑
        log.info("处理盘口数据: {}", data);
    }

    /**
     * 处理K线数据
     */
    private void handleKlineData(JSONObject data) {
        // 实现业务逻辑
        log.info("处理K线数据: {}", data);
    }

    /**
     * 连接关闭时触发
     */
    @OnClose
    public void onClose(Session session, CloseReason reason) {
        log.info("Connection closed: {}, reason: {}", session.getId(), reason);
        // 清空session，触发重连
        this.session = null;
    }

    /**
     * 连接出错时触发
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("WebSocket错误，sessionId: {}", session.getId(), error);
        // 清空session，触发重连
        this.session = null;
    }

    /**
     * 生成唯一traceId（示例实现，可替换为UUID）
     */
    private String generateTraceId() {
        return java.util.UUID.randomUUID().toString();
    }

    /**
     * 销毁Bean时关闭线程池和连接
     */
    @PreDestroy
    public void destroy() {
        log.info("销毁WebSocket客户端，关闭资源...");
        // 关闭session
        if (session != null && session.isOpen()) {
            try {
                session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "应用关闭"));
            } catch (IOException e) {
                log.error("关闭session失败", e);
            }
        }
        // 关闭线程池
        if (scheduledExecutorService != null) {
            scheduledExecutorService.shutdown();
            try {
                if (!scheduledExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduledExecutorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduledExecutorService.shutdownNow();
            }
        }
    }
}

import asyncio
import json
import uuid
import logging
from typing import Optional
import websockets
from websockets.exceptions import ConnectionClosed, WebSocketException

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("websocket-client")

class CryptoWebsocketClient:
    """加密货币行情WebSocket客户端（对等Java版本逻辑）"""
    
    def __init__(self, api_key: str, business: str = "crypto"):
        # WebSocket连接配置
        self.ws_url = f"wss://data.infoway.io/ws?business={business}&apikey={api_key}"
        # 核心状态
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self.is_connected = False
        self.reconnect_interval = 10  # 重连间隔（秒）
        self.heartbeat_interval = 30  # 心跳间隔（秒）
        # 任务对象（用于管理异步任务）
        self.reconnect_task: Optional[asyncio.Task] = None
        self.heartbeat_task: Optional[asyncio.Task] = None

    async def connect(self) -> None:
        """建立WebSocket连接"""
        try:
            # 关闭旧连接（防止连接泄露）
            if self.ws and not self.ws.closed:
                await self.ws.close()
            
            # 建立新连接
            self.ws = await websockets.connect(self.ws_url)
            self.is_connected = True
            logger.info(f"WebSocket连接成功，地址: {self.ws_url}")
            
            # 连接成功后启动订阅和心跳
            await self._send_all_subscribe_requests()
            self._start_heartbeat_task()
        
        except WebSocketException as e:
            self.is_connected = False
            logger.error(f"WebSocket连接失败: {str(e)}")
            raise

    async def _send_all_subscribe_requests(self) -> None:
        """发送所有订阅请求（成交明细、盘口、K线）"""
        if not self.ws or self.ws.closed:
            logger.error("连接未建立，无法发送订阅请求")
            return
        
        try:
            # 1. 订阅实时成交明细（协议号10000）
            await self._send_trade_subscribe()
            await asyncio.sleep(5)  # 间隔5秒
            
            # 2. 订阅实时盘口数据（协议号10003）
            await self._send_depth_subscribe()
            await asyncio.sleep(5)  # 间隔5秒
            
            # 3. 订阅1分钟K线数据（协议号10006）
            await self._send_kline_subscribe()
            
        except Exception as e:
            logger.error(f"发送订阅请求失败: {str(e)}")
            raise

    def _generate_trace_id(self) -> str:
        """生成唯一trace ID（对等Java的UUID）"""
        return str(uuid.uuid4())

    async def _send_trade_subscribe(self) -> None:
        """发送实时成交明细订阅请求"""
        subscribe_msg = {
            "code": 10000,
            "trace": self._generate_trace_id(),
            "data": {"codes": "BTCUSDT"}
        }
        await self.ws.send(json.dumps(subscribe_msg))
        logger.info(f"发送成交明细订阅请求: {subscribe_msg}")

    async def _send_depth_subscribe(self) -> None:
        """发送实时盘口数据订阅请求"""
        subscribe_msg = {
            "code": 10003,
            "trace": self._generate_trace_id(),
            "data": {"codes": "BTCUSDT"}
        }
        await self.ws.send(json.dumps(subscribe_msg))
        logger.info(f"发送盘口数据订阅请求: {subscribe_msg}")

    async def _send_kline_subscribe(self) -> None:
        """发送1分钟K线数据订阅请求"""
        subscribe_msg = {
            "code": 10006,
            "trace": self._generate_trace_id(),
            "data": {
                "arr": [
                    {"type": 1, "codes": "BTCUSDT"}  # type=1 表示1分钟K线
                ]
            }
        }
        await self.ws.send(json.dumps(subscribe_msg))
        logger.info(f"发送K线数据订阅请求: {subscribe_msg}")

    async def _send_heartbeat(self) -> None:
        """发送心跳包（协议号10010）"""
        if not self.ws or self.ws.closed:
            return
        
        heartbeat_msg = {
            "code": 10010,
            "trace": self._generate_trace_id()
        }
        try:
            await self.ws.send(json.dumps(heartbeat_msg))
            logger.debug(f"发送心跳包: {heartbeat_msg}")
        except Exception as e:
            logger.error(f"发送心跳包失败: {str(e)}")
            raise

    def _start_heartbeat_task(self) -> None:
        """启动心跳任务（后台定时发送）"""
        if self.heartbeat_task and not self.heartbeat_task.done():
            self.heartbeat_task.cancel()
        
        async def heartbeat_loop():
            while self.is_connected and self.ws and not self.ws.closed:
                try:
                    await self._send_heartbeat()
                    await asyncio.sleep(self.heartbeat_interval)
                except Exception as e:
                    logger.error(f"心跳任务异常: {str(e)}")
                    break
        
        self.heartbeat_task = asyncio.create_task(heartbeat_loop())

    async def _message_listener(self) -> None:
        """监听服务端推送的消息"""
        while self.is_connected and self.ws and not self.ws.closed:
            try:
                # 阻塞等待接收消息
                message = await self.ws.recv()
                logger.info(f"收到服务端消息: {message}")
                
                # 解析消息并处理（对等Java的OnMessage逻辑）
                self._handle_received_message(message)
                
            except ConnectionClosed:
                logger.warning("WebSocket连接已关闭，停止消息监听")
                self.is_connected = False
                break
            except Exception as e:
                logger.error(f"接收/处理消息异常: {str(e)}")

    def _handle_received_message(self, message: str) -> None:
        """处理接收到的消息（按协议号分类）"""
        try:
            msg_data = json.loads(message)
            code = msg_data.get("code")
            trace = msg_data.get("trace")
            data = msg_data.get("data", {})
            
            if code == 10000:
                logger.info(f"处理成交明细数据 [trace={trace}]: {data}")
            elif code == 10003:
                logger.info(f"处理盘口数据 [trace={trace}]: {data}")
            elif code == 10006:
                logger.info(f"处理K线数据 [trace={trace}]: {data}")
            elif code == 10010:
                logger.debug(f"收到心跳响应 [trace={trace}]")
            else:
                logger.warning(f"未知协议号消息 [code={code}]: {message}")
                
        except json.JSONDecodeError:
            logger.error(f"消息格式错误，无法解析JSON: {message}")
        except Exception as e:
            logger.error(f"处理消息异常: {str(e)}")

    async def _reconnect_loop(self) -> None:
        """自动重连循环（对等Java的startReconnection）"""
        while True:
            if not self.is_connected:
                logger.info(f"尝试重连WebSocket（间隔{self.reconnect_interval}秒）...")
                try:
                    await self.connect()
                except Exception:
                    # 重连失败，等待后重试
                    await asyncio.sleep(self.reconnect_interval)
            else:
                # 连接正常，短暂等待后再次检查
                await asyncio.sleep(1)

    async def start(self) -> None:
        """启动客户端（主入口）"""
        # 启动自动重连任务
        self.reconnect_task = asyncio.create_task(self._reconnect_loop())
        
        try:
            # 首次连接
            await self.connect()
            
            # 启动消息监听
            await self._message_listener()
            
        finally:
            # 清理资源
            self.is_connected = False
            if self.heartbeat_task:
                self.heartbeat_task.cancel()
            if self.reconnect_task:
                self.reconnect_task.cancel()
            if self.ws and not self.ws.closed:
                await self.ws.close()
            logger.info("WebSocket客户端已停止")

async def main():
    """主函数"""
    # 替换为你的实际API Key
    API_KEY = "yourApikey"
    client = CryptoWebsocketClient(api_key=API_KEY)
    await client.start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("用户中断，退出客户端")
    except Exception as e:
        logger.error(f"客户端运行异常: {str(e)}")

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"net/url"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

// CryptoWebsocketClient 加密货币行情WebSocket客户端
type CryptoWebsocketClient struct {
	apiKey             string        // 认证API Key
	business           string        // 业务类型：stock/crypto/common
	wsURL              string        // WebSocket连接地址
	conn               *websocket.Conn // WebSocket连接实例
	isConnected        bool          // 连接状态
	reconnectInterval  time.Duration // 重连间隔
	heartbeatInterval  time.Duration // 心跳间隔
	done               chan struct{} // 退出信号通道
}

// NewCryptoWebsocketClient 创建客户端实例
func NewCryptoWebsocketClient(apiKey, business string) *CryptoWebsocketClient {
	// 构建WS URL
	params := url.Values{}
	params.Set("business", business)
	params.Set("apikey", apiKey)
	wsURL := fmt.Sprintf("wss://data.infoway.io/ws?%s", params.Encode())

	return &CryptoWebsocketClient{
		apiKey:             apiKey,
		business:           business,
		wsURL:              wsURL,
		reconnectInterval:  10 * time.Second,  // 10秒重连间隔
		heartbeatInterval:  30 * time.Second,  // 30秒心跳间隔
		done:               make(chan struct{}),
	}
}

// generateTraceID 生成唯一的trace ID（对齐Java/Python的UUID）
func (c *CryptoWebsocketClient) generateTraceID() string {
	return uuid.New().String()
}

// connect 建立WebSocket连接
func (c *CryptoWebsocketClient) connect() error {
	// 关闭旧连接
	if c.conn != nil {
		c.conn.Close()
		c.conn = nil
	}

	// 建立新连接
	conn, _, err := websocket.DefaultDialer.Dial(c.wsURL, nil)
	if err != nil {
		c.isConnected = false
		return fmt.Errorf("连接失败: %w", err)
	}

	c.conn = conn
	c.isConnected = true
	log.Printf("WebSocket连接成功: %s", c.wsURL)

	// 连接成功后发送订阅请求
	go c.sendAllSubscribeRequests()

	// 启动心跳任务
	go c.startHeartbeat()

	// 启动消息监听
	go c.listenMessages()

	return nil
}

// sendAllSubscribeRequests 发送所有订阅请求（成交明细、盘口、K线）
func (c *CryptoWebsocketClient) sendAllSubscribeRequests() {
	if !c.isConnected || c.conn == nil {
		log.Println("连接未建立，无法发送订阅请求")
		return
	}

	// 1. 订阅实时成交明细（协议号10000）
	if err := c.sendTradeSubscribe(); err != nil {
		log.Printf("发送成交明细订阅失败: %v", err)
		return
	}
	time.Sleep(5 * time.Second) // 间隔5秒

	// 2. 订阅实时盘口数据（协议号10003）
	if err := c.sendDepthSubscribe(); err != nil {
		log.Printf("发送盘口订阅失败: %v", err)
		return
	}
	time.Sleep(5 * time.Second) // 间隔5秒

	// 3. 订阅1分钟K线数据（协议号10006）
	if err := c.sendKlineSubscribe(); err != nil {
		log.Printf("发送K线订阅失败: %v", err)
		return
	}
}

// sendTradeSubscribe 发送实时成交明细订阅请求
func (c *CryptoWebsocketClient) sendTradeSubscribe() error {
	msg := map[string]interface{}{
		"code":  10000,
		"trace": c.generateTraceID(),
		"data": map[string]string{
			"codes": "BTCUSDT",
		},
	}
	return c.sendJSONMessage(msg)
}

// sendDepthSubscribe 发送实时盘口数据订阅请求
func (c *CryptoWebsocketClient) sendDepthSubscribe() error {
	msg := map[string]interface{}{
		"code":  10003,
		"trace": c.generateTraceID(),
		"data": map[string]string{
			"codes": "BTCUSDT",
		},
	}
	return c.sendJSONMessage(msg)
}

// sendKlineSubscribe 发送1分钟K线数据订阅请求
func (c *CryptoWebsocketClient) sendKlineSubscribe() error {
	msg := map[string]interface{}{
		"code":  10006,
		"trace": c.generateTraceID(),
		"data": map[string]interface{}{
			"arr": []map[string]interface{}{
				{
					"type":  1,        // 1分钟K线
					"codes": "BTCUSDT",
				},
			},
		},
	}
	return c.sendJSONMessage(msg)
}

// sendHeartbeat 发送心跳包（协议号10010）
func (c *CryptoWebsocketClient) sendHeartbeat() error {
	msg := map[string]interface{}{
		"code":  10010,
		"trace": c.generateTraceID(),
	}
	return c.sendJSONMessage(msg)
}

// sendJSONMessage 发送JSON格式消息
func (c *CryptoWebsocketClient) sendJSONMessage(msg interface{}) error {
	if !c.isConnected || c.conn == nil {
		return fmt.Errorf("连接已断开")
	}

	// 序列化为JSON
	data, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("消息序列化失败: %w", err)
	}

	// 发送消息
	if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil {
		c.isConnected = false
		return fmt.Errorf("发送消息失败: %w", err)
	}

	log.Printf("发送消息: %s", string(data))
	return nil
}

// startHeartbeat 启动心跳保活任务
func (c *CryptoWebsocketClient) startHeartbeat() {
	ticker := time.NewTicker(c.heartbeatInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// 定时发送心跳
			if err := c.sendHeartbeat(); err != nil {
				log.Printf("发送心跳失败: %v", err)
				return
			}
			log.Printf("发送心跳包成功")
		case <-c.done:
			// 收到退出信号，停止心跳
			return
		}
	}
}

// listenMessages 监听服务端推送的消息
func (c *CryptoWebsocketClient) listenMessages() {
	for {
		if !c.isConnected || c.conn == nil {
			return
		}

		// 读取消息
		_, msgData, err := c.conn.ReadMessage()
		if err != nil {
			c.isConnected = false
			log.Printf("读取消息失败/连接断开: %v", err)
			return
		}

		// 打印原始消息
		log.Printf("收到服务端消息: %s", string(msgData))

		// 解析并处理消息
		c.handleReceivedMessage(msgData)
	}
}

// handleReceivedMessage 处理接收到的消息（按协议号分类）
func (c *CryptoWebsocketClient) handleReceivedMessage(msgData []byte) {
	var msg map[string]interface{}
	if err := json.Unmarshal(msgData, &msg); err != nil {
		log.Printf("解析消息失败: %v", err)
		return
	}

	// 提取核心字段
	code := int(msg["code"].(float64)) // JSON数字默认解析为float64
	trace := msg["trace"].(string)
	data := msg["data"]

	// 按协议号处理不同类型数据
	switch code {
	case 10000:
		log.Printf("处理成交明细数据 [trace=%s]: %+v", trace, data)
	case 10003:
		log.Printf("处理盘口数据 [trace=%s]: %+v", trace, data)
	case 10006:
		log.Printf("处理K线数据 [trace=%s]: %+v", trace, data)
	case 10010:
		log.Printf("收到心跳响应 [trace=%s]", trace)
	default:
		log.Printf("未知协议号消息 [code=%d]: %s", code, string(msgData))
	}
}

// startReconnectLoop 启动自动重连循环
func (c *CryptoWebsocketClient) startReconnectLoop() {
	for {
		select {
		case <-c.done:
			// 收到退出信号，停止重连
			return
		default:
			if !c.isConnected {
				log.Printf("尝试重连WebSocket（间隔%v）...", c.reconnectInterval)
				// 重连前随机延迟（避免高频重试）
				time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
				if err := c.connect(); err != nil {
					log.Printf("重连失败: %v", err)
					time.Sleep(c.reconnectInterval)
					continue
				}
			}
			// 连接正常时，短暂休眠后再次检查
			time.Sleep(1 * time.Second)
		}
	}
}

// Start 启动客户端
func (c *CryptoWebsocketClient) Start() {
	// 启动自动重连循环
	go c.startReconnectLoop()

	// 首次连接
	if err := c.connect(); err != nil {
		log.Printf("首次连接失败: %v", err)
	}

	// 监听系统退出信号（Ctrl+C）
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan

	// 收到退出信号，清理资源
	log.Println("收到退出信号，关闭客户端...")
	close(c.done)
	if c.conn != nil {
		c.conn.Close()
	}
	c.isConnected = false
}

func main() {
	// 替换为你的实际API Key
	apiKey := "yourApikey"
	// 业务类型：crypto/stock/common
	business := "crypto"

	// 创建客户端
	client := NewCryptoWebsocketClient(apiKey, business)

	// 启动客户端
	client.Start()
}

<?php

require_once __DIR__ . '/vendor/autoload.php';

use WebSocket\Client;
use WebSocket\ConnectionException;

/**
 * 加密货币行情WebSocket客户端
 */
class CryptoWebsocketClient
{
    // 配置项
    private $apiKey;             // 认证API Key
    private $business;           // 业务类型：stock/crypto/common
    private $wsUrl;              // WebSocket连接地址
    private $reconnectInterval;  // 重连间隔（秒）
    private $heartbeatInterval;  // 心跳间隔（秒）
    
    // 运行状态
    private $client;             // WebSocket客户端实例
    private $isConnected = false;// 连接状态
    private $running = true;     // 客户端运行状态

    /**
     * 构造函数
     * @param string $apiKey API密钥
     * @param string $business 业务类型
     */
    public function __construct(string $apiKey, string $business = 'crypto')
    {
        $this->apiKey = $apiKey;
        $this->business = $business;
        // 构建WS URL
        $this->wsUrl = sprintf(
            'wss://data.infoway.io/ws?business=%s&apikey=%s',
            urlencode($business),
            urlencode($apiKey)
        );
        $this->reconnectInterval = 10;  // 10秒重连间隔
        $this->heartbeatInterval = 30;  // 30秒心跳间隔
    }

    /**
     * 生成唯一Trace ID
     * @return string
     */
    private function generateTraceId(): string
    {
        return sprintf(
            '%04x%04x-%04x-%04x-%04x-%04x%04x%04x',
            mt_rand(0, 0xffff),
            mt_rand(0, 0xffff),
            mt_rand(0, 0xffff),
            mt_rand(0, 0x0fff) | 0x4000,
            mt_rand(0, 0x3fff) | 0x8000,
            mt_rand(0, 0xffff),
            mt_rand(0, 0xffff),
            mt_rand(0, 0xffff)
        );
    }

    /**
     * 建立WebSocket连接
     * @return bool
     */
    private function connect(): bool
    {
        try {
            // 关闭旧连接
            if ($this->client) {
                try {
                    $this->client->close();
                } catch (Exception $e) {
                    // 忽略关闭异常
                }
                $this->client = null;
            }

            // 创建新连接
            $this->client = new Client($this->wsUrl, [
                'timeout' => 30,
                'fragment_size' => 8192,
            ]);
            $this->isConnected = true;
            echo date('Y-m-d H:i:s') . " - WebSocket连接成功: {$this->wsUrl}\n";
            
            // 发送所有订阅请求
            $this->sendAllSubscribeRequests();
            
            return true;
        } catch (ConnectionException $e) {
            $this->isConnected = false;
            echo date('Y-m-d H:i:s') . " - WebSocket连接失败: {$e->getMessage()}\n";
            return false;
        } catch (Exception $e) {
            $this->isConnected = false;
            echo date('Y-m-d H:i:s') . " - 连接异常: {$e->getMessage()}\n";
            return false;
        }
    }

    /**
     * 发送所有订阅请求
     */
    private function sendAllSubscribeRequests(): void
    {
        if (!$this->isConnected || !$this->client) {
            echo date('Y-m-d H:i:s') . " - 连接未建立，无法发送订阅请求\n";
            return;
        }

        try {
            // 1. 订阅实时成交明细（协议号10000）
            $this->sendTradeSubscribe();
            sleep(5); // 间隔5秒

            // 2. 订阅实时盘口数据（协议号10003）
            $this->sendDepthSubscribe();
            sleep(5); // 间隔5秒

            // 3. 订阅1分钟K线数据（协议号10006）
            $this->sendKlineSubscribe();
        } catch (Exception $e) {
            echo date('Y-m-d H:i:s') . " - 发送订阅请求失败: {$e->getMessage()}\n";
            $this->isConnected = false;
        }
    }

    /**
     * 发送成交明细订阅请求
     */
    private function sendTradeSubscribe(): void
    {
        $msg = [
            'code'  => 10000,
            'trace' => $this->generateTraceId(),
            'data'  => ['codes' => 'BTCUSDT']
        ];
        $this->sendJsonMessage($msg);
    }

    /**
     * 发送盘口数据订阅请求
     */
    private function sendDepthSubscribe(): void
    {
        $msg = [
            'code'  => 10003,
            'trace' => $this->generateTraceId(),
            'data'  => ['codes' => 'BTCUSDT']
        ];
        $this->sendJsonMessage($msg);
    }

    /**
     * 发送K线数据订阅请求
     */
    private function sendKlineSubscribe(): void
    {
        $msg = [
            'code'  => 10006,
            'trace' => $this->generateTraceId(),
            'data'  => [
                'arr' => [
                    ['type' => 1, 'codes' => 'BTCUSDT'] // 1分钟K线
                ]
            ]
        ];
        $this->sendJsonMessage($msg);
    }

    /**
     * 发送心跳包
     */
    private function sendHeartbeat(): void
    {
        $msg = [
            'code'  => 10010,
            'trace' => $this->generateTraceId()
        ];
        $this->sendJsonMessage($msg);
    }

    /**
     * 发送JSON格式消息
     * @param array $msg 消息数组
     */
    private function sendJsonMessage(array $msg): void
    {
        if (!$this->isConnected || !$this->client) {
            return;
        }

        try {
            $jsonStr = json_encode($msg, JSON_UNESCAPED_UNICODE);
            $this->client->send($jsonStr);
            echo date('Y-m-d H:i:s') . " - 发送消息: {$jsonStr}\n";
        } catch (ConnectionException $e) {
            $this->isConnected = false;
            echo date('Y-m-d H:i:s') . " - 发送消息失败: {$e->getMessage()}\n";
        } catch (Exception $e) {
            echo date('Y-m-d H:i:s') . " - 消息序列化失败: {$e->getMessage()}\n";
        }
    }

    /**
     * 监听服务端消息
     */
    private function listenMessages(): void
    {
        if (!$this->isConnected || !$this->client) {
            return;
        }

        try {
            // 设置非阻塞读取（避免卡死）
            $socket = $this->client->getSocket();
            stream_set_blocking($socket, false);

            $message = $this->client->receive();
            if ($message !== '') {
                echo date('Y-m-d H:i:s') . " - 收到消息: {$message}\n";
                $this->handleReceivedMessage($message);
            }
        } catch (ConnectionException $e) {
            $this->isConnected = false;
            echo date('Y-m-d H:i:s') . " - 连接断开/读取消息失败: {$e->getMessage()}\n";
        } catch (Exception $e) {
            // 忽略非阻塞读取的空消息异常
            if (strpos($e->getMessage(), 'no data received') === false) {
                echo date('Y-m-d H:i:s') . " - 监听消息异常: {$e->getMessage()}\n";
            }
        }
    }

    /**
     * 处理接收到的消息
     * @param string $message 原始消息字符串
     */
    private function handleReceivedMessage(string $message): void
    {
        try {
            $msg = json_decode($message, true);
            if (json_last_error() !== JSON_ERROR_NONE) {
                echo date('Y-m-d H:i:s') . " - 解析消息失败: 格式错误\n";
                return;
            }

            $code = $msg['code'] ?? 0;
            $trace = $msg['trace'] ?? '';
            $data = $msg['data'] ?? [];

            switch ($code) {
                case 10000:
                    echo date('Y-m-d H:i:s') . " - 处理成交明细 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
                    break;
                case 10003:
                    echo date('Y-m-d H:i:s') . " - 处理盘口数据 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
                    break;
                case 10006:
                    echo date('Y-m-d H:i:s') . " - 处理K线数据 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n";
                    break;
                case 10010:
                    echo date('Y-m-d H:i:s') . " - 收到心跳响应 [trace={$trace}]\n";
                    break;
                default:
                    echo date('Y-m-d H:i:s') . " - 未知协议号 [code={$code}]: {$message}\n";
            }
        } catch (Exception $e) {
            echo date('Y-m-d H:i:s') . " - 处理消息异常: {$e->getMessage()}\n";
        }
    }

    /**
     * 启动心跳任务
     * @param int $lastHeartbeatTime 上次心跳时间戳引用
     */
    private function checkHeartbeat(&$lastHeartbeatTime): void
    {
        $currentTime = time();
        if ($currentTime - $lastHeartbeatTime >= $this->heartbeatInterval) {
            $this->sendHeartbeat();
            $lastHeartbeatTime = $currentTime;
        }
    }

    /**
     * 启动客户端
     */
    public function start(): void
    {
        echo date('Y-m-d H:i:s') . " - 启动WebSocket客户端...\n";
        
        // 注册退出信号处理
        pcntl_async_signals(true);
        pcntl_signal(SIGINT, function () {
            echo "\n" . date('Y-m-d H:i:s') . " - 收到退出信号，停止客户端...\n";
            $this->running = false;
            if ($this->client) {
                $this->client->close();
            }
            exit(0);
        });

        $lastHeartbeatTime = 0;
        $reconnectDelay = 0;

        // 主循环
        while ($this->running) {
            // 处理重连逻辑
            if (!$this->isConnected) {
                if ($reconnectDelay <= 0) {
                    $this->connect();
                    $reconnectDelay = $this->reconnectInterval;
                    $lastHeartbeatTime = time();
                } else {
                    $reconnectDelay--;
                    echo date('Y-m-d H:i:s') . " - 等待重连... ({$reconnectDelay}秒)\n";
                }
            } else {
                // 监听消息
                $this->listenMessages();
                // 检查心跳
                $this->checkHeartbeat($lastHeartbeatTime);
            }

            // 主循环延迟（避免CPU占用过高）
            usleep(100000); // 100毫秒
        }

        // 清理资源
        if ($this->client) {
            $this->client->close();
        }
        echo date('Y-m-d H:i:s') . " - 客户端已停止\n";
    }

    /**
     * 析构函数
     */
    public function __destruct()
    {
        if ($this->client) {
            try {
                $this->client->close();
            } catch (Exception $e) {
                // 忽略
            }
        }
    }
}

// 主程序入口
if (php_sapi_name() !== 'cli') {
    die("该脚本仅支持CLI模式运行！");
}

// 配置（替换为你的实际API Key）
$apiKey = 'yourApikey';
$business = 'crypto';

// 创建并启动客户端
$client = new CryptoWebsocketClient($apiKey, $business);
$client->start();

/**
 * 加密货币行情WebSocket客户端（JS通用版）
 * 支持浏览器/Node.js环境（Node.js需确保版本≥10.0.0）
 */
class CryptoWebsocketClient {
  /**
   * 构造函数
   * @param {string} apiKey - 认证API Key
   * @param {string} business - 业务类型：stock/crypto/common
   */
  constructor(apiKey, business = 'crypto') {
    // 配置项
    this.apiKey = apiKey;
    this.business = business;
    this.wsUrl = `wss://data.infoway.io/ws?business=${encodeURIComponent(business)}&apikey=${encodeURIComponent(apiKey)}`;
    this.reconnectInterval = 10000; // 重连间隔（10秒，单位ms）
    this.heartbeatInterval = 30000; // 心跳间隔（30秒，单位ms）

    // 运行状态
    this.ws = null; // WebSocket实例
    this.isConnected = false; // 连接状态
    this.heartbeatTimer = null; // 心跳定时器
    this.reconnectTimer = null; // 重连定时器
    this.subscribeSent = false; // 订阅请求是否已发送

    // 绑定方法上下文（避免回调中this丢失）
    this.onOpen = this.onOpen.bind(this);
    this.onMessage = this.onMessage.bind(this);
    this.onClose = this.onClose.bind(this);
    this.onError = this.onError.bind(this);
    this.sendHeartbeat = this.sendHeartbeat.bind(this);
  }

  /**
   * 生成唯一Trace ID（模拟UUID）
   * @returns {string}
   */
  generateTraceId() {
    return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (c) => {
      const r = Math.random() * 16 | 0;
      const v = c === 'x' ? r : (r & 0x3 | 0x8);
      return v.toString(16);
    });
  }

  /**
   * 建立WebSocket连接
   */
  connect() {
    // 清理旧连接
    if (this.ws) {
      try {
        this.ws.close(1000, 'Reconnecting');
      } catch (e) {
        console.error('关闭旧连接失败:', e);
      }
      this.ws = null;
    }

    try {
      // 创建新连接
      this.ws = new WebSocket(this.wsUrl);
      // 注册事件回调
      this.ws.onopen = this.onOpen;
      this.ws.onmessage = this.onMessage;
      this.ws.onclose = this.onClose;
      this.ws.onerror = this.onError;
      console.log(`[${new Date().toLocaleString()}] 正在连接WebSocket: ${this.wsUrl}`);
    } catch (e) {
      this.isConnected = false;
      console.error(`[${new Date().toLocaleString()}] 连接失败:`, e);
      this.scheduleReconnect();
    }
  }

  /**
   * 连接成功回调
   */
  onOpen() {
    this.isConnected = true;
    this.subscribeSent = false;
    console.log(`[${new Date().toLocaleString()}] WebSocket连接成功`);

    // 清理重连定时器
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }

    // 发送订阅请求
    this.sendAllSubscribeRequests();

    // 启动心跳
    this.startHeartbeat();
  }

  /**
   * 发送所有订阅请求
   */
  sendAllSubscribeRequests() {
    if (this.subscribeSent || !this.isConnected) return;
    this.subscribeSent = true;

    // 1. 订阅实时成交明细（协议号10000）
    this.sendTradeSubscribe();
    
    // 间隔5秒发送盘口订阅
    setTimeout(() => {
      if (this.isConnected) {
        this.sendDepthSubscribe();
      }
    }, 5000);

    // 间隔10秒发送K线订阅（5+5）
    setTimeout(() => {
      if (this.isConnected) {
        this.sendKlineSubscribe();
      }
    }, 10000);
  }

  /**
   * 发送成交明细订阅请求
   */
  sendTradeSubscribe() {
    const msg = {
      code: 10000,
      trace: this.generateTraceId(),
      data: { codes: 'BTCUSDT' }
    };
    this.sendJsonMessage(msg);
  }

  /**
   * 发送盘口数据订阅请求
   */
  sendDepthSubscribe() {
    const msg = {
      code: 10003,
      trace: this.generateTraceId(),
      data: { codes: 'BTCUSDT' }
    };
    this.sendJsonMessage(msg);
  }

  /**
   * 发送K线数据订阅请求
   */
  sendKlineSubscribe() {
    const msg = {
      code: 10006,
      trace: this.generateTraceId(),
      data: {
        arr: [{ type: 1, codes: 'BTCUSDT' }] // 1分钟K线
      }
    };
    this.sendJsonMessage(msg);
  }

  /**
   * 发送心跳包
   */
  sendHeartbeat() {
    const msg = {
      code: 10010,
      trace: this.generateTraceId()
    };
    this.sendJsonMessage(msg);
  }

  /**
   * 发送JSON格式消息
   * @param {object} msg - 消息对象
   */
  sendJsonMessage(msg) {
    if (!this.isConnected || !this.ws || this.ws.readyState !== WebSocket.OPEN) {
      console.warn(`[${new Date().toLocaleString()}] 连接未就绪，跳过消息发送`);
      return;
    }

    try {
      const jsonStr = JSON.stringify(msg);
      this.ws.send(jsonStr);
      console.log(`[${new Date().toLocaleString()}] 发送消息:`, jsonStr);
    } catch (e) {
      console.error(`[${new Date().toLocaleString()}] 发送消息失败:`, e);
      this.isConnected = false;
      this.scheduleReconnect();
    }
  }

  /**
   * 接收消息回调
   * @param {MessageEvent} event - 消息事件
   */
  onMessage(event) {
    const message = event.data;
    console.log(`[${new Date().toLocaleString()}] 收到消息:`, message);
    this.handleReceivedMessage(message);
  }

  /**
   * 处理接收到的消息
   * @param {string} message - 原始消息字符串
   */
  handleReceivedMessage(message) {
    try {
      const msg = JSON.parse(message);
      const code = msg.code || 0;
      const trace = msg.trace || '';
      const data = msg.data || {};

      switch (code) {
        case 10000:
          console.log(`[${new Date().toLocaleString()}] 处理成交明细 [trace=${trace}]:`, JSON.stringify(data));
          break;
        case 10003:
          console.log(`[${new Date().toLocaleString()}] 处理盘口数据 [trace=${trace}]:`, JSON.stringify(data));
          break;
        case 10006:
          console.log(`[${new Date().toLocaleString()}] 处理K线数据 [trace=${trace}]:`, JSON.stringify(data));
          break;
        case 10010:
          console.log(`[${new Date().toLocaleString()}] 收到心跳响应 [trace=${trace}]`);
          break;
        default:
          console.warn(`[${new Date().toLocaleString()}] 未知协议号 [code=${code}]:`, message);
      }
    } catch (e) {
      console.error(`[${new Date().toLocaleString()}] 处理消息失败:`, e);
    }
  }

  /**
   * 连接关闭回调
   * @param {CloseEvent} event - 关闭事件
   */
  onClose(event) {
    this.isConnected = false;
    this.subscribeSent = false;
    console.log(`[${new Date().toLocaleString()}] 连接关闭: 代码=${event.code}, 原因=${event.reason}`);
    
    // 清理心跳定时器
    this.stopHeartbeat();
    
    // 触发重连
    this.scheduleReconnect();
  }

  /**
   * 错误回调
   * @param {Event} error - 错误事件
   */
  onError(error) {
    console.error(`[${new Date().toLocaleString()}] WebSocket错误:`, error);
    this.isConnected = false;
    this.scheduleReconnect();
  }

  /**
   * 启动心跳任务
   */
  startHeartbeat() {
    // 清理旧定时器
    this.stopHeartbeat();
    
    // 立即发送一次心跳，然后定时发送
    this.sendHeartbeat();
    this.heartbeatTimer = setInterval(() => {
      if (this.isConnected) {
        this.sendHeartbeat();
      }
    }, this.heartbeatInterval);
  }

  /**
   * 停止心跳任务
   */
  stopHeartbeat() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
  }

  /**
   * 调度重连
   */
  scheduleReconnect() {
    if (this.reconnectTimer) return;
    
    console.log(`[${new Date().toLocaleString()}] ${this.reconnectInterval/1000}秒后尝试重连...`);
    this.reconnectTimer = setTimeout(() => {
      this.reconnectTimer = null;
      this.connect();
    }, this.reconnectInterval);
  }

  /**
   * 启动客户端
   */
  start() {
    console.log(`[${new Date().toLocaleString()}] 启动WebSocket客户端...`);
    this.connect();

    // Node.js环境下监听退出信号
    if (typeof process !== 'undefined' && process.on) {
      process.on('SIGINT', () => {
        console.log(`\n[${new Date().toLocaleString()}] 收到退出信号，关闭客户端...`);
        this.stop();
        process.exit(0);
      });
    }
  }

  /**
   * 停止客户端
   */
  stop() {
    this.isConnected = false;
    this.subscribeSent = false;
    
    // 清理定时器
    this.stopHeartbeat();
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }

    // 关闭连接
    if (this.ws) {
      this.ws.close(1000, 'Client stopped');
      this.ws = null;
    }

    console.log(`[${new Date().toLocaleString()}] 客户端已停止`);
  }
}

// ==================== 运行示例 ====================
// 浏览器环境：直接在控制台执行以下代码
// Node.js环境：直接运行该脚本

// 替换为你的实际API Key
const API_KEY = 'yourApikey';
const BUSINESS = 'crypto';

// 创建并启动客户端
const client = new CryptoWebsocketClient(API_KEY, BUSINESS);
client.start();

// 如需停止客户端，执行：client.stop();

上一页Websocket订阅地址说明下一页Websocket 订阅方法

最后更新于1天前

import asyncio import json import uuid import logging from typing import Optional import websockets from websockets.exceptions import ConnectionClosed, WebSocketException # 配置日志 logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) logger = logging.getLogger("websocket-client") class CryptoWebsocketClient: """加密货币行情WebSocket客户端（对等Java版本逻辑）""" def __init__(self, api_key: str, business: str = "crypto"): # WebSocket连接配置 self.ws_url = f"wss://data.infoway.io/ws?business={business}&apikey={api_key}" # 核心状态 self.ws: Optional[websockets.WebSocketClientProtocol] = None self.is_connected = False self.reconnect_interval = 10 # 重连间隔（秒） self.heartbeat_interval = 30 # 心跳间隔（秒） # 任务对象（用于管理异步任务） self.reconnect_task: Optional[asyncio.Task] = None self.heartbeat_task: Optional[asyncio.Task] = None async def connect(self) -> None: """建立WebSocket连接""" try: # 关闭旧连接（防止连接泄露） if self.ws and not self.ws.closed: await self.ws.close() # 建立新连接 self.ws = await websockets.connect(self.ws_url) self.is_connected = True logger.info(f"WebSocket连接成功，地址: {self.ws_url}") # 连接成功后启动订阅和心跳 await self._send_all_subscribe_requests() self._start_heartbeat_task() except WebSocketException as e: self.is_connected = False logger.error(f"WebSocket连接失败: {str(e)}") raise async def _send_all_subscribe_requests(self) -> None: """发送所有订阅请求（成交明细、盘口、K线）""" if not self.ws or self.ws.closed: logger.error("连接未建立，无法发送订阅请求") return try: # 1. 订阅实时成交明细（协议号10000） await self._send_trade_subscribe() await asyncio.sleep(5) # 间隔5秒 # 2. 订阅实时盘口数据（协议号10003） await self._send_depth_subscribe() await asyncio.sleep(5) # 间隔5秒 # 3. 订阅1分钟K线数据（协议号10006） await self._send_kline_subscribe() except Exception as e: logger.error(f"发送订阅请求失败: {str(e)}") raise def _generate_trace_id(self) -> str: """生成唯一trace ID（对等Java的UUID）""" return str(uuid.uuid4()) async def _send_trade_subscribe(self) -> None: """发送实时成交明细订阅请求""" subscribe_msg = { "code": 10000, "trace": self._generate_trace_id(), "data": {"codes": "BTCUSDT"} } await self.ws.send(json.dumps(subscribe_msg)) logger.info(f"发送成交明细订阅请求: {subscribe_msg}") async def _send_depth_subscribe(self) -> None: """发送实时盘口数据订阅请求""" subscribe_msg = { "code": 10003, "trace": self._generate_trace_id(), "data": {"codes": "BTCUSDT"} } await self.ws.send(json.dumps(subscribe_msg)) logger.info(f"发送盘口数据订阅请求: {subscribe_msg}") async def _send_kline_subscribe(self) -> None: """发送1分钟K线数据订阅请求""" subscribe_msg = { "code": 10006, "trace": self._generate_trace_id(), "data": { "arr": [ {"type": 1, "codes": "BTCUSDT"} # type=1 表示1分钟K线 ] } } await self.ws.send(json.dumps(subscribe_msg)) logger.info(f"发送K线数据订阅请求: {subscribe_msg}") async def _send_heartbeat(self) -> None: """发送心跳包（协议号10010）""" if not self.ws or self.ws.closed: return heartbeat_msg = { "code": 10010, "trace": self._generate_trace_id() } try: await self.ws.send(json.dumps(heartbeat_msg)) logger.debug(f"发送心跳包: {heartbeat_msg}") except Exception as e: logger.error(f"发送心跳包失败: {str(e)}") raise def _start_heartbeat_task(self) -> None: """启动心跳任务（后台定时发送）""" if self.heartbeat_task and not self.heartbeat_task.done(): self.heartbeat_task.cancel() async def heartbeat_loop(): while self.is_connected and self.ws and not self.ws.closed: try: await self._send_heartbeat() await asyncio.sleep(self.heartbeat_interval) except Exception as e: logger.error(f"心跳任务异常: {str(e)}") break self.heartbeat_task = asyncio.create_task(heartbeat_loop()) async def _message_listener(self) -> None: """监听服务端推送的消息""" while self.is_connected and self.ws and not self.ws.closed: try: # 阻塞等待接收消息 message = await self.ws.recv() logger.info(f"收到服务端消息: {message}") # 解析消息并处理（对等Java的OnMessage逻辑） self._handle_received_message(message) except ConnectionClosed: logger.warning("WebSocket连接已关闭，停止消息监听") self.is_connected = False break except Exception as e: logger.error(f"接收/处理消息异常: {str(e)}") def _handle_received_message(self, message: str) -> None: """处理接收到的消息（按协议号分类）""" try: msg_data = json.loads(message) code = msg_data.get("code") trace = msg_data.get("trace") data = msg_data.get("data", {}) if code == 10000: logger.info(f"处理成交明细数据 [trace={trace}]: {data}") elif code == 10003: logger.info(f"处理盘口数据 [trace={trace}]: {data}") elif code == 10006: logger.info(f"处理K线数据 [trace={trace}]: {data}") elif code == 10010: logger.debug(f"收到心跳响应 [trace={trace}]") else: logger.warning(f"未知协议号消息 [code={code}]: {message}") except json.JSONDecodeError: logger.error(f"消息格式错误，无法解析JSON: {message}") except Exception as e: logger.error(f"处理消息异常: {str(e)}") async def _reconnect_loop(self) -> None: """自动重连循环（对等Java的startReconnection）""" while True: if not self.is_connected: logger.info(f"尝试重连WebSocket（间隔{self.reconnect_interval}秒）...") try: await self.connect() except Exception: # 重连失败，等待后重试 await asyncio.sleep(self.reconnect_interval) else: # 连接正常，短暂等待后再次检查 await asyncio.sleep(1) async def start(self) -> None: """启动客户端（主入口）""" # 启动自动重连任务 self.reconnect_task = asyncio.create_task(self._reconnect_loop()) try: # 首次连接 await self.connect() # 启动消息监听 await self._message_listener() finally: # 清理资源 self.is_connected = False if self.heartbeat_task: self.heartbeat_task.cancel() if self.reconnect_task: self.reconnect_task.cancel() if self.ws and not self.ws.closed: await self.ws.close() logger.info("WebSocket客户端已停止") async def main(): """主函数""" # 替换为你的实际API Key API_KEY = "yourApikey" client = CryptoWebsocketClient(api_key=API_KEY) await client.start() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("用户中断，退出客户端") except Exception as e: logger.error(f"客户端运行异常: {str(e)}")

package main import ( "encoding/json" "fmt" "log" "math/rand" "net/url" "os" "os/signal" "syscall" "time" "github.com/google/uuid" "github.com/gorilla/websocket" ) // CryptoWebsocketClient 加密货币行情WebSocket客户端 type CryptoWebsocketClient struct { apiKey string // 认证API Key business string // 业务类型：stock/crypto/common wsURL string // WebSocket连接地址 conn *websocket.Conn // WebSocket连接实例 isConnected bool // 连接状态 reconnectInterval time.Duration // 重连间隔 heartbeatInterval time.Duration // 心跳间隔 done chan struct{} // 退出信号通道 } // NewCryptoWebsocketClient 创建客户端实例 func NewCryptoWebsocketClient(apiKey, business string) *CryptoWebsocketClient { // 构建WS URL params := url.Values{} params.Set("business", business) params.Set("apikey", apiKey) wsURL := fmt.Sprintf("wss://data.infoway.io/ws?%s", params.Encode()) return &CryptoWebsocketClient{ apiKey: apiKey, business: business, wsURL: wsURL, reconnectInterval: 10 * time.Second, // 10秒重连间隔 heartbeatInterval: 30 * time.Second, // 30秒心跳间隔 done: make(chan struct{}), } } // generateTraceID 生成唯一的trace ID（对齐Java/Python的UUID） func (c *CryptoWebsocketClient) generateTraceID() string { return uuid.New().String() } // connect 建立WebSocket连接 func (c *CryptoWebsocketClient) connect() error { // 关闭旧连接 if c.conn != nil { c.conn.Close() c.conn = nil } // 建立新连接 conn, _, err := websocket.DefaultDialer.Dial(c.wsURL, nil) if err != nil { c.isConnected = false return fmt.Errorf("连接失败: %w", err) } c.conn = conn c.isConnected = true log.Printf("WebSocket连接成功: %s", c.wsURL) // 连接成功后发送订阅请求 go c.sendAllSubscribeRequests() // 启动心跳任务 go c.startHeartbeat() // 启动消息监听 go c.listenMessages() return nil } // sendAllSubscribeRequests 发送所有订阅请求（成交明细、盘口、K线） func (c *CryptoWebsocketClient) sendAllSubscribeRequests() { if !c.isConnected || c.conn == nil { log.Println("连接未建立，无法发送订阅请求") return } // 1. 订阅实时成交明细（协议号10000） if err := c.sendTradeSubscribe(); err != nil { log.Printf("发送成交明细订阅失败: %v", err) return } time.Sleep(5 * time.Second) // 间隔5秒 // 2. 订阅实时盘口数据（协议号10003） if err := c.sendDepthSubscribe(); err != nil { log.Printf("发送盘口订阅失败: %v", err) return } time.Sleep(5 * time.Second) // 间隔5秒 // 3. 订阅1分钟K线数据（协议号10006） if err := c.sendKlineSubscribe(); err != nil { log.Printf("发送K线订阅失败: %v", err) return } } // sendTradeSubscribe 发送实时成交明细订阅请求 func (c *CryptoWebsocketClient) sendTradeSubscribe() error { msg := map[string]interface{}{ "code": 10000, "trace": c.generateTraceID(), "data": map[string]string{ "codes": "BTCUSDT", }, } return c.sendJSONMessage(msg) } // sendDepthSubscribe 发送实时盘口数据订阅请求 func (c *CryptoWebsocketClient) sendDepthSubscribe() error { msg := map[string]interface{}{ "code": 10003, "trace": c.generateTraceID(), "data": map[string]string{ "codes": "BTCUSDT", }, } return c.sendJSONMessage(msg) } // sendKlineSubscribe 发送1分钟K线数据订阅请求 func (c *CryptoWebsocketClient) sendKlineSubscribe() error { msg := map[string]interface{}{ "code": 10006, "trace": c.generateTraceID(), "data": map[string]interface{}{ "arr": []map[string]interface{}{ { "type": 1, // 1分钟K线 "codes": "BTCUSDT", }, }, }, } return c.sendJSONMessage(msg) } // sendHeartbeat 发送心跳包（协议号10010） func (c *CryptoWebsocketClient) sendHeartbeat() error { msg := map[string]interface{}{ "code": 10010, "trace": c.generateTraceID(), } return c.sendJSONMessage(msg) } // sendJSONMessage 发送JSON格式消息 func (c *CryptoWebsocketClient) sendJSONMessage(msg interface{}) error { if !c.isConnected || c.conn == nil { return fmt.Errorf("连接已断开") } // 序列化为JSON data, err := json.Marshal(msg) if err != nil { return fmt.Errorf("消息序列化失败: %w", err) } // 发送消息 if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil { c.isConnected = false return fmt.Errorf("发送消息失败: %w", err) } log.Printf("发送消息: %s", string(data)) return nil } // startHeartbeat 启动心跳保活任务 func (c *CryptoWebsocketClient) startHeartbeat() { ticker := time.NewTicker(c.heartbeatInterval) defer ticker.Stop() for { select { case <-ticker.C: // 定时发送心跳 if err := c.sendHeartbeat(); err != nil { log.Printf("发送心跳失败: %v", err) return } log.Printf("发送心跳包成功") case <-c.done: // 收到退出信号，停止心跳 return } } } // listenMessages 监听服务端推送的消息 func (c *CryptoWebsocketClient) listenMessages() { for { if !c.isConnected || c.conn == nil { return } // 读取消息 _, msgData, err := c.conn.ReadMessage() if err != nil { c.isConnected = false log.Printf("读取消息失败/连接断开: %v", err) return } // 打印原始消息 log.Printf("收到服务端消息: %s", string(msgData)) // 解析并处理消息 c.handleReceivedMessage(msgData) } } // handleReceivedMessage 处理接收到的消息（按协议号分类） func (c *CryptoWebsocketClient) handleReceivedMessage(msgData []byte) { var msg map[string]interface{} if err := json.Unmarshal(msgData, &msg); err != nil { log.Printf("解析消息失败: %v", err) return } // 提取核心字段 code := int(msg["code"].(float64)) // JSON数字默认解析为float64 trace := msg["trace"].(string) data := msg["data"] // 按协议号处理不同类型数据 switch code { case 10000: log.Printf("处理成交明细数据 [trace=%s]: %+v", trace, data) case 10003: log.Printf("处理盘口数据 [trace=%s]: %+v", trace, data) case 10006: log.Printf("处理K线数据 [trace=%s]: %+v", trace, data) case 10010: log.Printf("收到心跳响应 [trace=%s]", trace) default: log.Printf("未知协议号消息 [code=%d]: %s", code, string(msgData)) } } // startReconnectLoop 启动自动重连循环 func (c *CryptoWebsocketClient) startReconnectLoop() { for { select { case <-c.done: // 收到退出信号，停止重连 return default: if !c.isConnected { log.Printf("尝试重连WebSocket（间隔%v）...", c.reconnectInterval) // 重连前随机延迟（避免高频重试） time.Sleep(time.Duration(rand.Intn(3)) * time.Second) if err := c.connect(); err != nil { log.Printf("重连失败: %v", err) time.Sleep(c.reconnectInterval) continue } } // 连接正常时，短暂休眠后再次检查 time.Sleep(1 * time.Second) } } } // Start 启动客户端 func (c *CryptoWebsocketClient) Start() { // 启动自动重连循环 go c.startReconnectLoop() // 首次连接 if err := c.connect(); err != nil { log.Printf("首次连接失败: %v", err) } // 监听系统退出信号（Ctrl+C） sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) <-sigChan // 收到退出信号，清理资源 log.Println("收到退出信号，关闭客户端...") close(c.done) if c.conn != nil { c.conn.Close() } c.isConnected = false } func main() { // 替换为你的实际API Key apiKey := "yourApikey" // 业务类型：crypto/stock/common business := "crypto" // 创建客户端 client := NewCryptoWebsocketClient(apiKey, business) // 启动客户端 client.Start() }

<?php require_once __DIR__ . '/vendor/autoload.php'; use WebSocket\Client; use WebSocket\ConnectionException; /** * 加密货币行情WebSocket客户端 */ class CryptoWebsocketClient { // 配置项 private $apiKey; // 认证API Key private $business; // 业务类型：stock/crypto/common private $wsUrl; // WebSocket连接地址 private $reconnectInterval; // 重连间隔（秒） private $heartbeatInterval; // 心跳间隔（秒） // 运行状态 private $client; // WebSocket客户端实例 private $isConnected = false;// 连接状态 private $running = true; // 客户端运行状态 /** * 构造函数 * @param string $apiKey API密钥 * @param string $business 业务类型 */ public function __construct(string $apiKey, string $business = 'crypto') { $this->apiKey = $apiKey; $this->business = $business; // 构建WS URL $this->wsUrl = sprintf( 'wss://data.infoway.io/ws?business=%s&apikey=%s', urlencode($business), urlencode($apiKey) ); $this->reconnectInterval = 10; // 10秒重连间隔 $this->heartbeatInterval = 30; // 30秒心跳间隔 } /** * 生成唯一Trace ID * @return string */ private function generateTraceId(): string { return sprintf( '%04x%04x-%04x-%04x-%04x-%04x%04x%04x', mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0x0fff) | 0x4000, mt_rand(0, 0x3fff) | 0x8000, mt_rand(0, 0xffff), mt_rand(0, 0xffff), mt_rand(0, 0xffff) ); } /** * 建立WebSocket连接 * @return bool */ private function connect(): bool { try { // 关闭旧连接 if ($this->client) { try { $this->client->close(); } catch (Exception $e) { // 忽略关闭异常 } $this->client = null; } // 创建新连接 $this->client = new Client($this->wsUrl, [ 'timeout' => 30, 'fragment_size' => 8192, ]); $this->isConnected = true; echo date('Y-m-d H:i:s') . " - WebSocket连接成功: {$this->wsUrl}\n"; // 发送所有订阅请求 $this->sendAllSubscribeRequests(); return true; } catch (ConnectionException $e) { $this->isConnected = false; echo date('Y-m-d H:i:s') . " - WebSocket连接失败: {$e->getMessage()}\n"; return false; } catch (Exception $e) { $this->isConnected = false; echo date('Y-m-d H:i:s') . " - 连接异常: {$e->getMessage()}\n"; return false; } } /** * 发送所有订阅请求 */ private function sendAllSubscribeRequests(): void { if (!$this->isConnected || !$this->client) { echo date('Y-m-d H:i:s') . " - 连接未建立，无法发送订阅请求\n"; return; } try { // 1. 订阅实时成交明细（协议号10000） $this->sendTradeSubscribe(); sleep(5); // 间隔5秒 // 2. 订阅实时盘口数据（协议号10003） $this->sendDepthSubscribe(); sleep(5); // 间隔5秒 // 3. 订阅1分钟K线数据（协议号10006） $this->sendKlineSubscribe(); } catch (Exception $e) { echo date('Y-m-d H:i:s') . " - 发送订阅请求失败: {$e->getMessage()}\n"; $this->isConnected = false; } } /** * 发送成交明细订阅请求 */ private function sendTradeSubscribe(): void { $msg = [ 'code' => 10000, 'trace' => $this->generateTraceId(), 'data' => ['codes' => 'BTCUSDT'] ]; $this->sendJsonMessage($msg); } /** * 发送盘口数据订阅请求 */ private function sendDepthSubscribe(): void { $msg = [ 'code' => 10003, 'trace' => $this->generateTraceId(), 'data' => ['codes' => 'BTCUSDT'] ]; $this->sendJsonMessage($msg); } /** * 发送K线数据订阅请求 */ private function sendKlineSubscribe(): void { $msg = [ 'code' => 10006, 'trace' => $this->generateTraceId(), 'data' => [ 'arr' => [ ['type' => 1, 'codes' => 'BTCUSDT'] // 1分钟K线 ] ] ]; $this->sendJsonMessage($msg); } /** * 发送心跳包 */ private function sendHeartbeat(): void { $msg = [ 'code' => 10010, 'trace' => $this->generateTraceId() ]; $this->sendJsonMessage($msg); } /** * 发送JSON格式消息 * @param array $msg 消息数组 */ private function sendJsonMessage(array $msg): void { if (!$this->isConnected || !$this->client) { return; } try { $jsonStr = json_encode($msg, JSON_UNESCAPED_UNICODE); $this->client->send($jsonStr); echo date('Y-m-d H:i:s') . " - 发送消息: {$jsonStr}\n"; } catch (ConnectionException $e) { $this->isConnected = false; echo date('Y-m-d H:i:s') . " - 发送消息失败: {$e->getMessage()}\n"; } catch (Exception $e) { echo date('Y-m-d H:i:s') . " - 消息序列化失败: {$e->getMessage()}\n"; } } /** * 监听服务端消息 */ private function listenMessages(): void { if (!$this->isConnected || !$this->client) { return; } try { // 设置非阻塞读取（避免卡死） $socket = $this->client->getSocket(); stream_set_blocking($socket, false); $message = $this->client->receive(); if ($message !== '') { echo date('Y-m-d H:i:s') . " - 收到消息: {$message}\n"; $this->handleReceivedMessage($message); } } catch (ConnectionException $e) { $this->isConnected = false; echo date('Y-m-d H:i:s') . " - 连接断开/读取消息失败: {$e->getMessage()}\n"; } catch (Exception $e) { // 忽略非阻塞读取的空消息异常 if (strpos($e->getMessage(), 'no data received') === false) { echo date('Y-m-d H:i:s') . " - 监听消息异常: {$e->getMessage()}\n"; } } } /** * 处理接收到的消息 * @param string $message 原始消息字符串 */ private function handleReceivedMessage(string $message): void { try { $msg = json_decode($message, true); if (json_last_error() !== JSON_ERROR_NONE) { echo date('Y-m-d H:i:s') . " - 解析消息失败: 格式错误\n"; return; } $code = $msg['code'] ?? 0; $trace = $msg['trace'] ?? ''; $data = $msg['data'] ?? []; switch ($code) { case 10000: echo date('Y-m-d H:i:s') . " - 处理成交明细 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n"; break; case 10003: echo date('Y-m-d H:i:s') . " - 处理盘口数据 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n"; break; case 10006: echo date('Y-m-d H:i:s') . " - 处理K线数据 [trace={$trace}]: " . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n"; break; case 10010: echo date('Y-m-d H:i:s') . " - 收到心跳响应 [trace={$trace}]\n"; break; default: echo date('Y-m-d H:i:s') . " - 未知协议号 [code={$code}]: {$message}\n"; } } catch (Exception $e) { echo date('Y-m-d H:i:s') . " - 处理消息异常: {$e->getMessage()}\n"; } } /** * 启动心跳任务 * @param int $lastHeartbeatTime 上次心跳时间戳引用 */ private function checkHeartbeat(&$lastHeartbeatTime): void { $currentTime = time(); if ($currentTime - $lastHeartbeatTime >= $this->heartbeatInterval) { $this->sendHeartbeat(); $lastHeartbeatTime = $currentTime; } } /** * 启动客户端 */ public function start(): void { echo date('Y-m-d H:i:s') . " - 启动WebSocket客户端...\n"; // 注册退出信号处理 pcntl_async_signals(true); pcntl_signal(SIGINT, function () { echo "\n" . date('Y-m-d H:i:s') . " - 收到退出信号，停止客户端...\n"; $this->running = false; if ($this->client) { $this->client->close(); } exit(0); }); $lastHeartbeatTime = 0; $reconnectDelay = 0; // 主循环 while ($this->running) { // 处理重连逻辑 if (!$this->isConnected) { if ($reconnectDelay <= 0) { $this->connect(); $reconnectDelay = $this->reconnectInterval; $lastHeartbeatTime = time(); } else { $reconnectDelay--; echo date('Y-m-d H:i:s') . " - 等待重连... ({$reconnectDelay}秒)\n"; } } else { // 监听消息 $this->listenMessages(); // 检查心跳 $this->checkHeartbeat($lastHeartbeatTime); } // 主循环延迟（避免CPU占用过高） usleep(100000); // 100毫秒 } // 清理资源 if ($this->client) { $this->client->close(); } echo date('Y-m-d H:i:s') . " - 客户端已停止\n"; } /** * 析构函数 */ public function __destruct() { if ($this->client) { try { $this->client->close(); } catch (Exception $e) { // 忽略 } } } } // 主程序入口 if (php_sapi_name() !== 'cli') { die("该脚本仅支持CLI模式运行！"); } // 配置（替换为你的实际API Key） $apiKey = 'yourApikey'; $business = 'crypto'; // 创建并启动客户端 $client = new CryptoWebsocketClient($apiKey, $business); $client->start();

/** * 加密货币行情WebSocket客户端（JS通用版） * 支持浏览器/Node.js环境（Node.js需确保版本≥10.0.0） */ class CryptoWebsocketClient { /** * 构造函数 * @param {string} apiKey - 认证API Key * @param {string} business - 业务类型：stock/crypto/common */ constructor(apiKey, business = 'crypto') { // 配置项 this.apiKey = apiKey; this.business = business; this.wsUrl = `wss://data.infoway.io/ws?business=${encodeURIComponent(business)}&apikey=${encodeURIComponent(apiKey)}`; this.reconnectInterval = 10000; // 重连间隔（10秒，单位ms） this.heartbeatInterval = 30000; // 心跳间隔（30秒，单位ms） // 运行状态 this.ws = null; // WebSocket实例 this.isConnected = false; // 连接状态 this.heartbeatTimer = null; // 心跳定时器 this.reconnectTimer = null; // 重连定时器 this.subscribeSent = false; // 订阅请求是否已发送 // 绑定方法上下文（避免回调中this丢失） this.onOpen = this.onOpen.bind(this); this.onMessage = this.onMessage.bind(this); this.onClose = this.onClose.bind(this); this.onError = this.onError.bind(this); this.sendHeartbeat = this.sendHeartbeat.bind(this); } /** * 生成唯一Trace ID（模拟UUID） * @returns {string} */ generateTraceId() { return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (c) => { const r = Math.random() * 16 | 0; const v = c === 'x' ? r : (r & 0x3 | 0x8); return v.toString(16); }); } /** * 建立WebSocket连接 */ connect() { // 清理旧连接 if (this.ws) { try { this.ws.close(1000, 'Reconnecting'); } catch (e) { console.error('关闭旧连接失败:', e); } this.ws = null; } try { // 创建新连接 this.ws = new WebSocket(this.wsUrl); // 注册事件回调 this.ws.onopen = this.onOpen; this.ws.onmessage = this.onMessage; this.ws.onclose = this.onClose; this.ws.onerror = this.onError; console.log(`[${new Date().toLocaleString()}] 正在连接WebSocket: ${this.wsUrl}`); } catch (e) { this.isConnected = false; console.error(`[${new Date().toLocaleString()}] 连接失败:`, e); this.scheduleReconnect(); } } /** * 连接成功回调 */ onOpen() { this.isConnected = true; this.subscribeSent = false; console.log(`[${new Date().toLocaleString()}] WebSocket连接成功`); // 清理重连定时器 if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } // 发送订阅请求 this.sendAllSubscribeRequests(); // 启动心跳 this.startHeartbeat(); } /** * 发送所有订阅请求 */ sendAllSubscribeRequests() { if (this.subscribeSent || !this.isConnected) return; this.subscribeSent = true; // 1. 订阅实时成交明细（协议号10000） this.sendTradeSubscribe(); // 间隔5秒发送盘口订阅 setTimeout(() => { if (this.isConnected) { this.sendDepthSubscribe(); } }, 5000); // 间隔10秒发送K线订阅（5+5） setTimeout(() => { if (this.isConnected) { this.sendKlineSubscribe(); } }, 10000); } /** * 发送成交明细订阅请求 */ sendTradeSubscribe() { const msg = { code: 10000, trace: this.generateTraceId(), data: { codes: 'BTCUSDT' } }; this.sendJsonMessage(msg); } /** * 发送盘口数据订阅请求 */ sendDepthSubscribe() { const msg = { code: 10003, trace: this.generateTraceId(), data: { codes: 'BTCUSDT' } }; this.sendJsonMessage(msg); } /** * 发送K线数据订阅请求 */ sendKlineSubscribe() { const msg = { code: 10006, trace: this.generateTraceId(), data: { arr: [{ type: 1, codes: 'BTCUSDT' }] // 1分钟K线 } }; this.sendJsonMessage(msg); } /** * 发送心跳包 */ sendHeartbeat() { const msg = { code: 10010, trace: this.generateTraceId() }; this.sendJsonMessage(msg); } /** * 发送JSON格式消息 * @param {object} msg - 消息对象 */ sendJsonMessage(msg) { if (!this.isConnected || !this.ws || this.ws.readyState !== WebSocket.OPEN) { console.warn(`[${new Date().toLocaleString()}] 连接未就绪，跳过消息发送`); return; } try { const jsonStr = JSON.stringify(msg); this.ws.send(jsonStr); console.log(`[${new Date().toLocaleString()}] 发送消息:`, jsonStr); } catch (e) { console.error(`[${new Date().toLocaleString()}] 发送消息失败:`, e); this.isConnected = false; this.scheduleReconnect(); } } /** * 接收消息回调 * @param {MessageEvent} event - 消息事件 */ onMessage(event) { const message = event.data; console.log(`[${new Date().toLocaleString()}] 收到消息:`, message); this.handleReceivedMessage(message); } /** * 处理接收到的消息 * @param {string} message - 原始消息字符串 */ handleReceivedMessage(message) { try { const msg = JSON.parse(message); const code = msg.code || 0; const trace = msg.trace || ''; const data = msg.data || {}; switch (code) { case 10000: console.log(`[${new Date().toLocaleString()}] 处理成交明细 [trace=${trace}]:`, JSON.stringify(data)); break; case 10003: console.log(`[${new Date().toLocaleString()}] 处理盘口数据 [trace=${trace}]:`, JSON.stringify(data)); break; case 10006: console.log(`[${new Date().toLocaleString()}] 处理K线数据 [trace=${trace}]:`, JSON.stringify(data)); break; case 10010: console.log(`[${new Date().toLocaleString()}] 收到心跳响应 [trace=${trace}]`); break; default: console.warn(`[${new Date().toLocaleString()}] 未知协议号 [code=${code}]:`, message); } } catch (e) { console.error(`[${new Date().toLocaleString()}] 处理消息失败:`, e); } } /** * 连接关闭回调 * @param {CloseEvent} event - 关闭事件 */ onClose(event) { this.isConnected = false; this.subscribeSent = false; console.log(`[${new Date().toLocaleString()}] 连接关闭: 代码=${event.code}, 原因=${event.reason}`); // 清理心跳定时器 this.stopHeartbeat(); // 触发重连 this.scheduleReconnect(); } /** * 错误回调 * @param {Event} error - 错误事件 */ onError(error) { console.error(`[${new Date().toLocaleString()}] WebSocket错误:`, error); this.isConnected = false; this.scheduleReconnect(); } /** * 启动心跳任务 */ startHeartbeat() { // 清理旧定时器 this.stopHeartbeat(); // 立即发送一次心跳，然后定时发送 this.sendHeartbeat(); this.heartbeatTimer = setInterval(() => { if (this.isConnected) { this.sendHeartbeat(); } }, this.heartbeatInterval); } /** * 停止心跳任务 */ stopHeartbeat() { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; } } /** * 调度重连 */ scheduleReconnect() { if (this.reconnectTimer) return; console.log(`[${new Date().toLocaleString()}] ${this.reconnectInterval/1000}秒后尝试重连...`); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.connect(); }, this.reconnectInterval); } /** * 启动客户端 */ start() { console.log(`[${new Date().toLocaleString()}] 启动WebSocket客户端...`); this.connect(); // Node.js环境下监听退出信号 if (typeof process !== 'undefined' && process.on) { process.on('SIGINT', () => { console.log(`\n[${new Date().toLocaleString()}] 收到退出信号，关闭客户端...`); this.stop(); process.exit(0); }); } } /** * 停止客户端 */ stop() { this.isConnected = false; this.subscribeSent = false; // 清理定时器 this.stopHeartbeat(); if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } // 关闭连接 if (this.ws) { this.ws.close(1000, 'Client stopped'); this.ws = null; } console.log(`[${new Date().toLocaleString()}] 客户端已停止`); } } // ==================== 运行示例 ==================== // 浏览器环境：直接在控制台执行以下代码 // Node.js环境：直接运行该脚本 // 替换为你的实际API Key const API_KEY = 'yourApikey'; const BUSINESS = 'crypto'; // 创建并启动客户端 const client = new CryptoWebsocketClient(API_KEY, BUSINESS); client.start(); // 如需停止客户端，执行：client.stop();