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. ...

向量数据库概述转载

也许你最近可能听过这样的新闻，某向量数据库的初创公司刚写好 PPT，就获得了几千万的投资，某公司的开源的向量数据库因其代码的简陋而登上了 Hackernews 等等。在过去几个月时间中， AI 应用的发展如火如荼，带动了 AI 应用技术栈上下游的火爆，而向量数据库就是其中最热门的之一。笔者最近因为开发 ChatFiles 和 VectorHub 两款开源项目的需要从而对向量数据库（Vector Database）进行了学习，在对主流的向量数据库和搜索算法有了大概的了解后，笔者决定将这些知识整理成一篇文章，希望能够帮助到大家。 GPT 的缺陷过去几个月的时间，我们正处于人工智能的革命中，其中最耀眼的莫过于 GPT-3.5/4 的横空出世，而 GPT-3.5/4 带给我们无限震撼的同时，其天然的缺陷和诸多的限制也让开发者头痛不已，例如其输入端上下文（tokens）大小的限制困扰着很多的开发者和消费者，像 gpt-3.5-turbo 模型它的限制是 4K tokens(～3000 字)，这意味着使用者最多只能输入 3000 字给 GPT 来理解和推理答案。有人可能会疑惑，我使用的 ChatGPT 是有对话记忆功能的，既然它可以做到聊天记忆，那么它的输入端 token 有限制也没什么关系，只要我将给 ChatGPT 的文字内容拆分成多次输入，它自然就可以记住我之前的对话，从而做到解除 token 限制。这个想法是不太正确的，GPT 作为 LLM 模型是没有记忆功能的，所谓的记忆功能只是开发者将对话记录存储在内存或者数据库中，当你发送消息给 gpt 模型时，程序会自动将最近的几次对话记录（基于对话的字数限制在 4096 tokens 内）通过 prompt 组合成最终的问题，并发送给 ChatGPT。简而言之，如果你的对话记忆超过了 4096 tokens，那么它就会忘记之前的对话，这就是目前 GPT 在需求比较复杂的任务中无法克服的缺陷。目前，不同模型对于 token 的限制也不同，gpt-4 是 32K tokens 的限制，而目前最大的 token 限制是 Claude 模型的 100K，这意味可以输入大约 75000 字的上下文给 GPT，这也意味着 GPT 直接理解一部《哈利波特》的所有内容并回答相关问题。但这样就能解决我们所有的问题了吗？答案是否定的，首先 Claude 给出的例子是 GPT 处理 72K tokens 上下文的响应速度是 22 秒。如果我们拥有 GB 级别或更大的文档需要进行 GPT 理解和问答，目前的算力很难带来良好体验，更关键的是目前 GPT API 的价格是按照 tokens 来收费的，所以输入的上下文越多，其价格越按昂贵。 ...

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去处理这个连接上的读写事件： ...