后端RPC框架【免费下载链接】finagleA fault tolerant, protocol-agnostic RPC system项目地址https://gitcode.com/gh_mirrors/fi/finagle点击查看免费下载Finagle 是一个容错、协议无关的 RPC 系统其服务端Server承担着监听端口、接收请求并快速分发的职责。本文以仓库中的 Servers.rst 为骨架结合 finagle-core 的源码实现系统讲解 Finagle 服务端的核心接口、默认模块栈可观测性、并发限制、请求拒绝、超时、会话过期及每个模块的底层原理、配置方法与对应指标帮助你从会调用进阶到理解其为何如此设计。服务端核心接口Server[Req, Rep]与ListeningServerFinagle 服务端实现了一个非常简单的接口定义于 finagle-core/src/main/scala/com/twitter/finagle/Server.scaladef serve( addr: SocketAddress, factory: ServiceFactory[Req, Rep] ): ListeningServer给定一个SocketAddress监听地址和一个ServiceFactory[Req, Rep]服务工厂服务端返回一个ListeningServer。ListeningServer承担服务端资源的管理职责通过boundAddress暴露实际绑定的地址当以:*绑定临时端口时可用它拿到真实端口实现ClosableOnce与Awaitable[Unit]关闭一个服务端实例会解绑端口并释放相关资源见 Server.scala 的closeOnce逻辑它同时关闭服务端与所有 announce 注册提供announce(addr: String)向服务发现系统如 ZooKeeper发布自身地址关闭时自动移除。接口同时提供了直接服务简单Service的变体serve(addr, service)内部通过ServiceFactory.const(service)把单个Service包装成ServiceFactory。而ServiceFactory变体用于连接状态有意义的协议——每条新连接会从ServiceFactory中获取一个新的Service请求在该连接上被分发到该服务客户端断开或连接终止时服务也随之关闭Server.scala。典型用法是Protocol.serve(...)形式例如以Http.server起一个 HTTP 服务import com.twitter.finagle.Service import com.twitter.finagle.Http import com.twitter.finagle.http.{Request, Response} import com.twitter.util.{Await, Future} val service: Service[Request, Response] new Service[Request, Response] { def apply(req: Request): Future[Response] Future.value(Response()) } val server Http.server.serve(:8080, service) Await.ready(server) // 阻塞等待直到服务端资源被释放serve的地址既可以是SocketAddress也可以是字符串如:8080、localhost:8080字符串形式会经ServerRegistry.register(addr)解析注册Server.scala。此外接口还提供了serveAndAnnounce(name, addr, service)系列变体绑定监听的同时把地址 announce 到命名服务。注意finagle-thrift服务端因其接口由 Thrift IDL 定义而暴露出更丰富的 API如ThriftMux.serve会生成按方法分发的服务详见 Protocols.rst 中关于 Thrift 与 Scrooge 的说明。服务端模块Server Modules总览Finagle 服务端的哲学是简单、快速因此服务端只被装配上最小限度的附加行为更复杂的逻辑重试、负载均衡等放在客户端侧。默认的 Finagle 服务端由一系列栈式模块Stack Module组成请求从左到右依次流经各模块。每个模块的视觉化布局可参考文档仓库中的 serverstack.svg。许多服务端模块扮演**准入控制器admission controller**的角色基于动态或静态属性决定当前服务端能否在维持某个 SLOService Level Objective服务水平目标的前提下处理到来的请求。术语约定下文中的Endpoint模块即用户通过Server.serve(...)传入的自定义Service它是请求处理的最终落点。可观测性Observability一个 Finagle 服务端自带若干用于观测与调试的模块监控Monitoring由 MonitorFilter.scala 实现。用户自定义Service中所有未被捕获的异常都会流经该服务端所使用的Monitor对象。MonitorFilter.apply在调用下游服务期间临时把Monitor设置为配置的实例无论是service(request)返回的Future失败respond(RespondFn)还是同步抛出的NonFatal异常都会被monitor.handle(exc)捕获处理MonitorFilter.scala。默认实例只是把异常日志输出到标准输出可通过Server.Monitor参数覆盖该行为。链路追踪Tracing由 TraceInitializerFilter.scala 实现。服务端的模块通过serverModule(newId false)装配每个请求都会把配置的Tracer压入追踪栈服务端不生成新的 TraceId区别于客户端的clientModule会设置新 TraceIdTraceInitializerFilter.scala。统计Stats由 StatsFilter.scala 实现向给定的StatsReceiver上报请求数、成功数、延迟等指标。核心逻辑在apply中记录开始时间、递增未完成请求计数并在响应返回时计算耗时、记录各类指标StatsFilter.scala。StatsFilter 还会依据响应分类器见下节把非异常形式的应用层失败记为com.twitter.finagle.service.ResponseClassificationSyntheticException归入failures/统计。响应分类Response Classification响应分类是服务端可观测性得以准确的前提。通过提供响应分类器ResponseClassifier开发者可以把应用层的成功/失败语义告诉 Finagle从而提升失败驱逐failure accrual的有效性并使成功率统计更准确详见 shared-modules/ResponseClassification.rst。对 HTTP 服务端常用HttpResponseClassifier.ServerErrorsAsFailures它把任意 HTTP 5xx 状态码归类为失败import com.twitter.finagle.Http import com.twitter.finagle.http.service.HttpResponseClassifier Http.server ... .withResponseClassifier(HttpResponseClassifier.ServerErrorsAsFailures)对 Thrift/ThriftMux 服务端常用ThriftResponseClassifier.ThriftExceptionsAsFailures把反序列化出的 Thrift 异常归类为失败。若用户未指定分类器或用户的分类器对某个请求/响应对未定义则回落到ResponseClassifier.Default其规则简单Returns视为成功、Throws视为失败。ResponseClassifier本质是PartialFunction[ReqRep, ResponseClass]。ReqRep是请求-响应对使分类器可以同时基于请求与响应做判断。ResponseClass定义了四类结果Success操作成功NonRetryableFailure出错了不要重试RetryableFailure出错了可以考虑重试Ignorable出错了但可以忽略注意Ignorable只适用于FailureFlags.Ignorable标记的Failure并不适用于任意响应。需要强调的是分类器只被咨询而不会被强制服从——分类器返回RetryableFailure不代表调用方一定会重试返回Ignorable也不代表调用方一定会忽略。自定义分类器时PartialFunction不必是全函数因为 Finagle 总是把用户分类器与覆盖所有情况的ResponseClassifier.Default组合使用。Thrift/ThriftMux 的分类器需要更多注意由于 IDL 服务的所有方法共用同一个Array[Byte] Array[Byte]的ServiceScrooge 与newService/serve等代码会先把响应反序列化为预期的应用类型使分类器可以针对 Scrooge 生成的$Service.$Method.Args请求类型与方法响应类型编写。以下是一个针对 IDLSocialGraph.follow的示例完整 IDL 见 ResponseClassification.rstimport com.twitter.finagle.service.{ReqRep, ResponseClass, ResponseClassifier} val classifier: ResponseClassifier { // #1 反序列化的 NotFoundException 视为不可重试失败 case ReqRep(_, Throw(_: NotFoundException)) ResponseClass.NonRetryableFailure // #2 检查成功响应内容如状态码语义 case ReqRep(_, Return(x: Int)) if x 0 ResponseClass.NonRetryableFailure // #3 注意基于请求内容的判断在 Thrift/ThriftMux 下应避免 case ReqRep(SocialGraph.Follow.Args(a, b), _) if a 0 ResponseClass.NonRetryableFailure // #4 反序列化的 InvalidQueryException 可视为成功 case ReqRep(_, Throw(_: InvalidQueryException)) ResponseClass.Success }其中 #3 需要特别谨慎如果异常发生在 Mux 层例如ClientDiscardedRequestException请求内容并不会被反序列化也就不会命中该模式。对 Thrift/ThriftMux 而言应优先在应用层例如通过Filter拒绝请求处理请求相关的细节而把响应分类保留给传输层面的响应判定。并发限制Concurrency LimitConcurrency Limit模块维护服务端的并发度由两个过滤器共同实现RequestSemaphoreFilter.scala基于AsyncSemaphore实现。请求无法获取信号量许可时立即以Failure.rejected(...)失败该 Failure 标记了Rejected与Retryable向调用方暗示操作可安全重试并记录request_concurrency当前并发请求数与request_queue_size因限制而等待的请求数两个 gaugeRequestSemaphoreFilter.scala。PendingRequestFilter.scala基于AtomicInteger的简单计数实现适合maxWaiters 0的场景省去AsyncSemaphore等待队列的开销。它用 CAS 循环把当前处理中请求数与上限比较超限则递增rejected计数并返回Failure.rejected(PendingRequestsLimitExceeded)PendingRequestFilter.scala。其测试 PendingRequestFilterTest.scala 验证了超限请求以可重试 Failure 被拒绝且后续请求在占用释放后得以放行的行为。默认该模块是关闭的即 Finagle 服务端的请求并发度无上限。启用并限制最大并发请求数的示例如下示例中的参数值仅为演示 API 用法并非生产建议值import com.twitter.finagle.Http val server Http.server .withAdmissionControl.concurrencyLimit( maxConcurrentRequests 10, maxWaiters 0 ) .serve(:8080, service)Concurrency Limit模块有两个配置参数maxConcurrentRequests—— 允许同时处理的请求数maxWaiters—— 在maxConcurrentRequests之上允许排队的请求数。该参数的值决定实际使用哪个过滤器maxWaiters为 0 时使用PendingRequestFilter省去AsyncSemaphore等待队列的开销否则使用RequestSemaphoreFilter。concurrencyLimit的实现位于 ServerAdmissionControlParams.scala入口withAdmissionControl由 WithServerAdmissionControl.scala 提供def concurrencyLimit(maxConcurrentRequests: Int, maxWaiters: Int): A if (maxWaiters 0) { // maxWaiters 为 0 时无需 AsyncSemaphore 的队列开销 // 使用 PendingRequestFilter 可获得更好性能 self.configured(PendingRequestFilter.Param(Some(maxConcurrentRequests))) } else { val semaphore if (maxConcurrentRequests Int.MaxValue) None else Some(new AsyncSemaphore(maxConcurrentRequests, maxWaiters)) self.configured(RequestSemaphoreFilter.Param(semaphore)) }所有超出(maxConcurrentRequests maxWaiters)的请求都会被服务端拒绝rejected因此该模块是一个静态准入控制器它监测的是当前请求的并发水位。相关指标见 Public.rstrequest_concurrency当前并发请求数gaugerequest_queue_size因并发限制而等待的请求数gauge。另外withAdmissionControl还提供deadlines/darkModeDeadlines/noDeadlines三兄弟用于启用或禁用DeadlineFilter默认Disabled基于请求携带的截止时间Deadline做准入控制ServerAdmissionControlParams.scala。拒绝请求Rejecting Requests服务端可以按具体请求逐案显式拒绝。要产生一个拒绝响应让服务返回一个设置了Rejected标志的Failure即可Failure.rejected是为此提供的便捷方法import com.twitter.finagle.Failure val rejection Future.exception(Failure.rejected(busy)) val nonRetryable Future.exception(Failure(Dont try again, Failure.Rejected|Failure.NonRetryable))这两个响应都会被 Finagle 视为被拒绝rejected。其语义在 Failure.scala 中定义Failure.rejected(why)创建同时带有Retryable | Rejected标志的Failure——Retryable标志表明该失败安全可重试客户端的 RequeueFilter.scala 会自动重试这类可重试失败RequeueFilter.Requeueable提取器基于RetryPolicy.RetryableWriteException判定见 RequeueFilter.scala重试受RetryBudget限速对于不应重试的请求服务端应返回设置了NonRetryable标志的Failure如上面的nonRetryable示例。关于 nack取决于协议被拒绝的请求可能被转换成 nack 消息目前 HTTP/1.1 与 Mux 支持参见 Glossary.rst 中nack词条。被拒绝的请求同样计入request_classification/requeue等分类指标Public.rst。请求超时Request TimeoutRequest Timeout模块由 TimeoutFilter.scala 实现其职责简单直接让fails所有在给定时间内未能处理完的请求。与客户端一样该模块默认关闭即超时无上限。TimeoutFilter的核心实现在apply与applyTimeout中TimeoutFilter.scala取出当前超时值支持Tunable[Duration]动态调整构造截止时间Deadline.ofTimeout必要时把 Deadline 通过Contexts.broadcast传播给下游applyTimeout用res.within(timer, timeout, internalTimeoutEx)给下游Future加上时限超时则抛出RequestTimeoutException默认IndividualRequestTimeoutException并res.raise(timeoutEx)取消仍在进行的下游处理。覆盖默认超时行为的配置示例如下源码TimeoutFilter构造参数包括timeoutFn、exceptionFn、timer以及propagateDeadlines、preferDeadlineOverTimeout等见 TimeoutFilter.scalaimport com.twitter.conversions.DurationOps._ import com.twitter.finagle.Http val server Http.server .withRequestTimeout(10.seconds) .serve(:8080, service)重要语义Request Timeout模块是fail失败而不是reject拒绝请求。这意味着它默认不会被远端客户端重试——因为无法确定请求是超时于队列等待阶段还是处理阶段重试可能导致重复执行。会话过期Session Expiration某些场景下服务端希望通过对会话session/连接的生命周期设限来约束自身资源。Session Expiration模块挂载在连接级别在空闲超过一定时长后过期expire一个 service/session。它由 ExpiringService.scala 实现服务端模块ExpiringService.server是一个Stack.Module3[Param, param.Timer, param.Stats, ...]当idleTime/lifeTime均为有限值时把连接上的每个Service包装为ExpiringService过期时执行onExpire()即关闭连接ExpiringService.scala过期机制内部startTimer用timer.schedule(t.fromNow) { expire(counter) }分别调度空闲定时器与寿命定时器请求到来时暂停空闲定时器idleTask.cancel()请求完成且无其他在途请求时重新启动空闲定时器ExpiringService.scala过期动作受AsyncLatch保护确保在途请求结束后才真正执行过期latch.await { expired(); counter.incr() }。默认配置是永不过期Param默认值为Duration.Top见 ExpiringService.scala。配置示例参数值仅为演示 API 用法import com.twitter.conversions.DurationOps._ import com.twitter.finagle.Http val twitter Http.server .withSession.maxLifeTime(20.seconds) .withSession.maxIdleTime(10.seconds) .newService(twitter.com)Expiration模块有两个参数maxLifeTime—— 会话被视为存活的最大时长maxIdleTime—— 会话允许空闲不发送任何请求的最大时长。对应指标定义于 metrics/IdleApoptosis.rstidle因两次请求之间空闲过久而过期expire的次数counterlifetime超过最大存活时长而过期expire的次数counter。服务端常用指标速查服务端各模块的指标在 metrics/Public.rst 中集中定义这里摘取与服务端强相关的一组StatsFilterPublic.rstrequests成功失败总数、success成功数、request_latency_ms延迟直方图、pending当前未完成请求数的 gauge、failures/exception_name各异常计数。若使用把非异常响应分类为失败的ResponseClassifier会以合成异常com.twitter.finagle.service.ResponseClassificationSyntheticException计入failures/。ServerStatsFilterPublic.rsthandletime_us建立 Future 链的耗时大值暗示 Finagle 线程上存在阻塞代码、transit_latency_ms跨跳传输时间、request_classification/total|retry|requeue|backup请求分类计数。RequestSemaphoreFilter / 并发限制Public.rstrequest_concurrency、request_queue_size含义见上文并发限制节。ExpiringServiceIdleApoptosis.rstidle、lifetime含义见上文会话过期节。小结Finagle 服务端的设计刻意保持最小化默认只装配可观测性监控、追踪、统计、响应分类、并发限制、请求拒绝、请求超时与会话过期等少数模块把重试、负载均衡等复杂行为留给客户端。理解每个模块是拒绝还是失败是否可重试默认开关与参数含义是正确配置生产级 Finagle 服务端的关键。本文涉及的源码与文档路径服务端接口finagle-core/src/main/scala/com/twitter/finagle/Server.scala服务端文档doc/src/sphinx/Servers.rst响应分类doc/src/sphinx/shared-modules/ResponseClassification.rst并发限制实现PendingRequestFilter.scala、RequestSemaphoreFilter.scala、ServerAdmissionControlParams.scala会话过期实现ExpiringService.scala指标文档doc/src/sphinx/metrics/Public.rst、doc/src/sphinx/metrics/IdleApoptosis.rst赞分享后端RPC框架【免费下载链接】finagleA fault tolerant, protocol-agnostic RPC system项目地址https://gitcode.com/gh_mirrors/fi/finagle点击查看免费下载相关推荐Path of Building PoE2免费开源构建模拟器让你成为流放之路2的构建大师Path of Building PoE2免费开源构建模拟器让你成为流放之路2的构建大师 你是否曾在流放之路2中花费数小时调整技能和装备却依然无法达到理想桌面应用游戏开发MeterSphere开发者手册后端服务模块代码结构解析MeterSphere开发者手册后端服务模块代码结构解析 1. 后端整体架构概览 MeterSphere后端采用模块化架构设计基于Spring Boot微服质量保障接口测试测试后端前端AI 应用DevOpsFinagle OpenCensus Tracing 模块解析让 Finagle 客户端与服务端无缝接入 OpenCensus 分布式追踪Finagle OpenCensus Tracing 模块解析让 Finagle 客户端与服务端无缝接入 OpenCensus 分布式追踪 本文围绕 fina后端RPC框架上一篇SystemInformer多语言界面配置终极国际化支持指南下一篇告别回调地狱ReactiveKit打造Swift响应式编程新范式创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考