如何用Haskell编写维护连接池并广播消息的TCP服务器？（Node.js代码转译）

要实现一个维护活跃连接池、将收到的消息广播至所有连接的TCP服务器，在Haskell里我们可以利用network库处理TCP通信，结合STM（软件事务内存）安全管理共享的连接状态——这比Node.js里直接操作数组更能避免并发竞态问题。

依赖准备 #

首先在你的Cabal配置里添加所需依赖：

build-depends: base >= 4.14 && < 5, network, stm

完整实现代码 #

import Network.Socket
import Network.Socket.ByteString (recv, sendAll)
import Control.Concurrent (forkIO)
import Control.Concurrent.STM
import Control.Exception (finally, catch, IOException)
import Data.ByteString (ByteString)
import qualified Data.ByteString as BS

-- 用TVar包裹连接列表，实现线程安全的连接池
type ConnectionPool = TVar [Socket]

-- 处理单个客户端连接的逻辑
handleConnection :: ConnectionPool -> Socket -> IO ()
handleConnection pool sock = do
    -- 将新连接加入连接池
    atomically $ modifyTVar pool (sock :)
    -- 循环接收客户端数据并广播
    let loop = do
            msg <- recv sock 4096
            if BS.null msg
                then return ()  -- 客户端断开连接，退出循环
                else do
                    -- 获取当前所有活跃连接并广播消息
                    activeConns <- readTVarIO pool
                    -- 发送消息时捕获异常，避免单个连接失败导致广播中断
                    mapM_ (\conn -> sendAll conn msg `catch` (\(_::IOException) -> return ())) activeConns
                    loop
    -- 无论连接是正常关闭还是异常终止，都从连接池移除当前socket
    loop `finally` atomically (modifyTVar pool (filter (/= sock)))

-- 服务器主逻辑
main :: IO ()
main = do
    -- 创建TCP套接字
    serverSock <- socket AF_INET Stream 0
    setSocketOption serverSock ReuseAddr 1  -- 允许端口快速复用
    -- 绑定到127.0.0.1的1337端口
    bind serverSock (SockAddrInet 1337 (tupleToHostAddress (127,0,0,1)))
    listen serverSock 5  -- 启动监听，最大等待队列5个
    -- 初始化空的连接池
    connPool <- newTVarIO []
    putStrLn "Server listening on port 1337..."
    -- 循环接受新连接，每个连接开启独立线程处理
    let acceptLoop = do
            (clientSock, _) <- accept serverSock
            forkIO $ handleConnection connPool clientSock
            acceptLoop
    -- 确保服务器退出时关闭监听套接字
    acceptLoop `finally` close serverSock

关键逻辑说明 #

1. 线程安全的连接池：用TVar（STM的事务变量）存储活跃连接列表，所有对连接池的修改（添加/移除连接）都通过atomically包裹，保证并发下的原子性，避免竞态条件。
2. 连接生命周期管理：新连接通过modifyTVar加入池；无论连接正常关闭还是异常断开，finally块都会确保将其从池里移除，不会残留无效连接。
3. 消息广播：收到客户端消息后，遍历当前所有活跃连接发送，同时捕获发送时的IO异常（比如客户端已断开但还在池里），防止单个连接的失败影响整个广播流程。
4. 并发处理：每个新连接通过forkIO启动独立线程处理，保证服务器能同时服务多个客户端。

与Node.js代码的对应关系 #

• Node.js里的conns数组 → Haskell里的ConnectionPool（TVar [Socket]）
• 新连接加入conns → atomically $ modifyTVar pool (sock :)
• 数据广播遍历conns → 读取TVar后mapM_发送消息
• 连接关闭时从conns移除 → finally块里的filter操作

内容的提问来源于stack exchange，提问作者Malik Ammar Faisal