Posts

gRPC阅读（3）—— 服务发现

服务发现概述平时用浏览器上过网都知道，输入一个网址比如google.com就能访问内容，背后是DNS帮我们将google.com解析成IP地址，最终浏览器才能基于TCP协议，从本地连接到这个服务提供商的IP地址。所以DNS属于服务发现的其中一种方式。所以服务发现提供的就是通过自动化的方式帮助服务在网络中找到彼此，无需手动配置。一个好的服务发现需要：服务地址动态变化：服务的 IP 或端口可能因为容器化或自动扩展而频繁改变。高可用：需要在服务实例宕机时快速感知并移除不健康的实例。负载均衡：服务发现需要为调用方提供负载均衡能力，选择最佳的服务实例。服务发现通常与负载均衡同时实现，分为两种方式：客户端服务发现（如eureka、consul）：在客户端做负载均衡，选择一个实例进行调用，优点是避免集中式LB可能存在的瓶颈，性能较好，但是每个客户端需要维护服务端列表，服务端这部分的负载可能变高。并且更新LB或其他相关组件的策略时需要所有客户端都一起更新，管理不方便。并且需要多语言支持代理服务发现（如k8s+coreDNS、nginx+consul）：客户端将请求发送到负载均衡器（如 API 网关），由负载均衡器查询服务注册中心并将请求转发给目标服务实例。独立LB进程：LB与消费者在同一个主机中，但分别作为不同的进程，避免了需要多语言支持，以及LB的更新不需要调用方改代码。服务发现的核心组件有：注册中心、服务提供者、客户端（服务消费者）服务发现的关键功能有：服务注册、服务查询、健康检查、动态更新 gRPC服务发现 gRPC使用客户端服务发现，gRPC中称为名称解析（Name Resolution），默认情况下使用DNS-resolver。通过服务发现解析出IP列表后就通过LB组件进行负载均衡并建立连接。下面基于target=localhost:50052这个服务端地址来进行分析，并且是默认的DNS作为resolver（不用官方例子的50051端口是因为被mac的launchd进程占用了）。首先gRPC在创建cc（ClientConn）的时候，使用initParsedTargetAndResolverBuilder创建resolver.Builder。这一步决定的是采用什么服务发现机制，默认是DNS。 func (cc *ClientConn) initParsedTargetAndResolverBuilder() error { logger.Infof("original dial target is: %q", cc.target) // 尝试直接解析target并获取相应的resolver.Builder var rb resolver.Builder parsedTarget, err := parseTarget(cc.target) if err == nil { rb = cc.getResolver(parsedTarget.URL.Scheme) if rb != nil { cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } } // target没有指定schema（比如我们的localhost:50052是没有指定schema的）或者无法匹配schema对应的resolver.Builder // 那么使用默认的schema，即dns defScheme := cc.dopts.defaultScheme if internal.UserSetDefaultScheme { defScheme = resolver.GetDefaultScheme() } // 此处canonicalTarget为dns:///localhost:50052 // "//"与第三个"/"之间的是authority canonicalTarget := defScheme + ":///" + cc.target // 再次尝试target并获取相应的resolver.Builder，此处会拿到dns.dnsBuilder parsedTarget, err = parseTarget(canonicalTarget) if err != nil { return err } rb = cc.getResolver(parsedTarget.URL.Scheme) if rb == nil { return fmt.Errorf("could not get resolver for default scheme: %q", parsedTarget.URL.Scheme) } // 保存parsedTarget和resolverBuilder cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } 那么resolverBuilder在什么时候会Build一个resolver出来呢？在ide的帮助下，可以直接定位到这个函数中： ...

请不要再称数据库是CP或者AP (Please stop calling databases CP or AP)

Note 文章转载自https://blog.the-pans.com/cap/ 其它参考： quorum-rw 后分布式时代: 多数派读写的’少数派’实现经Martin Kleppman本人同意，这篇文章是他英文原文的中文翻译。Authorized by Martin Kleppmann, this is a Chinese translation of his original blog post. ...

向量数据库概述

Note 文章转载自https://guangzhengli.com/blog/zh/vector-database/ 也许你最近可能听过这样的新闻，某向量数据库的初创公司刚写好 PPT，就获得了几千万的投资，某公司的开源的向量数据库因其代码的简陋而登上了 Hackernews 等等。在过去几个月时间中， AI 应用的发展如火如荼，带动了 AI 应用技术栈上下游的火爆，而向量数据库就是其中最热门的之一。 ...

gRPC阅读（2）—— 客户端

启动客户端客户端的启动也是三部曲：初始化grpc.ClientConn 创建service对应的Client（比如codegen生成的GreeterClient）发起rpc调用第二步比较简单，只是把ClientConn作为GreeterClient的成员变量，重点分析建立连接和RPC调用初始化ClientConn 初始化ClientConn做了很多准备工作，包括但不限于：应用选项（DialOption）构建拦截器调用链（Interceptor）决定使用什么resolver（resolver.Builder）检查传输层凭证，比如TLS（TransportCredentials）解析自定义服务端配置（ServerConfig） … 但还有一些配置是在真正发起RPC调用的时候才会被设置和触发，比如重试限流器、RPC配置选择器、RPC负载均衡器等。 func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error) { cc := &ClientConn{ target: target, conns: make(map[*addrConn]struct{}), dopts: defaultDialOptions(), } // 重试限流器 cc.retryThrottler.Store((*retryThrottler)(nil)) // 配置选择器，动态选择每个RPC的调用配置 cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil}) cc.ctx, cc.cancel = context.WithCancel(context.Background()) // options ... // 确定使用哪个resolver（默认为dns） if err := cc.initParsedTargetAndResolverBuilder(); err != nil { return nil, err } // 内部使用的全局perTarget options for _, opt := range globalPerTargetDialOptions { opt.DialOptionForTarget(cc.parsedTarget.URL).apply(&cc.dopts) } // 初始化拦截器调用链 chainUnaryClientInterceptors(cc) chainStreamClientInterceptors(cc) // 验证安全传输，如TLS if err := cc.validateTransportCredentials(); err != nil { return nil, err } // 解析以json格式指定的配置 // 如负载均衡配置、per-RPC方法超时等 if cc.dopts.defaultServiceConfigRawJSON != nil { scpr := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON, cc.dopts.maxCallAttempts) if scpr.Err != nil { return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, scpr.Err) } cc.dopts.defaultServiceConfig, _ = scpr.Config.(*ServiceConfig) } // keepalive对服务端探活 cc.mkp = cc.dopts.copts.KeepaliveParams // 获取authority，作为请求头中的:authority字段 if err = cc.initAuthority(); err != nil { return nil, err } // 注册channelz，用于监测grpc的运行 // 可通过http协议访问/grpc/channelz/v1查看grpc的状态 cc.channelzRegistration(target) channelz.Infof(logger, cc.channelz, "parsed dial target is: %#v", cc.parsedTarget) channelz.Infof(logger, cc.channelz, "Channel authority set to %q", cc.authority) // 连接状态管理器 cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz) // 负载均衡器，动态选择每个RPC的子通道 cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers) // stats cc.metricsRecorderList = stats.NewMetricsRecorderList(cc.dopts.copts.StatsHandlers) cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc. // idle状态管理 cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout) return cc, nil } 这么一套下来可以看到，初始化ClientConn的时候并没有建立连接，所以猜测是在第一次发起RPC调用的时候才去尝试建立连接。还有一种验证方法是，把服务端关闭，尝试NewClient，是不会返回错误的。 ...

gRPC阅读（1）—— 服务端

gRPC介绍 gRPC 是一种由 Google 开发的高性能远程过程调用（RPC）框架，适用于分布式系统间的通信。它基于 HTTP/2 进行传输，使用 Protocol Buffers 进行序列化，提供跨平台的兼容性。gRPC 的核心理念是让客户端像调用本地函数一样调用远程服务，简化服务间的调用流程。通过编写与具体编程语言无关的 IDL （默认是 protobuf）来定义 RPC 方法，gRPC 框架就会生成语言相关的客户端/服务端代码。 HTTP/2介绍相比http1，具有更高的传输效率（多路复用：在同一个链连接上同时处理多个请求），更低的延迟（服务端推送，减少请求数量、简化header大小）、带宽利用率更高（头部压缩、数据流优先）、更安全（基于tls）。 http2具体特性有：帧、消息、流：帧是小通信数据单元；消息由一个或多个帧组成。例如请求的消息和响应的消息；一个连接中包含多个流，每个流包含多个帧。帧通过流id进行标识属于哪个流二进制分帧：每个消息由若干个帧组成，帧是最小传输单位，并且原来基于文本编码变成基于二进制，进一步减小帧大小压缩header 多路复用：即在同一连接中的多个stream的传输互不影响服务端推送流量控制和资源优先级：流量控制以有效利用多路复用机制，确保只有接收者使用的数据会被传输。优先级机制可以确保重要的资源被优先传输。启动服务端通过官方的 helloworld 例子可以看到，服务端的启动分为三步：创建gRPC的Server 将业务handler注册到Server 调用Server.Serve在端口上进行监听第一步没什么好说的，注意下第二步注册进去的东西： // 注册进去的ServiceDesc var Greeter_ServiceDesc = grpc.ServiceDesc{ ServiceName: "helloworld.Greeter", HandlerType: (*GreeterServer)(nil), Methods: []grpc.MethodDesc{ { MethodName: "SayHello", Handler: _Greeter_SayHello_Handler, }, }, Streams: []grpc.StreamDesc{}, Metadata: "helloworld/helloworld.proto", } // Method对应的handler func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(HelloRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(GreeterServer).SayHello(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: Greeter_SayHello_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest)) } return interceptor(ctx, in, info, handler) } 前两步比较简单，再来看第三步的如何建立连接并进行处理。类似标准库http的ListenAndServe，本质上就是创建一个死循环等待有新的连接到来，然后开新的goroutine去处理这个连接上的读写事件： ...