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.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介绍 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去处理这个连接上的读写事件： ...
短链接系统实现 如何保证生成短链接不重复 如何存储短链接 用302（临时）还是301（永久）重定向 https://cloud.tencent.com/developer/article/1858351 https://blog.csdn.net/codejas/article/details/106102452 https://juejin.cn/post/7312353213348741132 秒杀 使用redis（保证秒杀效率）的lua脚本（保证原子性）进行库存扣减，使用分布式事务的二阶段消息解决事务数据一致性。二阶段消息适用于无需回滚的这一类数据一致性问题，主要是为了保证第一阶段操作执行成功后，后续阶段一定能感知并执行。 二阶段消息的回查操作，主要还是依赖事务中第一阶段使用的数据库，来保证第一阶段整体操作的原子性以及幂等。 无论是请求执行lua脚本的服务端宕机，还是redis服务本身宕机，lua脚本都保证原子性，即写操作均无效 库存扣减lua脚本： -- KEYS[1] 库存 -- KEYS[2] 事务当前操作 -- KEYS[3] 如果事务当前操作是回滚操作，则为回滚所对应的操作 local v = redis.call('GET', KEYS[1]) -- 库存 local e1 = redis.call('GET', KEYS[2]) -- 事务当前操作的状态 if v == false or v + ARGV[1] < 0 then -- 库存不足 return 'FAILURE' end if e1 ~= false then -- 当前状态不为空，幂等退出 return 'DUPLICATE' end -- 设置当前操作为已完成 redis.call('SET', KEYS[2], 'op', 'EX', ARGV[3]) if ARGV[2] ~= '' then local e2 = redis.call('GET', KEYS[3]) if e2 == false then -- 如果是回滚操作，将回滚对应操作的状态设置为已回滚 redis.call('SET', KEYS[3], 'rollback', 'EX', ARGV[3]) return end end -- 库存扣减 redis.call('INCRBY', KEYS[1], ARGV[1]) 回查lua脚本： local v = redis.call('GET', KEYS[1]) -- 扣减库存操作的状态 if v == false then -- 为空则直接回滚 redis.call('SET', KEYS[1], 'rollback', 'EX', ARGV[1]) v = 'rollback' end -- 如果阶段1是回滚，直接返回事务失败 if v == 'rollback' then return 'FAILURE' end -- 如果不是回滚，说明事务成功 以下是可能出现的各个场景。 ...