Production-Ready WebSocket Connection Pool for Real-Time Market Data with Load Balancing

Fatal problems of native WebSocket connections in real‑time market scenarios

Massive subscription use‑cases such as stock quotes, cryptocurrency prices, commodity feeds, live‑room pushes, and IoT device monitoring expose the following fatal issues when each subscription creates a new WebSocketClient:

No connection reuse – file‑descriptor leak Each subscription opens a new socket; rapid switching exhausts system file handles and throws too many open files , crashing the service.

No automatic reconnection – market data stalls Network jitter, node restarts or partitions break connections; without a recovery mechanism users see stale prices and must manually refresh.

Reconnection storms overwhelm the market service During a market swing many clients disconnect simultaneously and retry at once, saturating CPU and connection queues.

Cluster lacks load balancing – connection skew Clients randomly connect to a node; a hot node becomes overloaded while others stay idle.

No heartbeat – zombie connections accumulate Firewalls/Nginx close idle sockets after 60 s; without heartbeats the server retains dead connections, wasting memory and handles.

No message buffering – loss or blocking When a connection drops, pending market messages are dropped; synchronous sending blocks the main thread.

No circuit‑breaker – endless retries on a bad node A failed node is retried forever, consuming resources and preventing automatic failover.

Core design goals for a production‑grade WebSocket pool

The pool must provide seven capabilities to support high‑concurrency real‑time feeds:

Connection reuse and pooling

Layered load‑balancing with hash sharding

Built‑in heartbeat for keep‑alive

Exponential back‑off reconnection

Asynchronous message buffer queue

Fault isolation and circuit‑breaker

Complete lifecycle management (create, reuse, subscribe, cancel, close, destroy)

Overall four‑layer architecture

Load‑balancing shard layer

Market nodes register to Nacos/Eureka. The client hashes the target code to select a node, providing health‑node filtering, faulty‑node eviction and automatic shard migration.

Global pool manager layer

Manages all node‑specific sub‑pools and exposes unified subscribe/unsubscribe APIs while enforcing a global maximum connection count.

Per‑node connection buffer layer

Each market node has an independent sub‑pool that caches reusable WebSocket connections and caps the per‑node connection number.

Connection instance layer

Wraps the native WebSocketClient with heartbeat, async send queue, reconnection logic, subscription cache and callback handling.

Hash‑sharding logic

Random round‑robin is unsuitable for market data. The pool uses deterministic hash sharding:

shardIndex = Math.abs(targetCode.hashCode()) % healthyNodeList.size()

This guarantees that the same instrument always routes to the same node, reduces cross‑node traffic, and enables smooth re‑hashing when a node fails or a new node joins.

Maven dependencies (Java WebSocket standard client)

<dependency>
  <groupId>org.java-websocket</groupId>
  <artifactId>Java-WebSocket</artifactId>
  <version>1.5.4</version>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>fastjson2</artifactId>
  <version>2.0.32</version>
</dependency>
<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

Market node entity and shard balancer

import lombok.Data;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Data
public class MarketNode {
    private String nodeId;               // Unique node identifier
    private String wsUrl;                 // e.g. ws://127.0.0.1:8090/market
    private volatile boolean healthy = true; // Node health flag
    private int failCount = 0;           // Consecutive failure count for circuit‑breaker
}

/** Market sharding load‑balancer */
@Component
public class MarketShardBalancer {
    private final CopyOnWriteArrayList<MarketNode> healthyNodes = new CopyOnWriteArrayList<>();

    // Refresh node list from Nacos (periodic)
    public void refreshNodes(List<MarketNode> nodes) {
        healthyNodes.clear();
        healthyNodes.addAll(nodes);
    }

    // Get node by hash sharding
    public MarketNode getShardNode(String targetCode) {
        if (healthyNodes.isEmpty()) {
            throw new RuntimeException("No available market service nodes");
        }
        int idx = Math.abs(targetCode.hashCode()) % healthyNodes.size();
        return healthyNodes.get(idx);
    }

    // Mark node as failed; trigger circuit‑breaker
    public void markNodeFail(String nodeId) {
        for (MarketNode node : healthyNodes) {
            if (node.getNodeId().equals(nodeId)) {
                node.setFailCount(node.getFailCount() + 1);
                if (node.getFailCount() >= 5) {
                    node.setHealthy(false);
                }
                break;
            }
        }
    }

    // Periodic health‑probe to recover nodes
    public void recoverNodeHealth() {
        for (MarketNode node : healthyNodes) {
            if (!node.isHealthy()) {
                node.setFailCount(0);
                node.setHealthy(true);
            }
        }
    }
}

Reusable WebSocket client (per‑node)

import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import java.net.URI;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.CopyOnWriteArraySet;

@Slf4j
public class ReusableMarketWsClient {
    private final MarketNode node;
    private WebSocketClient webSocketClient;
    private final CopyOnWriteArraySet<String> subscribeTargets = new CopyOnWriteArraySet<>();
    private final BlockingQueue<String> msgQueue = new ArrayBlockingQueue<>(2000);
    private ScheduledFuture<?> heartbeatTask;
    private final int[] retryDelay = {1000, 2000, 4000, 8000};
    private int retryIndex = 0;
    private final AtomicBoolean isConnected = new AtomicBoolean(false);
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public ReusableMarketWsClient(MarketNode node) {
        this.node = node;
        initClient();
    }

    // Initialise the native client
    private void initClient() {
        try {
            URI uri = new URI(node.getWsUrl());
            webSocketClient = new WebSocketClient(uri) {
                @Override
                public void onOpen(ServerHandshake handshakedata) {
                    log.info("Market node {} connection opened", node.getNodeId());
                    isConnected.set(true);
                    retryIndex = 0;
                    startHeartbeat();
                    resubscribeAll();
                }

                @Override
                public void onMessage(String message) {
                    handleMarketMsg(message);
                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.warn("Market node {} connection closed, reason: {}", node.getNodeId(), reason);
                    isConnected.set(false);
                    stopHeartbeat();
                    scheduleReconnect();
                }

                @Override
                public void onError(Exception ex) {
                    log.error("Market node {} connection error", node.getNodeId(), ex);
                    MarketShardBalancer balancer = SpringContextUtil.getBean(MarketShardBalancer.class);
                    balancer.markNodeFail(node.getNodeId());
                }
            };
            webSocketClient.connect();
            startMsgSender();
        } catch (Exception e) {
            log.error("Failed to create ws connection", e);
            scheduleReconnect();
        }
    }

    // Start 30 s periodic bidirectional heartbeat
    private void startHeartbeat() {
        heartbeatTask = scheduler.scheduleAtFixedRate(() -> {
            if (isConnected.get()) {
                sendMsg("{\"type\":\"heartbeat\"}");
            }
        }, 0, 30, TimeUnit.SECONDS);
    }

    private void stopHeartbeat() {
        if (heartbeatTask != null) {
            heartbeatTask.cancel(true);
        }
    }

    // Async message sender thread
    private void startMsgSender() {
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    String msg = msgQueue.take();
                    if (isConnected.get() && webSocketClient.isOpen()) {
                        webSocketClient.send(msg);
                    }
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("Failed to send market message", e);
                }
            }
        }, "ws-msg-sender-" + node.getNodeId()).start();
    }

    // Public API to send a message (non‑blocking)
    public void sendMsg(String msg) {
        if (!msgQueue.offer(msg)) {
            log.warn("Message queue full, dropping market message");
        }
    }

    // Subscribe a target symbol
    public void subscribe(String targetCode) {
        subscribeTargets.add(targetCode);
        String subMsg = String.format("{\"type\":\"subscribe\",\"code\":\"%s\"}", targetCode);
        sendMsg(subMsg);
    }

    // Unsubscribe a target symbol
    public void unsubscribe(String targetCode) {
        subscribeTargets.remove(targetCode);
        String unSubMsg = String.format("{\"type\":\"unsubscribe\",\"code\":\"%s\"}", targetCode);
        sendMsg(unSubMsg);
    }

    // Schedule reconnection with exponential back‑off
    private void scheduleReconnect() {
        int delay = retryDelay[Math.min(retryIndex, retryDelay.length - 1)];
        retryIndex++;
        scheduler.schedule(this::initClient, delay, TimeUnit.MILLISECONDS);
        log.info("Node {} will attempt reconnection after {} ms", node.getNodeId(), delay);
    }

    // After reconnection, restore all subscriptions
    private void resubscribeAll() {
        for (String code : subscribeTargets) {
            subscribe(code);
        }
    }

    // Business callback placeholder (implemented by user)
    private void handleMarketMsg(String message) {
        // Forward to market processing, cache, or front‑end push
    }

    // Determine if the connection is idle (no subscriptions)
    public boolean isIdle() {
        return subscribeTargets.isEmpty();
    }

    // Close and clean up resources
    public void close() {
        stopHeartbeat();
        scheduler.shutdownNow();
        if (webSocketClient != null) {
            webSocketClient.close();
        }
        subscribeTargets.clear();
        msgQueue.clear();
    }

    public boolean isConnected() {
        return isConnected.get();
    }
}

Single‑node connection pool (controls max connections, reuses, recycles idle sockets)

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.concurrent.*;

@Slf4j
public class NodeWsPool {
    private final MarketNode node;
    private final BlockingQueue<ReusableMarketWsClient> clientQueue = new LinkedBlockingQueue<>();
    private final int maxConn = 8;                     // Max connections per node
    private static final long IDLE_TIMEOUT = 60 * 1000L; // 60 s idle timeout

    public NodeWsPool(MarketNode node) {
        this.node = node;
    }

    // Obtain a client: reuse idle one first, otherwise create up to maxConn, else return head of queue
    public synchronized ReusableMarketWsClient getClient() {
        ReusableMarketWsClient idle = clientQueue.poll();
        if (idle != null && idle.isConnected()) {
            clientQueue.add(idle);
            return idle;
        }
        if (clientQueue.size() < maxConn) {
            ReusableMarketWsClient newClient = new ReusableMarketWsClient(node);
            clientQueue.add(newClient);
            return newClient;
        }
        // Max reached – reuse the oldest connection
        return clientQueue.peek();
    }

    // Periodic cleanup of long‑idle connections
    @Scheduled(fixedRate = 30000)
    public void cleanIdleClient() {
        int size = clientQueue.size();
        for (int i = 0; i < size; i++) {
            ReusableMarketWsClient client = clientQueue.poll();
            if (client == null) break;
            if (client.isIdle()) {
                client.close();
                log.info("Recycled idle ws connection for node {}", node.getNodeId());
            } else {
                clientQueue.add(client);
            }
        }
    }

    // Destroy all connections of this node (e.g., on node removal)
    public void destroyAll() {
        ReusableMarketWsClient client;
        while ((client = clientQueue.poll()) != null) {
            client.close();
        }
    }
}

Global pool manager (unified entry point for business code)

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class GlobalMarketWsPoolManager {
    @Autowired
    private MarketShardBalancer balancer;
    private final Map<String, NodeWsPool> nodePoolMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // Periodically pull node list from Nacos and refresh pools (implementation omitted)
    }

    // Business‑level subscription entry
    public void subscribeMarket(String targetCode) {
        MarketNode node = balancer.getShardNode(targetCode);
        NodeWsPool pool = nodePoolMap.computeIfAbsent(node.getNodeId(), k -> new NodeWsPool(node));
        ReusableMarketWsClient client = pool.getClient();
        client.subscribe(targetCode);
    }

    // Business‑level un‑subscription entry
    public void unsubscribeMarket(String targetCode) {
        MarketNode node = balancer.getShardNode(targetCode);
        NodeWsPool pool = nodePoolMap.get(node.getNodeId());
        if (pool == null) return;
        ReusableMarketWsClient client = pool.getClient();
        client.unsubscribe(targetCode);
    }

    @PreDestroy
    public void shutdown() {
        nodePoolMap.values().forEach(NodeWsPool::destroyAll);
        log.info("All global WebSocket connections have been destroyed");
    }
}

Practical tips derived from the analysis

Issue: Bare connections cause "too many open files". Solution: Pool reuse and idle‑reclamation.

Issue: Massive simultaneous reconnections overload CPU. Solution: Exponential back‑off spreads reconnection attempts.

Issue: Random direct connections lead to node skew. Solution: Hash‑based sharding distributes load evenly.

Issue: Firewalls close idle sockets, stopping market updates. Solution: 30 s bidirectional heartbeat.

Issue: Synchronous message sending blocks business threads. Solution: Independent async queue decouples IO from business logic.

Capability comparison – Native WebSocket vs. Production Pool

Connection reuse : Native – none (new socket each time). Production – supported; multiple symbols share one long‑lived connection.

File‑handle control : Native – no limit, easy overflow. Production – per‑node max connections, idle reclamation.

Reconnection : Native – none, manual implementation required. Production – exponential back‑off with automatic subscription restore.

Heartbeat : Native – none. Production – built‑in 30 s heartbeat to keep connections alive.

Message sending : Native – synchronous, blocks main thread. Production – async queue, non‑blocking.

Cluster load balancing : Native – random direct, leads to skew. Production – hash sharding, even traffic distribution.

Fault isolation : Native – infinite retries on bad node. Production – circuit‑breaker, automatic failover to healthy nodes.

Stability for massive feeds : Native – poor, prone to avalanche and message loss. Production – high‑availability with rate‑limit, buffering, fault isolation.

Full‑text conclusion

For any scenario that relies on massive long‑lived connections—stock quotes, cryptocurrency tickers, IoT status reports, live‑room bullet chats—using a one‑off WebSocket client is untenable. The presented architecture delivers three pillars: pooled reuse with connection caps, hash‑based sharding for cluster‑wide load balancing, and heartbeat + exponential back‑off for resilient reconnection. Combined with async buffering, circuit‑breaker, idle reclamation and monitoring, it eliminates handle leaks, reconnection storms, connection skew, data gaps and message loss, enabling stable delivery of hundreds of thousands of real‑time subscriptions.