★ FEATURED ARTICLE 后端Web框架【免费下载链接】fountain一个用于服务器应用开发的综合工具库。 - 零配置文件 - 环境变量和命令行参数配置 - 约定优于配置 - 深刻利用仓颉语言特性 - 只需要开发动态链接库fboot负责加载、初始化并运行 - 覆盖服务器应用开发的各种场景项目地址https://gitcode.com/Cangjie-SIG/fountain点击查看免费下载f_net 是 fountain 框架的 TCP 传输层模块它把f_protocol的Message帧读写真正落到 TCP 连接上并统一管理连接生命周期心跳巡检、失败计数、限流执行、关闭时序向上层f_rpc的RPCClient提供多节点客户端池MultiClient。读完本文你将掌握 f_net 的包结构、Client/Server的完整 API 与构建参数、心跳与断链重建机制以及MultiClient累计权重桶的负载均衡语义并能在自己的服务端应用里直接写出可运行的收发代码。f_net 的定位连接之上的帧读写与生命周期管理f_net 本身不定义消息协议它复用的是 f_protocol 的Message。Message的帧格式为version(1B) cmd(1B) len(4B 大端) payload crc32(4B)帧校验采用 f_util 的标准 CRC-32IEEE 802.3增量接口见 Message.cj。f_net 的角色是把一条条Message帧写进 socket、从 socket 读出一帧帧MessageSocketBuffer.cj管理连接生命周期心跳巡检、连续失败计数、限流执行、关闭时序对上层的RPCClient提供「多节点客户端池」MultiClient。从 cjpm.toml 可以看到 f_net 的依赖为f_base、f_concurrent、f_data、f_log、f_random、f_protocol输出类型为dynamic动态链接库由 fboot 加载初始化编译选项--dy-stdcjc 版本要求 1.2.0。模块描述为「事件驱动的网络通讯模块」。包结构与公开类型一览f_net 划分为三个包包内容fountain::f_netMessageFuture、SocketBuffer、SocketParamsinternal仅经 builder 间接使用fountain::f_net.clientClient/ClientBuilder、MultiClient/MultiClientBuilder、ListeningClient、ClientExceptionfountain::f_net.serverServer/ServerBuilderSocketParams是内部类见 SocketParams.cj它承载bindToDevice/keepAlive/linger/noDelay/quickAcknowledge/readTimeout/receiveBufferSize/sendBufferSize/writeTimeout及三类原生socketOption设置最终通过populate(socket)落到TcpSocket上ClientBuilder与ServerBuilder都复用这一套参数。SocketParams提供copy()生成独立副本ServerBuilder.build()传params.copy()避免与监听 socket 的设置互相污染对应 net_test.cj 中copyProducesIndependentClone用例。基本类型之一MessageFuture —— 并发安全的响应占位符MessageFutureDE是「等待某个响应的句柄」公共函数并发安全public class MessageFutureDE { /** 没有数据就一直阻塞直到读取线程解码出结果并唤醒本线程 */ public func get(): DE /** 没有数据立即返回 NoneDE */ public func tryGet(): ?DE /** 没有数据就阻塞到 timeout超时后立即返回超时前有数据会被唤醒 */ public func tryGet(timeout: Duration): ?DE }从源码看其实现是Mutex 条件变量 ?DE槽位见 MessageFuture.cjset()在锁内写入ref并condition.notify()get()循环condition.wait()直到非 NonetryGet(timeout)用DateTime.now()计算 deadline 做限时等待。set()不是 public只有同包的读取线程会调用它填充结果。两个要点MessageFuture持有的是原始Message即MessageFutureMessage载荷的多态类型要调用方自己convertT(msg.data)转换net_test.cj中 MessageFutureTest 覆盖了「先 set 后 get」「未 set 时 tryGet 返回 None」「限时等待中 set 会唤醒并返回」「重复 get 幂等」等行为。基本类型之二SocketBuffer —— 一条连接上的帧缓冲区SocketBuffer : IOStream Resource每个 TCP 连接对应一个由Client/Server内部创建构造器不公开public class SocketBuffer : IOStream Resource { /** 编码 message 并写到 socket整帧写出的互斥见 close() */ public func transfer(message: Message): Unit public func read(bytes: ArrayByte): Int64 // 读满一帧用见 f_protocol.Message.decode public func write(bytes: ArrayByte): Unit public func flush(): Unit public func isClosed(): Bool /** 先等在写的那一帧退场上限 2s再关 socket对端看到的是「完整帧 EOF」不是半帧 */ public func close(): Unit public prop localAddress: SocketAddress public prop remoteAddress: SocketAddress }源码级行为见 SocketBuffer.cjtransfer(message)并不直接写 socket而是把写闭包投进一个ArrayBlockingQueue() - Unit容量即queueSize由后台 spawn 线程逐个取出执行——写路径异步、按队列串行整帧写出的互斥beginFrame()/endFrame()用frameLock保护writingFrame标志关闭时序close()先置停止信号再awaitFrameWrite()等待正在写的那一帧退场上限FRAME_CLOSE_WAIT 2s最后才关 socket。这样正常关闭时序下对端看到的是「完整帧 EOF」而不是半帧若 2 秒内仍等不到写帧结束记 WARN 后照常关闭——不把「对端看到半帧」换成「close 挂死」SocketBuffer.cjisClosed()是「懒关闭」的发现isStopped()而 socket 未关时同样先等待写帧退场再关 socket。客户端 Client单点连接的请求 / 响应通道ClientT是单点连接的客户端T约束T : DataFieldsT服务端解码载荷用的类型。public class ClientT where T : DataFieldsT { /** 4 个 builder 重载host!port! / address!:(String,UInt16) / address!:String / address!:IPSocketAddress */ public static func builder(queueSize!: Int64 1024, socketCount!: Int64 1, address!: String, checkDuration!: Duration, checker!: () - ?Message): ClientBuilderT /** 消息自带 idMessage.id返回原始响应 Message */ public func transfer(message: Message): MessageFutureMessage /** 等待响应的超时版本超时或 client 已关闭返回 None */ public func transfer(message: Message, timeout: Duration): ?Message /** 订阅/推送式收流executor 收到的是该 Message 之后的响应流 */ public func transferT(message: Message, executor: (Message) - Unit): ListeningClientT where T : DataFieldsT public prop address: IPSocketAddress public func isClosed(): Bool /** 幂等取消巡检定时器 关闭全部连接缓冲 */ public func close(): Unit }builder有 4 个重载地址可给host port、(String, UInt16)、host:port字符串、或IPSocketAddress见 client.cj。socketCount表示同时建立的连接数默认 1queueSize是写任务队列容量默认 1024checkDuration是连接有效性巡检周期checker是巡检用的检查包生成函数。常用构建参数ClientBuilderTbuild(): ClientT的常用设置bindToDevice/keepAlive/linger/noDelay/quickAcknowledge/readTimeout/receiveBufferSize/sendBufferSize/writeTimeout/socketOption(level, option, value, valueLength)/socketOptionBool/socketOptionIntNative/pingTimeout。pingTimeout默认 1 秒见 client.cj。这个值不能设得太短10ms 对一次 PING 往返4 次线程唤醒 2 次编解码过紧调度稍有抖动即超时会引发拆链重连风暴放宽到 1s 是为了把「单次抖动」与「真实断链」区分开。读循环与中性解码Client构造时为每个 socket 建一个SocketBuffer并 spawn 一个读线程见 client.cj。读循环的关键点是用Message.decodeDataAny做中性解码客户端收到的响应载荷类型是多样的ArrayString、ArrayServiceMeta…若按T如RPCMessage解码会把这类载荷「宽松转换」成字段为空的RPCMessage调用方随后再转换就会出错。因此这里解码成DataAny原始Data交给调用方自行convertT收到PING直接回Message.ack(id)并移除对应 future按message.id从futuresConcurrentHashMapMessageID, MessageFutureMessage取出 future 并set()填充结果若响应到了但该 id 上已无等待者id 不匹配 / 已被超时移除记 DEBUG 日志此前是完全静默是排查时的盲区读到 EOFInputClosedException或解码失败 字节流已错位必须断链不能继续在错位的流上解析。心跳巡检与断链重建巡检定时器按checkDuration周期执行见 client.cj对每条连接连接已关闭被 reader 或定时器关闭→ 直接new()重建否则永久失去该连接发送checker()生成的检查包在pingTimeout内等待响应一次检查成功要求res.id m.id res.command Command.ACK单次失败不立刻拆链checkFailures[i]累加连续失败达到CHECK_FAILURE_TOLERANCE 3client.cj才buffer.close()后重建——把单次调度抖动放大成重连风暴是设计上明确要避免的所有 buffer 都关闭后Client自行close()。行为要点汇总所有 buffer 关闭后Client自行关闭transfer在已关闭时抛Exception(Client is closed)close()幂等取消巡检定时器 关闭全部连接缓冲client.cj构建失败会抛出ClientExceptionClientException.cj继承f_base.BaseException。MultiClient多节点客户端池与累计权重桶MultiClientW, T是Resource IterableClientT把负载均衡算法与一组Client绑在一起是f_rpc的多节点调用基础。public class MultiClientW, T : Resource IterableClientT { public static func builder(queueSize!: Int64 1024, socketCount!: Int64 1, checkDuration!: Duration, checker!: () - ?Message): MultiClientBuilderW, T } public class MultiClientBuilderW, T where W : AddableW ComparableW, T : DataFieldsT { /** 轮询step 步长、min/max 算法取值区间、origin 是累计权重键的起点缺省 min兼容旧行为 */ public func roundRobin(step: W, min: W, max: W, origin!: ?W None): This /** 随机在 [min, max) 上按权重取值 */ public func random(min: W, max: W): This /** 添加节点weight 为该节点的权重地址可给 String / (host, port) / IPSocketAddress也可传 Iterable */ public func add(weight: W, address: String): This public func build(): MultiClientW, T }构建顺序是先调roundRobin(...)或random(...)选定算法再add(weight, address)添加节点最后build()源码注释在 MultiClient.cj 反复强调这一顺序。add支持单个地址String / hostport / IPSocketAddress 三种形式或三种Iterable(W, ...)批量形式每次add会用同一套参数构建一个Client并加入负载均衡器。约束与语义W只支持Int64/Float64其它类型抛LoadBalanceException见 MultiClient.cj 的 match 分支迭代语义MultiClient是IterableClientTfor(client in multiClient)按负载算法顺序产出客户端首元素即本次选中的节点后续元素是重试节点。f_rpc的RPCClient正是依赖这一语义实现失败重试见下文。权重语义累计权重桶务必注意的坑内部实现用 TreeMap 以各节点的累计权重为键origin w₁ … wᵢ取节点 第一个键 ≥ 算法给出的值因此节点 i 覆盖区间(K_{i-1}, K_i]。由此推导出权重场景的正确参数传origin 0min要落在第一个桶内部例如step / 2max Σw所有权重之和。若origin与min相同取值序列会与桶边界对齐例如取值 0.5/1.5/2.5 正好踩在累计键 1.5/2.5 的边界上从而永远命中第一个节点——这是实际踩过的坑等权[1.0, 1.0]实测 A100/B0见 bug-archived-20261004-2.md。实战参照RPCClient 如何正确调用 roundRobin / randomf_rpc 的 ClientConfig.createClient 是MultiClient的标准用法可作为参数取值模板var sum 0.0f64 for((w, _) in addresses){ sum w } match(loadbalance){ // 随机在 (0, Σw) 上均匀取值落进各桶的概率与权重成正比 case random builder.random(0.0f64, sum) // 轮询step 固定 1.0从「桶中点」0.5 起步起点落在桶边界上会永远归前一个节点 // 累计键原点必须是 0 origin不能与取值起点 0.5 混用 case roundrobin builder.roundRobin(1.0f64, 0.5f64, sum, origin: 0.0f64) case x throw LoadBalanceException(unsupported loadbalance algorithm: ${x}) } builder.add(addresses) builder.build()注意源码注释的结论传给算法的取值区间不是「权重的范围」而是「累计权重的区间」roundRobin的 step 固定为 1.0、起点 0.5桶中点origin: 0.0max sum。整数权重下比例精确。而RPCClient.callR见 RPCClient.cj正是通过for(client in availableClients(message.meta))依次尝试首个迭代元素是首选节点client.transfer(sending).get()失败后tried继续迭代到下一个节点做重试直到超过retryCount。这就是「首元素即选中节点、后续元素为重试节点」语义在真实框架中的落地。ListeningClient订阅 / 推送式收流transferT(message, executor)的返回值它向executor持续投喂该 Message 之后的响应流适合订阅 / 推送式场景。实现上见 ListeningClient.cj它clone()一个独立SocketBuffer以once: false发起 transfer该 id 的 future 不注册进futures后台线程循环Message.decodeT并spawn{executor(m)}。close()幂等AtomicBool.compareAndSwap(false, true)保证只执行一次取消定时器、取消读协程、关闭 buffer。服务端 Serveraccept 循环 内置 PING/ACK 限流执行public class ServerT where T : DataFieldsT { public init(bufferQueueSize!: Int64 1024, bindAt!: UInt16, checkDuration!: Duration, executors!: Int64 200, unavailableChecked!: Int64 3, defaultResp!: (MessageID) - ?Message {_ None}, limiter!: RateLimiterUnit UnlimitedRateLimiterUnit(), checker!: () - ?Message) public static func builder(bufferQueueSize!: Int64 1024, checkDuration!: Duration, checker!: () - ?Message, bindAt!: UInt16): ServerBuilderT /** 阻塞的 accept 循环调用方自己 spawn */ public func start(executor: (SocketBuffer, Message) - ?Message): Unit public func isClosed(): Bool public func close(): Unit }参数说明bindAt是监听端口bufferQueueSize是每连接的写任务队列容量executors是执行业务的线程池大小默认 200checkDuration是连接巡检周期checker生成巡检包unavailableChecked是连接不可用判定阈值默认 3defaultResp是 executor 返回None时的兜底响应limiter是限流器默认UnlimitedRateLimiter。start() 的每连接循环start(executor)是阻塞的 accept 循环调用方自己spawn启动每 accept 一条连接就建一个SocketBuffer并 spawn 连接循环见 server.cj。循环内Message.decodeT(buffer)解出一帧读到 EOF对端干净关闭按 DEBUG 记tcp closed并断链解码失败含消息中途 EOF /CodecException 字节流已错位记 WARNerror occurred on decoding并必须断链不能continuePING 由 Server 内置回 ACKCommand.PING直接buffer.transfer(Message.ack(req.id))心跳 ACKcheckerId里记着的 id即巡检发出去那个 PING 的应答也被识别并从checkerId移除——这两类消息都不会进 executor业务消息走executors.call(limiter){executor(buffer, req)}executor 拿到buffer 已解码的 Message注意不仅是 req还带着 buffer方便直接回写executor 返回None时用defaultResp(req.id)兜底仍为None则不回包连接不可用判定巡检线程Timer.repeat周期执行server.cj对每条连接发checker()包用checkerId记录该连接的 ping 计数超过unavailableChecked默认 3即判不可用并关闭连接buffer.close()不再放回可用队列。巡检队列的消费细节定时回调用buffers.tryRemove()而非阻塞remove()——size只是快照若被并发消费其它 tick 的回调、或close()时抽干队列就会少元素阻塞remove()会把巡检线程永久卡住server.cj 的源码注释记录了该坑。限流只作用于 executor 路径限流只作用于第 3 步executor 业务路径PING/ACK 不受限。ServerBuilder提供四个限流器构造实现来自f_concurrentanyMomentRateLimiter(maxTokens!, timeout!)leakingBucketRateLimiter(timeout!, maxWaitings!, leakingPerDuration!, leakingDuration!)slidingWindowRateLimiter(window!, timeout!, limit!)tokenBucketRateLimiter(tokens!, timeout!, populationPeriod!)ServerBuilder 的双层参数ServerBuilder的参数分两层见 server.cj无前缀bindToDevice/receiveBufferSize/sendBufferSize/reuseAddress/backlogSize/socketOption*/setSocketOptionBool/setSocketOptionIntNative作用于监听 socketTcpServerSocketsocket*前缀socketBindToDevice/socketNoDelay/socketLinger/socketQuickAcknowledge/socketReadTimeout/socketReceiveBufferSize/socketSendBufferSize/socketWriteTimeout经SocketParams作用于accept 之后的每条连接其它executors(n)、defaultResponse(fn)、unavailableChecked(n)与四个限流器构造。从测试看行为保证f_net_tcp_test.cj 是 f_net 的 TCP 集成测试覆盖了读取 / 发送路径三条主线可作为最小的可运行示例测试 1Server bind 裸 socket connect——ServerArrayInt64.builder(bufferQueueSize: 64, checkDuration: ..., checker: pingChecker, bindAt: PORT).reuseAddress(true).build()spawn { server.start({ _, msg msg }) }再用裸TcpSocket验证端口可达测试 2Client.transfer() PING → 服务端内置 ACK——验证 PING/ACK 由Server.start内置逻辑处理、不会走到 executor断言resp.id req.id且command Command.ACK测试 3executor 请求 / 响应 载荷往返——服务端server.start({ _, req Message(Command.RESP(0u8), req.id, req.data) })把载荷原样带回客户端convertArrayInt64(r.data)后逐字节断言[7, 8, 9]往返一致这是「帧 编解码」合起来的行为保证。测试里T用ArrayInt64而非对象类型f_data 已为集合实现DataFields集合走独立的 dataType不需要对象类型创建者注册从而不依赖DefaultCodec的注册表状态。相关文档f_net/doc/审查报告.md2026-05-03 的静态审查快照当时的文件数 / 用例数已与现状不符仅供历史参考原f_net/doc/下的 API 摘录基本类型 / 客户端 / 服务端已删除以 f_net/README_en.md 与源码为准累计权重桶边界问题与整帧写出 / 关闭时序等历史缺陷的完整记录见 bug-archived-on-20261004.md 与 bug-archived-20261004-2.md。小结f_net 以「帧读写 连接生命周期」为职责边界SocketBuffer保证整帧写出与「完整帧 EOF」的关闭时序Client用Message.decodeDataAny中性解码并以 3 次连续失败为阈值做心跳断链重建Server内置 PING/ACK、用unavailableChecked判定连接不可用、限流只作用于 executor 路径MultiClient以累计权重桶实现加权轮询 / 加权随机配合「首元素选中、后续元素重试」的迭代语义成为f_rpc多节点调用与失败重试的基石。掌握这些 API 与参数语义即可在 fountain 应用中直接搭建可靠的 TCP 通信层。赞分享后端Web框架【免费下载链接】fountain一个用于服务器应用开发的综合工具库。 - 零配置文件 - 环境变量和命令行参数配置 - 约定优于配置 - 深刻利用仓颉语言特性 - 只需要开发动态链接库fboot负责加载、初始化并运行 - 覆盖服务器应用开发的各种场景项目地址https://gitcode.com/Cangjie-SIG/fountain点击查看免费下载相关推荐N_m3u8DL-RE 完整使用指南从部署、参数调优到直播录制的工程实践N_m3u8DL RE 完整使用指南从部署、参数调优到直播录制的工程实践 N_m3u8DL RE 是一个跨平台的 MPDDASH/ M3U8HLS/后端Web框架Dgraph客户端负载均衡故障转移与连接池管理Dgraph客户端负载均衡故障转移与连接池管理 在分布式数据库系统中客户端如何高效地与服务端通信、处理节点故障并保持连接稳定性是保障系统高可用的关键。Dg数据库图数据库分布式数据库后端HAProxy四层负载均衡终极指南TCP连接管理与性能优化HAProxy四层负载均衡终极指南TCP连接管理与性能优化 HAProxy作为业界领先的高性能负载均衡器在四层TCP负载均衡领域展现出了卓越的连接管理能力和负载均衡反向代理后端API网关高可用网络上一篇QQ空间历史说说备份指南3步永久保存你的青春记忆下一篇AutoRaise用鼠标悬停彻底改变你的macOS窗口管理体验创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
阅读完成 · 觉得有帮助？
