LaserStream WebSocket：实时 Solana 数据流

什么是 LaserStream WebSocket?

LaserStream WebSocket 是 LaserStream 的 WebSocket 协议变体。它提供标准的 Solana JSON-RPC 订阅方法（accountSubscribe, programSubscribe, logsSubscribe, signatureSubscribe, slotSubscribe, …)，以及诸如 transactionSubscribe 的 Helius 特定扩展用于高级过滤——全部在同一统一 wss://mainnet.helius-rpc.com 端点上，并在与 gRPC 产品相同的 LaserStream 后端上。 WebSockets 为您提供持久的实时连接：而不是反复轮询“是否有变化？”，区块链会在发生变化的那一刻通知您。

实时更新

当 Solana 帐户更改、交易发生或生成新区块时，立即收到通知

高效资源使用

一个持久连接，而不是轮询数百个 HTTP 请求

低延迟

比标准 Agave 基于 RPC 的 WebSockets 快最多 200 毫秒，由 LaserStream 提供服务

Helius 扩展

transactionSubscribe 和增强的 accountSubscribe 用于在标准 Solana 方法之上进行高级过滤

端点

WebSockets ——包括标准 Solana 方法和 Helius 特定扩展 —— 都从每个网络的单一统一端点提供： 主网: wss://mainnet.helius-rpc.com/?api-key=<API_KEY> 测试网: wss://devnet.helius-rpc.com/?api-key=<API_KEY>

Gatekeeper (Beta)

要测试我们最快的 WebSocket 产品，可以试用新的 Gatekeeper (Beta) 端点： wss://beta.helius-rpc.com/?api-key=<API_KEY>

定价与计量

WebSocket 使用按照每 0.1 MB 流数据 2 积分计量。打开连接成本为 1 积分。对于专业计划中的高负载工作，数据附加项提供可预测成本的月度数据限额（5TB至100TB）。数据附加项可以应用于LaserStream和WebSockets。

特定于Helius的方法 transactionSubscribe 和 accountSubscribe（带有Helius过滤器）需要开发者、企业或专业计划。标准Solana方法如 programSubscribe、logsSubscribe 和 signatureSubscribe 在所有计划中都可用。

核心概念

订阅类型

账户订阅

监控特定账户的更改，如钱包余额、代币账户或程序数据。用例：

钱包余额跟踪
代币账户监控
智能合约状态变化
NFT所有权更新

程序订阅

监视特定程序拥有的所有账户的任何更改。用例：

DEX交易监控
DeFi协议跟踪
NFT市场活动
游戏资产变化

交易订阅

当特定交易被确认或交易涉及某些账户时接收通知。用例：

支付确认
交易状态跟踪
多重签名批准
合约执行监控

插槽订阅

监控区块链在插槽级别的进展和最终性。用例：

区块浏览器应用程序
网络健康监控
共识跟踪
性能分析

承诺等级

理解承诺等级对可靠应用至关重要：

已处理
已确认
已完成

最快 - 交易已由验证器处理但未确认

延迟：约400毫秒
风险：可能被删除或重新排序
用于：实时UI更新（需谨慎）

想了解更多关于承诺级别的信息吗？ 请阅读我们的综合博客文章：理解 Solana 承诺级别

可用的 WebSocket 方法

Helius 支持完整的 Solana WebSocket 订阅方法集，加上 Helius 特定的扩展。所有方法共享相同的 wss://mainnet.helius-rpc.com 端点。

Helius 扩展

两种方法提供比 Solana 标准更丰富的过滤：

transactionSubscribe — 订阅符合包含/排除/必需账户列表的交易、投票/失败标志和特定签名
accountSubscribe — Helius 在标准 accountSubscribe 形状基础上添加高级过滤器

标准 Solana 方法

accountSubscribe, blockSubscribe, logsSubscribe, programSubscribe, rootSubscribe, signatureSubscribe, slotSubscribe, slotsUpdatesSubscribe, voteSubscribe 及其对应的 *Unsubscribe 配对。

完整 API 参考

探索所有 18+ WebSocket 方法。每种方法包括详细的参数、响应格式和示例。

如何实现重连逻辑

WebSocket 连接可能因多种原因断开 - 网络问题、服务器维护或临时中断。生产应用程序必须实现重连逻辑以确保可靠性。

断开连接的原因

网络问题

互联网连接问题
移动设备的 WiFi 切换
企业防火墙超时
ISP 路由更改

服务器端事件

计划的维护窗口
负载均衡器重启
RPC 节点更新
临时过载保护

客户端问题

浏览器标签后台运行
移动应用暂停
计算机休眠/冬眠
进程崩溃或重启

检测断开连接

JavaScript/Browser
Node.js

class ConnectionMonitor {
  constructor(ws) {
    this.ws = ws;
    this.pingInterval = null;
    this.lastPong = Date.now();
    this.isConnected = false;
    
    this.setupEventListeners();
    this.startPingMonitoring();
  }
  
  setupEventListeners() {
    this.ws.onopen = () => {
      console.log('Connected');
      this.isConnected = true;
      this.lastPong = Date.now();
    };
    
    this.ws.onclose = (event) => {
      console.log('Disconnected:', event.code, event.reason);
      this.isConnected = false;
      this.stopPingMonitoring();
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
      this.isConnected = false;
    };
    
    // Listen for pong responses (server acknowledgment)
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.method === 'pong') {
        this.lastPong = Date.now();
      }
      // Handle other messages...
    };
  }
  
  startPingMonitoring() {
    this.pingInterval = setInterval(() => {
      if (this.isConnected) {
        // Send ping to check connection health
        this.ws.send(JSON.stringify({
          jsonrpc: '2.0',
          method: 'ping',
          id: Date.now()
        }));
        
        // Check if we received a pong recently
        const timeSinceLastPong = Date.now() - this.lastPong;
        if (timeSinceLastPong > 30000) { // 30 seconds timeout
          console.warn('No pong received, connection may be stale');
          this.ws.close();
        }
      }
    }, 10000); // Ping every 10 seconds
  }
  
  stopPingMonitoring() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }
}

const WebSocket = require('ws');

class NodeConnectionMonitor {
  constructor(url) {
    this.url = url;
    this.ws = null;
    this.pingInterval = null;
    this.isAlive = false;
  }
  
  connect() {
    this.ws = new WebSocket(this.url);
    
    this.ws.on('open', () => {
      console.log('Connected');
      this.isAlive = true;
      this.startHeartbeat();
    });
    
    this.ws.on('close', () => {
      console.log('Disconnected');
      this.isAlive = false;
      this.stopHeartbeat();
    });
    
    this.ws.on('error', (error) => {
      console.error('WebSocket error:', error);
      this.isAlive = false;
    });
    
    // Built-in ping/pong handling
    this.ws.on('pong', () => {
      this.isAlive = true;
    });
  }
  
  startHeartbeat() {
    this.pingInterval = setInterval(() => {
      if (!this.isAlive) {
        console.log('Connection lost, terminating');
        return this.ws.terminate();
      }
      
      this.isAlive = false;
      this.ws.ping();
    }, 30000); // Ping every 30 seconds
  }
  
  stopHeartbeat() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }
}

重连策略

指数退避
断路器模式

class ExponentialBackoffReconnector {
  constructor(url, maxRetries = 10) {
    this.url = url;
    this.maxRetries = maxRetries;
    this.retryCount = 0;
    this.baseDelay = 1000; // Start with 1 second
    this.maxDelay = 30000; // Cap at 30 seconds
    this.ws = null;
    this.subscriptions = new Map();
    this.isReconnecting = false;
  }
  
  connect() {
    if (this.isReconnecting) return;
    
    try {
      this.ws = new WebSocket(this.url);
      this.setupEventHandlers();
    } catch (error) {
      console.error('Failed to create WebSocket:', error);
      this.scheduleReconnect();
    }
  }
  
  setupEventHandlers() {
    this.ws.onopen = () => {
      console.log('Connected successfully');
      this.retryCount = 0; // Reset retry count on successful connection
      this.isReconnecting = false;
      this.resubscribeAll(); // Restore subscriptions
    };
    
    this.ws.onclose = (event) => {
      console.log('Connection closed:', event.code);
      if (!this.isReconnecting) {
        this.scheduleReconnect();
      }
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };
  }
  
  scheduleReconnect() {
    if (this.retryCount >= this.maxRetries) {
      console.error('Max retry attempts reached. Giving up.');
      return;
    }
    
    this.isReconnecting = true;
    this.retryCount++;
    
    // Calculate delay with exponential backoff + jitter
    const delay = Math.min(
      this.baseDelay * Math.pow(2, this.retryCount - 1),
      this.maxDelay
    );
    
    // Add jitter to prevent thundering herd
    const jitteredDelay = delay + (Math.random() * 1000);
    
    console.log(`Reconnecting in ${jitteredDelay}ms (attempt ${this.retryCount}/${this.maxRetries})`);
    
    setTimeout(() => {
      this.connect();
    }, jitteredDelay);
  }
  
  subscribe(method, params, callback) {
    const id = this.generateId();
    this.subscriptions.set(id, { method, params, callback });
    
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.sendSubscription(id, method, params);
    }
    
    return id;
  }
  
  resubscribeAll() {
    console.log(`Restoring ${this.subscriptions.size} subscriptions`);
    for (const [id, sub] of this.subscriptions) {
      this.sendSubscription(id, sub.method, sub.params);
    }
  }
  
  sendSubscription(id, method, params) {
    this.ws.send(JSON.stringify({
      jsonrpc: '2.0',
      id: id,
      method: method,
      params: params
    }));
  }
  
  generateId() {
    return Date.now() + Math.random();
  }
}

class CircuitBreakerWebSocket {
  constructor(url, options = {}) {
    this.url = url;
    this.failureThreshold = options.failureThreshold || 5;
    this.recoveryTimeout = options.recoveryTimeout || 60000;
    this.checkInterval = options.checkInterval || 10000;
    
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    this.failureCount = 0;
    this.lastFailureTime = null;
    this.ws = null;
    this.subscriptions = new Map();
  }
  
  connect() {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime > this.recoveryTimeout) {
        console.log('Circuit breaker moving to HALF_OPEN state');
        this.state = 'HALF_OPEN';
      } else {
        console.log('Circuit breaker is OPEN, connection blocked');
        return false;
      }
    }
    
    try {
      this.ws = new WebSocket(this.url);
      this.setupEventHandlers();
      return true;
    } catch (error) {
      this.recordFailure();
      return false;
    }
  }
  
  setupEventHandlers() {
    this.ws.onopen = () => {
      console.log('Connected - Circuit breaker CLOSED');
      this.state = 'CLOSED';
      this.failureCount = 0;
      this.resubscribeAll();
    };
    
    this.ws.onclose = () => {
      this.recordFailure();
    };
    
    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
      this.recordFailure();
    };
  }
  
  recordFailure() {
    this.failureCount++;
    this.lastFailureTime = Date.now();
    
    console.log(`Failure recorded: ${this.failureCount}/${this.failureThreshold}`);
    
    if (this.failureCount >= this.failureThreshold) {
      console.log('Circuit breaker opened due to repeated failures');
      this.state = 'OPEN';
    }
  }
  
  startHealthCheck() {
    setInterval(() => {
      if (this.state === 'OPEN' || 
          (this.ws && this.ws.readyState !== WebSocket.OPEN)) {
        this.connect();
      }
    }, this.checkInterval);
  }
}

测试重新连接逻辑

网络模拟
自动化测试

// Test disconnection scenarios
class NetworkSimulator {
  constructor(wsManager) {
    this.wsManager = wsManager;
  }
  
  // Simulate network outage
  simulateNetworkOutage(duration = 5000) {
    console.log('Simulating network outage...');
    
    // Force close the connection
    if (this.wsManager.ws) {
      this.wsManager.ws.close(1006, 'Network outage simulation');
    }
    
    // Block reconnection temporarily
    const originalConnect = this.wsManager.connect.bind(this.wsManager);
    this.wsManager.connect = () => {
      console.log('Connection blocked during outage simulation');
    };
    
    // Restore connection after duration
    setTimeout(() => {
      console.log('Network restored');
      this.wsManager.connect = originalConnect;
      this.wsManager.connect();
    }, duration);
  }
  
  // Simulate intermittent connectivity
  simulateIntermittentConnectivity() {
    setInterval(() => {
      if (Math.random() < 0.1) { // 10% chance every 10 seconds
        console.log('Simulating connection drop...');
        this.wsManager.ws?.close(1006, 'Intermittent connectivity');
      }
    }, 10000);
  }
}

// Usage
const simulator = new NetworkSimulator(wsManager);
simulator.simulateNetworkOutage(10000); // 10 second outage

// Jest test example
describe('WebSocket Reconnection', () => {
  let wsManager;
  
  beforeEach(() => {
    wsManager = new ProductionWebSocketManager({
      endpoint: 'ws://localhost:8080',
      apiKey: 'test-key'
    });
  });
  
  test('should reconnect after connection loss', async () => {
    const reconnectPromise = new Promise((resolve) => {
      wsManager.on('connected', resolve);
    });
    
    await wsManager.connect();
    
    // Simulate connection loss
    wsManager.ws.close(1006, 'Test disconnection');
    
    // Wait for reconnection
    await reconnectPromise;
    
    expect(wsManager.connectionState).toBe('CONNECTED');
    expect(wsManager.metrics.reconnections).toBeGreaterThan(0);
  });
  
  test('should restore subscriptions after reconnection', async () => {
    await wsManager.connect();
    
    const messages = [];
    const subscriptionId = wsManager.subscribe(
      'accountSubscribe',
      ['test-account', {}],
      (data) => messages.push(data)
    );
    
    // Verify subscription exists
    expect(wsManager.subscriptions.has(subscriptionId)).toBe(true);
    
    // Simulate disconnection and reconnection
    wsManager.ws.close(1006, 'Test disconnection');
    
    await new Promise(resolve => wsManager.on('connected', resolve));
    
    // Verify subscription was restored
    const subscription = wsManager.subscriptions.get(subscriptionId);
    expect(subscription.pending).toBe(true); // Should be re-subscribing
  });
});

生产环境关键：在生产应用中实现正确的重新连接逻辑是不可或缺的。WebSocket 连接会断开 - 需要为此做好计划、测试并在生产中进行监控。

疑难解答

连接失败

症状：WebSocket 无法初始连接解决方案：

确认您的 API key 正确并具有足够的配额
检查端点 URL 格式：wss://mainnet.helius-rpc.com?api-key=YOUR_KEY
确保防火墙允许 WebSocket 在端口 443 上的连接
先进行简单的 ping 测试以验证基本连接

频繁断线

症状：连接每隔几分钟就会断开解决方案：

WebSockets 有一个 10 分钟的空闲计时器。至少每分钟发送一次 ping 保持连接（见上述重新连接示例和 WebSocket 常见问题）
实现正确的 ping/pong 心跳（见上述重新连接示例）
检查网络稳定性和企业防火墙设置
监控您的订阅数量 - 过多可能导致不稳定
确认您正确处理了连接的生命周期

消息丢失

症状：未接收到预期的订阅更新解决方案：

确认您的订阅已确认（检查响应）
确保您正在监控的账户/程序有实际活动
监控您的连接状态 - 丢失消息通常表示断开连接

高延迟

症状：消息传递缓慢，更新延迟解决方案：

使用“confirmed”而不是“finalized”承诺
减少活跃订阅的数量
优化您的消息处理逻辑
考虑使用多个连接以分散负载
检查您的网络连接质量

内存泄漏

症状：应用程序内存使用量随时间增长解决方案：

实现适当的订阅清理
在组件卸载时移除事件监听器
定期清除消息日志
监控订阅数量并强制限制
尽可能使用弱引用回调函数

从标准RPC迁移

如果您当前使用的是HTTP轮询，这是迁移到WebSockets的方法：

// Old approach - HTTP polling
class HTTPAccountMonitor {
  constructor(connection, accountAddress) {
    this.connection = connection;
    this.accountAddress = accountAddress;
    this.interval = null;
    this.lastKnownBalance = null;
  }

  start() {
    this.interval = setInterval(async () => {
      try {
        const accountInfo = await this.connection.getAccountInfo(
          new PublicKey(this.accountAddress)
        );
        
        const currentBalance = accountInfo?.lamports || 0;
        
        if (this.lastKnownBalance !== currentBalance) {
          console.log(`Balance changed: ${currentBalance}`);
          this.lastKnownBalance = currentBalance;
        }
      } catch (error) {
        console.error('Failed to fetch account info:', error);
      }
    }, 2000); // Poll every 2 seconds
  }

  stop() {
    if (this.interval) {
      clearInterval(this.interval);
      this.interval = null;
    }
  }
}

// New approach - WebSocket subscription
class WebSocketAccountMonitor {
  constructor(wsManager, accountAddress) {
    this.wsManager = wsManager;
    this.accountAddress = accountAddress;
    this.subscriptionId = null;
  }

  start() {
    this.subscriptionId = this.wsManager.subscribe(
      'accountSubscribe',
      [
        this.accountAddress,
        { encoding: 'jsonParsed', commitment: 'confirmed' }
      ],
      (data) => {
        const currentBalance = data.value.lamports;
        console.log(`Balance changed: ${currentBalance}`);
        
        // Handle the change immediately - no polling delay!
      }
    );
  }

  stop() {
    if (this.subscriptionId) {
      this.wsManager.unsubscribe(this.subscriptionId);
      this.subscriptionId = null;
    }
  }
}

增强功能

对于需要更高级功能的应用程序，请考虑使用LaserStream gRPC：

24小时历史重播

在您的应用程序离线时补上错过的数据

多节点聚合

通过来自多个验证器的数据提高可靠性

更高吞吐量

处理更多订阅和更高的消息速率

企业功能

高级监控、分析和数据管道

​什么是 LaserStream WebSocket?

实时更新

高效资源使用

低延迟

Helius 扩展

​端点

​Gatekeeper (Beta)

​定价与计量

​核心概念

​订阅类型

​承诺等级

​可用的 WebSocket 方法

​Helius 扩展

​标准 Solana 方法

完整 API 参考

​如何实现重连逻辑

​断开连接的原因

​检测断开连接

​重连策略

​测试重新连接逻辑

​疑难解答

​从标准RPC迁移

​增强功能

24小时历史重播

多节点聚合

更高吞吐量

企业功能

​开始使用

什么是 LaserStream WebSocket?

端点

Gatekeeper (Beta)

定价与计量

核心概念

订阅类型

承诺等级

可用的 WebSocket 方法

Helius 扩展

标准 Solana 方法

如何实现重连逻辑

断开连接的原因

检测断开连接

重连策略

测试重新连接逻辑

疑难解答

从标准RPC迁移

增强功能

开始使用