CI / CD 全链路：从个人项目到企业级的完整拆解
完整的CI/CD流水线，会在你掌握的核心主干基础上，补充大量前置门禁、安全合规、发布管控、故障兜底的环节，形成全流程自动化的安全闭环，下面给你按流水线顺序，完整拆解清楚。一、最小可用CI/CD完整步骤这是CI/CD的灵魂，不管流程怎么扩展，核心永远不变，完全可以直接用于个人项目、小型业务：→ 代码提交 → CI（合并主干 → 代码lint → 单元/集成测试 → 代码验证通过） → 持续交付（生产级标准构建 → 预发布环境全量校验 → 版本化推送到Harbor制品库 → 待上线就绪） → 持续部署（持续交付全流程 + 自动审批 + 自动部署到生产服务器）二、完整企业级CI/CD全链路步骤以下是中大型企业核心业务的标准流水线，按执行顺序拆解，补充的都是你之前没提到、但生产环境必做的环节：【前置阶段：代码提交前（CI的隐形起点，提前拦截80%的低级问题）】本地pre-commit钩子：开发者提交代码前，自动在本地跑lint、单元测试、简单格式校验，有问题直接拦在本地，不用推到仓库才发现注意: 当前功能在jetbrain家的产品中没有显式标明, 实际在提交时会自动执行. 但可以通过ide自带的lint和预执行单元测试命令实现, 比写git脚本更简单, 方便, 直接.代码评审（CR）：代码提交后，必须经过团队成员评审通过，才能触发CI流水线，避免单人编写的风险代码进入流程【CI阶段：代码集成与基线验证（核心主干不变，补充企业级门禁）】核心主干：拉取主干代码 → 冲突检测 → 语法lint → 单元测试 → 模块集成测试补充必做环节（全是门禁，不通过直接终止流水线，绝不进入下一环节）：代码质量门禁：扫描代码重复率、圈复杂度、潜在bug、安全坏味道，不达标直接拦截第三方依赖安全扫描（SCA）：检测项目用的开源组件有没有已知高危漏洞（比如log4j漏洞），有风险直接拦截构建缓存优化：缓存不变的依赖包，避免每次构建重复下载，大幅提升流水线速度阶段终点：得到经过验证、可用于生产构建的干净源代码【持续交付阶段：制品标准化与上线就绪】核心主干：生产级构建 → 预发布环境验证 → 推送到Harbor制品库补充必做环节（彻底解决环境差异、安全风险、合规问题，也是你之前疑惑的“持续交付到底干了什么”的核心补充）：严格执行「一次构建原则」：全程只构建一次，这个制品会贯穿预发布测试、入库、生产部署全流程，绝对不会重新构建，从根源避免“测试环境能跑，生产环境跑不起来”的问题制品签名与防篡改校验：给构建好的Docker镜像/程序包打上唯一数字签名，后续所有环节都会校验签名，确保制品没被篡改，守住供应链安全制品二次安全扫描：推入库前，再次扫描制品本身的漏洞（比如镜像里的系统漏洞、恶意文件），有问题直接拦截生产级全量验证：把制品部署到和生产环境1:1的预发布环境，跑端到端全链路测试、性能压测、兼容性测试、安全渗透测试，确保制品在生产环境100%可用版本全生命周期管理：给制品打上不可修改的唯一版本号，和代码提交记录、测试报告、构建日志强绑定，实现任何版本可追溯、出问题可秒级回滚合规审批与审计：自动触发上线审批流（负责人、安全团队审批），全流程所有操作留存审计日志，满足企业合规要求阶段终点：制品完成所有上线前准备，锁在制品库里，随时可以一键安全上线到生产环境，无需任何额外加工【持续部署阶段：自动化上线与故障兜底（核心逻辑不变，补充生产级管控）】核心主干（你已完全掌握）：完整持续交付全流程 + 自动部署到生产服务器补充必做环节（避免自动部署变成“自动故障”，是企业级持续部署的核心）：灰度/金丝雀发布：不直接全量上线，先把1%-5%的流量切到新版本，观察无异常再逐步全量，把故障影响范围降到最低无损上下线与健康检查：先启动新版本实例，自动校验服务健康状态（能否正常响应请求），健康后再逐步切流量，等老版本实例的所有请求处理完再下线，完全不影响用户线上自动化验收：上线后自动跑核心链路冒烟测试，确保线上核心功能正常监控联动与自动回滚：实时对接监控系统，观察新版本的错误率、响应时间、资源占用等核心指标，一旦发现异常，自动触发回滚，瞬间切回上一个稳定版本，无需人工干预上线结果通知：不管成功失败，自动把上线结果、日志、核心指标推送给相关开发、运维人员【后置闭环阶段：全流程归档与持续优化】全流程日志、测试报告、部署记录、审计日志归档留存，用于后续问题排查与合规审计收集本次上线的监控数据、用户反馈，持续优化流水线的效率、拦截率与稳定性关键补充说明没有绝对的“所有步骤”：CI/CD流水线是可定制的，个人小项目、非核心业务，用你梳理的核心主干就完全够用；金融、政务、大型互联网的核心业务，必须把补充环节落地，才能满足安全、合规、高可用的要求。核心本质永远不变：不管流程怎么扩展，CI/CD的核心始终是把代码交付的全流程自动化，提前拦截问题，让上线更快、更稳、更安全，减少人工操作带来的风险。注：文档内容由 AI 生成, 感谢豆包AI. 后续我会在实践后手动修改文档
在写自己项目和公司项目里面都没实际编写过单元测试, 即使要也是使用AI自动编写, 说实在的, AI编写单元测试真是一把好手. 这个文章只是梳理一下go单元测试的各种库, 以便后续需要手动编写时候能快速找到对应的库.基础测试用例（Unit Test）核心：通过编写原生代码调用目标函数，验证返回结果是否与预期一致、执行是否报错。Go 原生支持：使用 testing 包，函数名以 Test 开头，通过 t.Error/t.Errorf 标记失败。Fuzz 测试（模糊测试）核心：测试鲁棒性（Robustness），通过自动生成随机 / 边界输入（如异常字符串、超大整数），发现函数可能的崩溃、 panic 或逻辑漏洞。Go 原生支持：1.18+ 版本内置，函数名以 Fuzz 开头。go-sqlmock核心：模拟数据库交互，无需真实数据库连接。验证执行的 SQL 语句（如查询 / 更新语法、参数）和预期响应（如模拟返回的行数据、影响行数）是否匹配。用途：隔离数据库依赖，快速测试数据访问层代码。gomonkey核心：函数 / 方法打桩（Stub）工具，可在运行时替换函数实现（包括全局函数、成员方法，全局变量）。用途：隔离外部依赖（如替换网络调用、文件 I/O），让测试不依赖真实环境。Ginkgo核心：BDD（行为驱动开发）风格的测试框架，提供更语义化的测试组织方式（如 Describe/Context/It 结构），搭配 Gomega 断言库使用。优势：测试用例更易读、易维护，适合复杂业务逻辑的测试组织。Goconvey同上gomock核心：Go 官方维护的 Mock 框架，根据 interface 自动生成 Mock 实现类。可在测试中设置预期的方法调用（如参数匹配、调用次数），并返回指定结果。用途：解耦依赖的 interface，验证代码是否正确调用了依赖的方法。Testify各大测试库几乎都会引用, 包含很多包, 其中 assert最为常用, 更多包查看: https://github.com/stretchr/testify注意事项http框架中, 应当分为handle service repository三层, 其中repo层应当有 interface.go 用于标记数据库操作需要实现的函数. 以便切换数据库和单元测试时方便go-sqlmock / 手动实现接口. 还方便 handle 和 service 层调用时 gomock 自动生成
搞懂otelgrpc库是如何做到跨程序链路追踪的
基础grpc是基于http/2.0协议的, otelgrpc库中大量出现metadata可以简单理解成http请求中的header核心文件go.opentelemetry.io\contrib\instrumentation\google.golang.org\grpc\otelgrpc@v0.65.0\stats_handler.go整个trace功能的核心都在这个文件里面.删除了import和注释, 删除了无用函数. 用$1 $2代表参数1 参数2// TagRPC 给每一次rpc调用附加信息 func (h *clientHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context { // 提取info.FullMethodName (/GreetService/SayHello) // name是判断FullMethodName如果第一个字符是/就去掉第一个字符返回, 不是直接返回FullMethodName. 第二个返回是[]attribute.KeyValue{semconv.RPCMethod(name)} // 路径: go.opentelemetry.io\contrib\instrumentation\google.golang.org\grpc\otelgrpc@v0.65.0\internal\parse.go:20 // name: GreetService/SayHello // attrs: 有一个key为rpc.method的值 name, attrs := internal.ParseFullMethod(info.FullMethodName) // attrs类型attribute.KeyValue类型, 一个key可以有多个value. 这里用semconv.RPCSystemNameGRPC再添加rpc.system.name的key attrs = append(attrs, semconv.RPCSystemNameGRPC) if record { // 创建一个新的切片, 以避免错误识别到metrics所使用的相同attrs切片中 spanAttributes := make([]attribute.KeyValue, 0, len(attrs)+len(h.SpanAttributes)) spanAttributes = append(append(spanAttributes, attrs...), h.SpanAttributes...) ctx, _ = h.tracer.Start( ctx, name, trace.WithSpanKind(trace.SpanKindClient), trace.WithAttributes(spanAttributes...), ) } // 1. inject内部用metadata获取$1 ctx的内容. // 2. 把上面h.tracer.Start创建的spanid、traceid注入到propagators. // 3. 以key为traceparent, value为00-{traceID}-{spanID}-01 格式来填入. // 4. 用metadata再把md包装一下返回ctx return inject(context.WithValue(ctx, gRPCContextKey{}, &gctx), h.Propagators) } 先来总结一下TagRPC做了哪些事情吧用pb里面定义的service名拼接调用的rpc函数名. 如果第一个是/就去掉赋值给name, 然后给attrs赋值一个key是rpc.method value是未去除/的name给attrs添加key是rpc.system.name value是grpc判断filter是否为空, 不为空用filter过滤一下info创建一个新的spanAttrs避免和metrics冲突创建一个子spanmetrics代码这里跳过...把子span返回的ctx包裹上gctx, 然后提取内容注入给全局propagator(如果没有应当会新建一个)结束, 执行HandleRPC核心就是inject添加了tranceparent的metadata. 然后在服务端接收, 然后otel.GetTextMapPropagator().Extract() 提取传递来的span, 他会自动new成一个trace.SpanContext返回, 然后需要手动trace.SpanContextFromContext(ctx) 提取成span. 进而就可以使用tracer.Start() 开启一个子span.客户端和服务端均需连接到otel. jaeger依靠traceID、parentID、spanID来创建一个trace. 你可以尝试给客户端的end前sleep一下, 会发现服务端的span已经到jaeger而没有客户端的span. sleep结束后jaeger自动合并两个span到一个trace.tranceparent传递格式: 00-{traceID}-{spanID}-01最后的01翻看Extract好像是采样有关的. 断点+翻代码+搜索搞了一天, 懒得再研究了, 相关内容太少了, 而且文档也不详细.简单实现服务端package main import ( pb "alitracing/client/proto" "context" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/jaeger" "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.21.0" "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" "google.golang.org/grpc/metadata" "log" "net" "time" ) func initTracer(serviceName string) (*sdktrace.TracerProvider, error) { exporter, err := jaeger.New( jaeger.WithCollectorEndpoint( jaeger.WithEndpoint("http://127.0.0.1:14268/api/traces"), ), ) if err != nil { return nil, err } log.Printf("[%s] Jaeger 导出器初始化成功", serviceName) // 创建资源属性 res, err := resource.New( context.Background(), resource.WithAttributes( semconv.ServiceNameKey.String(serviceName), semconv.ServiceVersionKey.String("1.0.0"), attribute.String("environment", "dev"), ), ) if err != nil { return nil, err } tp := sdktrace.NewTracerProvider( sdktrace.WithSyncer(exporter), sdktrace.WithResource(res), ) // 设置全局 TracerProvider 和 Propagator otel.SetTracerProvider(tp) otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( propagation.TraceContext{}, propagation.Baggage{}, )) log.Printf("[%s] OTel 初始化完成", serviceName) return tp, nil } type metadataCarrier struct { md metadata.MD } func (c *metadataCarrier) Get(key string) string { values := c.md.Get(key) if len(values) == 0 { return "" } return values[0] } func (c *metadataCarrier) Set(key, value string) { c.md.Set(key, value) } func (c *metadataCarrier) Keys() []string { keys := make([]string, 0, len(c.md)) for k := range c.md { keys = append(keys, k) } return keys } type server struct { pb.UnimplementedGreetServiceServer tracer trace.Tracer } func (s *server) SayHello(ctx context.Context, req *pb.SayHelloReq) (*pb.SayHelloResp, error) { log.Println("[服务端] 收到请求，开始提取 Trace 上下文") // 提取入站 Metadata md, ok := metadata.FromIncomingContext(ctx) if !ok { md = metadata.New(nil) log.Println("[服务端] 未提取到入站 Metadata，使用空 Metadata") } else { log.Printf("[服务端] 提取到入站 Metadata: %v", md) } // 从 Metadata 提取远程 SpanContext carrier := &metadataCarrier{md: md} ctx = otel.GetTextMapPropagator().Extract(ctx, carrier) // 验证提取到的 SpanContext extractedSC := trace.SpanContextFromContext(ctx) if extractedSC.IsValid() && extractedSC.IsRemote() { log.Printf("[服务端] 成功提取远程 SpanContext: TraceID=%s, SpanID=%s", extractedSC.TraceID(), extractedSC.SpanID()) } else { log.Println("[服务端] 未提取到有效的远程 SpanContext，将创建新的根 Span") } // 创建服务端 Span ctx, span := s.tracer.Start( ctx, "Greeter/SayHello", trace.WithAttributes( semconv.RPCMethodKey.String("SayHello"), semconv.RPCServiceKey.String("Greeter"), attribute.String("request.name", req.Name), ), ) defer func() { span.End() log.Printf("[服务端] Span 结束: TraceID=%s, SpanID=%s", span.SpanContext().TraceID(), span.SpanContext().SpanID()) }() time.Sleep(time.Second) // 业务逻辑 return &pb.SayHelloResp{Msg: "Hello " + req.Name}, nil } func main() { tp, err := initTracer("grpc-server") if err != nil { log.Fatalf("初始化 OTel 失败: %v", err) } defer func() { _ = tp.Shutdown(context.Background()) }() lis, err := net.Listen("tcp", ":50051") if err != nil { log.Fatalf("监听失败: %v", err) } s := grpc.NewServer() pb.RegisterGreetServiceServer(s, &server{tracer: otel.Tracer("grpc-server-tracer")}) log.Println("服务端启动在 :50051") if err := s.Serve(lis); err != nil { log.Fatalf("服务启动失败: %v", err) } time.Sleep(time.Hour) } 客户端package main import ( pb "alitracing/client/proto" "context" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/jaeger" "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.21.0" "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/metadata" "log" "time" ) func initTracer(serviceName string) (*sdktrace.TracerProvider, error) { exporter, err := jaeger.New( jaeger.WithCollectorEndpoint( jaeger.WithEndpoint("http://127.0.0.1:14268/api/traces"), ), ) if err != nil { return nil, err } log.Printf("[%s] Jaeger 导出器初始化成功", serviceName) // 创建资源属性 res, err := resource.New( context.Background(), resource.WithAttributes( semconv.ServiceNameKey.String(serviceName), semconv.ServiceVersionKey.String("1.0.0"), attribute.String("environment", "dev"), ), ) if err != nil { return nil, err } tp := sdktrace.NewTracerProvider( sdktrace.WithSyncer(exporter), sdktrace.WithResource(res), ) // 设置全局 TracerProvider 和 Propagator otel.SetTracerProvider(tp) otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( propagation.TraceContext{}, propagation.Baggage{}, )) log.Printf("[%s] OTel 初始化完成", serviceName) return tp, nil } type metadataCarrier struct { md metadata.MD } func (c *metadataCarrier) Get(key string) string { values := c.md.Get(key) if len(values) == 0 { return "" } return values[0] } func (c *metadataCarrier) Set(key, value string) { c.md.Set(key, value) } func (c *metadataCarrier) Keys() []string { keys := make([]string, 0, len(c.md)) for k := range c.md { keys = append(keys, k) } return keys } func main() { time.Sleep(1 * time.Second) tp, err := initTracer("grpc-client") if err != nil { log.Fatalf("初始化 OTel 失败: %v", err) } defer func() { // 确保 TracerProvider 关闭，Flush 剩余 Span ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := tp.Shutdown(ctx); err != nil { log.Printf("关闭 TracerProvider 失败: %v", err) } log.Println("[客户端] TracerProvider 已关闭") }() tracer := otel.Tracer("grpc-client-tracer") conn, err := grpc.Dial( "localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()), ) if err != nil { log.Fatalf("连接失败: %v", err) } defer conn.Close() client := pb.NewGreetServiceClient(conn) // 创建客户端根 Span ctx, span := tracer.Start( context.Background(), "ClientCall", trace.WithAttributes( semconv.RPCMethodKey.String("SayHello"), semconv.RPCServiceKey.String("Greeter"), ), ) defer func() { span.End() log.Printf("[客户端] Span 结束: TraceID=%s, SpanID=%s", span.SpanContext().TraceID(), span.SpanContext().SpanID()) }() log.Printf("[客户端] 创建客户端 Span: TraceID=%s, SpanID=%s", span.SpanContext().TraceID(), span.SpanContext().SpanID()) // 注入 SpanContext 到 Metadata md := metadata.New(nil) carrier := &metadataCarrier{md: md} otel.GetTextMapPropagator().Inject(ctx, carrier) log.Printf("[客户端] 注入 Trace 上下文到 Metadata: %v", md) // 发起 RPC ctx = metadata.NewOutgoingContext(ctx, md) resp, err := client.SayHello(ctx, &pb.SayHelloReq{Name: "Alice"}) if err != nil { log.Fatalf("调用失败: %v", err) } log.Printf("[客户端] 收到响应: %s", resp.Msg) time.Sleep(2 * time.Second) } 作者bb我将永远谴责文档写的不全的人. 同时谴责只会照搬example 发博客的啥X.感谢[]: https://www.doubao.com/chat/ "豆包"别看了下面代码跟传递span无关下面来看HandleRPC函数// HandleRPC processes the RPC stats. func (h *clientHandler) HandleRPC(ctx context.Context, rs stats.RPCStats) { // 这里直接调用的下面handleRPC省略了 } type int64Hist interface { RecordSet(context.Context, int64, attribute.Set) } func (c *config) handleRPC( ctx context.Context, rs stats.RPCStats, duration metric.Float64Histogram, inSize, outSize int64Hist, recordStatus func(*status.Status) (codes.Code, string), ) { // 从之前TagRPC里拿出来metrics的东西 gctx, _ := ctx.Value(gRPCContextKey{}).(*gRPCContext) if gctx != nil && !gctx.record { // 这里似乎一直不会执行, 因为TagRPC中record一直为true, 没有看到有其他更改的地方 return } // 从ctx里面拿出TagRPC里创建的span span := trace.SpanFromContext(ctx) // 默认为0 var messageId int64 // 这里用switch判断rs的六个状态执行不同代码, 在下面进行解释 switch rs := rs.(type) 在断点时一共有7次断点,下面列出每次断点rs的类型.stats.Begin 未进入switch???stats.DelayedPickComplete 未进入switchstats.OutHeader 进入case// 如果span活跃且可以记录事件时, 返回true if span.IsRecording() { // 从ctx获取对方节点信息, 即服务端 IP+PORT if p, ok := peer.FromContext(ctx); ok { // p.Addr.String() : 127.0.0.1:8085 // serverAddrAttrs(p.Addr.String()) : attribute.KeyValue类型 // k1: server.address v1: 127.0.0.1 // k2: server.port v2: 8085 // 最后给span设置attrs如果k存在则覆写 span.SetAttributes(serverAddrAttrs(p.Addr.String())...) } }stats.OutPayload 进入case// metrics处理 if gctx != nil { messageId = atomic.AddInt64(&gctx.outMessages, 1) outSize.RecordSet(ctx, int64(rs.Length), gctx.metricAttrSet) } // c.SentEvent为false 没进入 if c.SentEvent && span.IsRecording() { span.AddEvent("message", trace.WithAttributes( semconv.RPCMessageTypeSent, semconv.RPCMessageIDKey.Int64(messageId), semconv.RPCMessageCompressedSizeKey.Int(rs.CompressedLength), semconv.RPCMessageUncompressedSizeKey.Int(rs.Length), ), ) }InHeader 未进入Intrailer 未进入InPayload 进入caseif gctx != nil { messageId = atomic.AddInt64(&gctx.inMessages, 1) inSize.RecordSet(ctx, int64(rs.Length), gctx.metricAttrSet) } if c.ReceivedEvent && span.IsRecording() { span.AddEvent("message", trace.WithAttributes( semconv.RPCMessageTypeReceived, semconv.RPCMessageIDKey.Int64(messageId), semconv.RPCMessageCompressedSizeKey.Int(rs.CompressedLength), semconv.RPCMessageUncompressedSizeKey.Int(rs.Length), ), ) }End 进入casevar rpcStatusAttr attribute.KeyValue var s *status.Status // 如果响应错误就用远程错误码赋值rpcStatusAttr, 否则用grpc_codes.OK.String()赋值 if rs.Error != nil { s, _ = status.FromError(rs.Error) rpcStatusAttr = semconv.RPCResponseStatusCode(s.Code().String()) } else { rpcStatusAttr = semconv.RPCResponseStatusCode(grpc_codes.OK.String()) } // 如果span还能记录, 把前面的rpcStatusAttr 设置给span 然后关闭span if span.IsRecording() { if s != nil { c, m := recordStatus(s) span.SetStatus(c, m) } span.SetAttributes(rpcStatusAttr) // 结束会自动赋值endTime span.End() } // metrics, 这里略过 var metricAttrs []attribute.KeyValue if gctx != nil { metricAttrs = make([]attribute.KeyValue, 0, len(gctx.metricAttrs)+1) metricAttrs = append(metricAttrs, gctx.metricAttrs...) } metricAttrs = append(metricAttrs, rpcStatusAttr) recordOpts := []metric.RecordOption{metric.WithAttributeSet(attribute.NewSet(metricAttrs...))} // 与上面span无关 elapsedTime := float64(rs.EndTime.Sub(rs.BeginTime)) / float64(time.Millisecond) duration.Record(ctx, elapsedTime, recordOpts...)
在了解方案前先了解一下几个理论理论CAPC consistency 一致性: 所有节点在同一时间访问时, 拿到的均为最新数据A availability 可用性: 向非故障节点发起请求后, 每一次请求都能在有限时间内获得合理的响应（而非错误或超时）P partition tolerate 分区容错性: 部分节点挂了, 依旧能正常对外提供服务在分布式系统下,P是前提,仅能实现CP/APCP 一致性 + 分区容错: 当网络分区发生时，为了保证数据一致，系统会拒绝服务或报错，牺牲可用性（A）。适用于银行转账、交易系统AP 可用性 + 分区容错: 当网络分区发生时，系统保证可用，但允许读取旧数据，牺牲强一致性（C），满足最终一致性。适用于大多数Web应用、社交媒体CA 一致性 + 可用性: 仅可在单机环境下成立,常用数据库如mysql oracle等BASEBASE是下面几个单词的缩写, 此理论是互联网实践而来,是对CAP理论的演变和权衡结果.Basically Available 基本可用: 在系统发生故障或高负载时, 必须保证系统核心功能可用, 非核心功能使用降级熔断等策略保证--基本可用Soft-state 软状态: 允许系统存在短暂、无锁、可过渡的中间状态，节点间数据暂时不一致, 但不影响整体可用性.​ 与硬状态(ACID事务, 实时锁定)相对Eventually Consistent 最终一致性: BASE理论强调最终一致性, 即在有限时间和有穷步后, 结果最终一致.一致性的三个状态一致性强调结果, 即数据最后能否一致强一致性: 系统写了什么, 读出来就是什么弱一致性: 不保证多久后读取数据是最新的, 甚至永远可能读不到最新值. 都不保证最终一致性: 现在不一定一致, 但在有限时间后一定一致幂等性同一个操作执行多次, 结果相同常见方案XA 标准化分布式事务这是对2PC 的标准化实现, 是X/Open 组织制定的、跨数据库/消息队列的分布式事务工业级标准协议. 流程参见2PC支持XA 的有MySQL、Oracle、PostgreSQL、SQL Server、RocketMQ、Kafka 等几乎所有主流中间件2PC 两阶段提交借用XA 的角色定义: AP 业务应用(Application Program), TM 事务管理器(Transaction Manager), RM 参与者(Resource Manager)AP 向TM 发起全局事务TM向RM A, B发送要执行的SQlRM A, B开启事务, 执行业务SQL, 两个执行成功/失败均向TM 返回TM 根据返回决定向RM 发送commit或rollback优点: 强一致, 开发简单. 适用于金融等系统缺点: 性能差, 协调者节点挂了会导致锁无法释放(可用心跳包等方法解决), 极端情况下会导致一部分commit, 一部分rollback3PC 三阶段提交CanCommit: 仅检查库存, 锁.PreCommit: 所有节点的Can 均返回成功时, 向所有节点发送PreCommit, 此步骤开启事务但不提交.DoCommit: 所有节点的PreCommit 均返回成功时,向所有节点发送DoCommit.3PC 为所有节点增加了超时决策逻辑, 在PreCommit 后超时可选自动Abort/DoCommit, 在CanCommit 超时自动Abort.极端情况下仍有可能导致协调者只给部分节点发送了DoCommit, 剩下节点可能Abort/DoCommit.生产几乎不用AT 自动事务模式 (Seata)官方链接: https://seata.apache.org/zh-cn/docs/overview/what-is-seata 你可以阅读官方文档获取更详细的说明三个角色TM（Transaction Manager）: 业务应用里的事务发起方TC（Transaction Coordinator）: Seata Server, 管理全局事务状态、协调各分支提交/回滚、维护全局锁RM（Resource Manager）: 由 Seata 代理数据源实现, 自动拦截 SQL、生成日志、执行回滚一阶段执行流程开启本地事务, 解析SQL 的操作(update等)、表（product），条件（where name = 'TXC'）等相关的信息查询前镜像：根据SQL定位到记录, 将记录保留一份执行业务 SQL, 将执行后的结果当做后镜像: 根据前镜像的结果，通过 主键 定位数据将前后镜像和业务SQL 组成一条回滚日志记录, 插入到UNDO_LOG 表中向TC 申请指定表的指定主键的记录的全局锁如果申请到全局锁, 本地事务提交(携带undo log一块). 如果拿锁失败则继续尝试, 超出尝试范围将放弃. 回滚本地事务, 释放本地锁将本地事务结果上报给TC二阶段执行流程回滚收到 TC 的分支回滚请求, 开启本地事务根据 XID(全局事务锁ID) 和 Branch ID(分支事务ID) 查找到 UNDO_LOG 表中对应记录那记录中的后镜像与当前数据比较, 如果不同, 则说明数据被当前全局事务之外的动作修改. 此时根据配置策略处理, 详见文档根据 UNDO_LOG 的前镜像和 SQL 语句生成对应回滚语句提交本地事务. 上报本地事务执行结果给 TC提交收到 TC 的分支提交请求, 将请求放入异步队列, 立刻响应处理成功结果给 TC异步队列删除 UNDO_LOG 表的记录更详细流程下方解释和图表来自对豆包的总结, 不保证正确.TC 是被动接收方, 通过 @GlobalTransactional 注解实现开启事务、提交、回滚、获取当前状态等方法. 根据context判断是rm还是tm, tc在一开始是不知道有多少rm的, 而是通过 tm 调用 rm, 然后 rm 主动携带 branchID xid 等向 tc 报道. 在完成后 tm 会根据执行向 tc 上报 commit/rollback步骤发起方协议 / 动作接收方作用1TMGlobalBeginRequestTC开启全局事务，获取 XID2TMRPC 调用 (含 XID)RM业务驱动 RM 干活3RM执行本地 SQL, 生成 UNDO LOGDB执行业务逻辑4RMBranchRegisterRequest (含 lockKey)TCRM 主动报到，TC 知道了 RM 的存在并分配全局锁5RMCommit Local TransactionDB提交本地事务，释放本地锁6TMGlobalCommit/RollbackRequestTC告诉 TC 是提交还是回滚7TCBranchCommit/RollbackRequestRMTC 主动推送命令，让 RM 执行二阶段操作8RM删除 UNDO_LOG / 执行回滚 SQLDB完成二阶段收尾TCC Try-Confirm-Cancel用户下单商品, 执行Try检查商品和预留字段的库存, 增加预留库存操作1 无问题执行Confirm, 实际扣减商品库存, 删除预留库存. 操作1 存在问题执行Cancel, 取消预留库存, 执行其他恢复代码优点: 性能好于2PC, 因为取消了事务导致的长锁(但可能需要短锁/乐观锁保证库存检查和预留)缺点: 开发量大, 每个业务均需编写三个接口: TrySell, ConfirmSell, CancelSell, 需修改数据库表结构衍生方案通用型 TCC 异步化try阶段不变, confirm/cancel 阶段基于 MQ 异步异步确保型 TCC借用 Seata 图片, 感觉类似于 RocketMQ 的半消息机制try阶段发消息给mq, 只做存储, 消费者无法读取到confirm/cancel 阶段消费者才可以读取补偿型 TCChttps://seata.apache.org/zh-cn/blog/tcc-mode-applicable-scenario-analysis/#%E5%9B%9B%E8%A1%A5%E5%81%BF%E5%9E%8B-tcc-%E8%A7%A3%E5%86%B3%E6%96%B9%E6%A1%88 seata官网文章补偿型 TCC 适用并发冲突少 / 外部业务, 只需要 Try 和 Cancel 两个接口即可实现我向A, B发送Do请求如果都成功则正常处理如果A / B 出现问题则发送 Compensate 请求补偿seata会自己解决原子性问题, 这里我懒得深入了解了基于本地消息的最终一致性方案拥有业务表A, 信息表B执行业务, 首先启动事务, 事务中执行业务, 同时将消息插入消息表中. commit/rollback 保证业务和消息100%不会丢失定时任务扫描信息表, 发送信息到MQ消费方读取消息处理优点: 实现简单, 不依赖外部消息中间件的事务, 本地消息绝对不会丢失缺点: 业务表和消息表耦合, 定时任务有延迟基于可靠消息的最终一致性方案即借用外部消息中间件的事务功能, RocketMQ叫半消息机制, Kafka叫生产者事务(0.11引入), RabbitMQ叫AMQP事务RocketMQ流程极端情况下, 业务代码在执行commit/rollback时可能会出现网络波动导致没有正确处理, 这时会有事务回查(补偿)机制, RocketMQ 会主动询问此事务是commit还是rollbackKafkakafka流程同上, 但没有回查机制, 依靠三种状态: committed, aborted, incomplete和事务协调器兜底.事务开启, 消息发送commit/abort没有送达根据transaction.timeout.ms 配置, 超时将事务标记为Aborted结束消费者默认仅读取committed 消息, 可通过isolation.level=read_committed 配置可以手写补偿机制完善.RabbitMQ基于AMQP协议的原生事务性能较差, 生产环境中几乎不用, 此事务流程同上更多使用发布确认机制(流程不同上):单个确认: 批量确认: 流程和单个确认相同, 是堆积到一定数量消息后, 将消息一次性发送过去等待整批响应. 缺点是无法确定批次中哪个消息出现问题异步确认(性能最优): 但需要手动完善机制才能实现RocketMQ效果最大努力通知方案优点: 实现简单, 最柔, 整体流程无锁缺点: 一致性最差Saga将长事务拆分为多个单步事务, 每个事务对应自己的补偿失误定义事务为T, 取消为C正常流程: T1->T2->T3->T4失败失败：T1->T2->T3失败->C3->C2->C1两种模式编排式(生产常用)由协调器调配: 先做1->再做2, 如果2失败协调器下令回滚1协作式服务直接靠消息/其他手段互相通知A 做完发消息, B 监听做, B 失败发消息, A 自己回滚结论优点: 支持长事务, 并发高, 整体流程无锁(即不会像2PC 那样所有T 均上锁)缺点: 补偿逻辑复杂, 中间状态会暴露给用户衍生方案嵌套 SagaSaga 只允许两个层次的嵌套, 即 父Saga-多个子Saga. 而嵌套 Saga 则是将一个/多个子事务合并成一个独立、完整的 Saga 事务.子 Saga 拥有自己完整的正向执行流和反向补偿流子 Saga 对外暴露原子化的事务接口, 父 Saga 将其视为普通事务节点, 不感知其内部实现, 也不能单独调用子 Saga 中的内部某个事务子 Saga 不能依赖父 Saga 的其他子 Saga 资源执行流程Root Saga 流程为 A → B → C，其中 B 是嵌套子 Saga（包含 B1→B2→B3）正向：A 成功 → 子 Saga B 全流程执行成功 → C 成功，全局事务完成异常：A 成功 → 子 Saga B 执行到 B2 失败 → 子 Saga B 先补偿 B2→补偿 B1，完成自回滚 → 向 Root Saga 返回失败 → Root Saga 补偿 A，全局回滚完成基于事件溯源（Event Sourcing, ES）的事务方案ES指: 不存储实体的当前状态，只存储所有导致状态变更的不可变事件. 比如update set a = 50 where id = 1; 这里我们不存储a=50这个结果, 而是存储 将 id 为 1 的记录, a 字段修改为 50 这个操作.ES + Outbox Pattern收到操作请求, 记录请求的操作到 Event Store 中,状态标记为 待发布.按顺序投递存储的操作到 MQ, 状态标记为 已发布下游监听 MQ, 执行本地事务并写入到自身 Event Store, 并且通过唯一 id 实现幂等性.ES + Saga订单服务接收请求, 在本地事务中生成订单创建事件写入 Event Store, 发布到事件总线Saga 编排器监听订单创建成功事件, 触发库存服务扣减库存, 库存服务完成后生成库存扣减成功事件写入自身 Event Store 并发布编排器依次监听和触发, 直到整个 saga 事务结束即使服务宕机也可以通过 ES 重放CQRS(读写分离) + ESCQRS: 命令查询职责分离, 将写操作和读操作完全拆分.写侧（命令端）：完全基于 ES 实现，只负责处理业务命令、生成不可变事件、写入 Event Store，保障写事务的原子性, 是整个系统唯一的事实源读侧（查询端）：通过监听写侧的事件流, 更新适配查询场景的读库（MySQL/ES/Redis 等），通过事件的可靠消费保障读写最终一致性跨服务分布式事务通过「命令→事件→命令」的事件驱动流转实现，全程无锁、无耦合。写侧处理：业务请求以 Command 形式提交到写服务，写服务验证命令合法性后，在本地事务中生成业务事件写入 Event Store，完成写侧事务，事件发布到事件总线跨服务事务流转：下游服务监听对应事件, 转化为自身的业务 Command, 执行本地写操作并生成新的事件写入 Event Store, 完成跨服务事务的原子性流转读侧同步：读侧的投影服务（Projection）监听事件总线，根据事件类型更新对应的读库（如订单列表写入 MySQL、热点数据写入 Redis）。一致性保障：通过事件唯一 ID 实现消费幂等，确保事件只被处理一次. 读侧更新失败会自动重试，直到成功；若读侧数据出现不一致，可直接重放 Event Store 的事件，重新生成全量读库数据，彻底修复一致性问题。事务回滚：跨服务事务失败时生成补偿事件，写侧 Event Store 持久化补偿事件，读侧投影服务同步更新读库，完成回滚后的全链路一致性。参考[]: https://seata.apache.org/zh-cn/docs/overview "Seata文档"[]: https://javaguide.cn/distributed-system/protocol/cap-and-base-theorem.html "CAP & BASE理论详解"[]: https://www.doubao.com/ "豆包"
哔哔叨去年上班时候也去想过实现i18n,但是由于进度要求就没有去查看别的库什么的,而是通过定义const枚举常量返回给前端,再让前端根据常量从json文件中找到对应的文本输出.这种方法导致的就是耦合性较强,同样带来的还有前端首屏加载负担大.好处是给后端减少了在内存中加载大量文本的问题.因此现在看到相关视频来看看go实现i18n的几个方法官方非标准库实现官方有实现golang.org/x/text包,但由于是非go的标准库,所以用例和注释可能会缺少,导致有使用成本# 先安装包 go get golang.org/x/text/language go get golang.org/x/text/message # 安装cli go install golang.org/x/text/cmd/gotext@latestproject/main.gopackage main import ( "golang.org/x/text/language" "golang.org/x/text/message" "strings" ) //go:generate gotext -srclang=zh -dir=./i18n/locales/ update -out=./i18n/catalog.go -lang=zh,en func main() { temp := &strings.Builder{} p := message.NewPrinter(language.MustParse("zh")) p.Print("test") p.Println("this is test string") p.Printf("this is test string1") p.Sprint("this is test string2") p.Sprintf("this is test string3") p.Sprintln("this is test string4") p.Fprint(temp, "this is test string5") p.Fprintln(temp, "this is test string6") p.Fprintf(temp, "this is test string7") }项目根目录执行 gotext -srclang=zh -dir=./i18n/locales extract -lang=zh,en . -srclang代表项目原始语言 -dir要把生成的文件放在哪 -lang代表要生成的语言有哪些 现在目录结构如下 project │ go.mod │ go.sum │ main.go │ └─i18n └─locales ├─en │ out.gotext.json │ └─zh out.gotext.json可以看到生成了/i18n/locales/en /i18n/locales/zh文件夹和两个out.gotext.json文件.我们将两个out.gotext.json文件复制为messages.gotext.json, 然后随便打开一个可以看到只有1,3,7的文本被提取出来, 即Printf Sprintf Fprintf函数的文本会被提取.我们现在在json文件中把translation字段填上对应文本然后执行命令 gotext -srclang=zh -dir=./i18n/locales/ update -out=./i18n/catalog.go -lang=zh,en或 go generate这时候生成了 i18n/catalog.go 文件,我们需要手动修改package为i18n然后在main中新增引用 _ "alitracing/i18n" 再次运行代码输出: testthis is test string // 忽略此行 这是测试字符串1我们忽略输出的第一行,这时候发现文本已经被替换了.要在项目中运用思路如下:用gin拦截器获取前端请求头的accept-language在拦截器中使用language的matcher自动匹配语言,如果没有就返回t的第一个语言t := []language.Tag{ language.English, // en language.Chinese, // zh } matcher := language.NewMatcher(t) tag, _ := language.MatchStrings(matcher, "zh-CN") t1, _ := tag.Base() fmt.Println(t1.String()) // zh将tag放进gin上下文中传递,在需要的地方实例化printer然后用xxxxf函数打印go-i18n待写...
基于物品的协同过滤Item base collaborative filtering,这个算法的核心逻辑是: 你喜欢物品A,根据其他喜欢A的用户喜欢的物品B来计算相似度,然后推荐.ItemCF核心算法1. 定义用户交互矩阵U = { "user1": { "item1": 1, "item2": 1, "item3": 0, }, "user2": { "item1": 1, "item2": 0, "item3": 1, }, "user3": { "item1": 1, "item2": 0, "item3": 1, } }2. 计算物品相似度实际时你应当只计算有相同交互物品的用户的其他物品相似度.使用下面的余弦相似度公式计算:$$ w_{ij} = \frac{C[i][j]}{\sqrt{|N(i)| \cdot |N(j)|}} $$C[i] [j]: 物品i物品j共同交互用户数量,遍历每个用户的物品,两两配对然后加一,这就是现实了共现矩阵(统计两个或多个元素在特定上下文内同时出现频率的二维矩阵)|N(i)|和物品i有交互的用户总数,|N(j)|一样.# 实现共现矩阵 collective_matrix = {} # 物品被操作计次 item_operate = {son_name: 0 for _,item in U.items() for son_name in item.keys()} for index, v in U.items(): for ci,item in enumerate(v.keys()): if ci == len(v.keys())-1: break collective_matrix[item] = {name: 0 for name in v.keys()} break for pk in U.keys(): for item in U[pk].keys(): if U[pk][item] != 0: item_operate[item] +=1 for name,v in U.items(): # 获取item长度 l = len(v) # 获取所有key all_keys = list(v.keys()) for i in range(l): # 如果用户没有交互此物品,跳过 if v[all_keys[i]] == 0: continue # user交互的物品总数减去已经遍历过的物品数 for j in range(l-i-1): if v[all_keys[i]] == v[all_keys[i+j+1]]: collective_matrix[all_keys[i]][all_keys[i+j+1]] += 1 # 输出 print(collective_matrix) print(item_operate) # 计算两个物品的相似度 similar_dict = {} for item_parent in collective_matrix.keys(): for son_key in (collective_matrix[item_parent]).keys(): # 默认没有C[i][i]这种情况 if son_key is item_parent: continue # 如果没有人交互i和j两个物品,跳过 if collective_matrix[item_parent][son_key] == 0: continue # 开始计算不同物品的相似度 if item_parent not in similar_dict: similar_dict[item_parent] = {} similar_dict[item_parent] |= { son_key: collective_matrix[item_parent][son_key]/math.sqrt(item_operate[item_parent] * item_operate[son_key]) } print(similar_dict) # 输出 {'item1': {'item1': 0, 'item2': 1, 'item3': 2}, 'item2': {'item1': 0, 'item2': 0, 'item3': 0}} {'item1': 3, 'item2': 1, 'item3': 2} {'item1': {'item2': 0.5773502691896258, 'item3': 0.8164965809277261}}用余弦相似度公式再次确认是否正确:$$ w_{ij} = \frac{1}{\sqrt{3 \cdot 1}} \approx 0.577 $$分子C[i] [j]取值为 1, 分母中 |N(i)| =3, |N(j)| =1. 算出item1和2的结果正确3. 候选物品推荐用户最近交互的物品是abc拿abc找出相似度最高的N个物品使用下面公司按照最后得分排序输出$$ p(u, i) = \sum_{j \in N(u)} w_{ij} \cdot r_{uj} $$sigma(Σ)是表示求和.N(u)代表用户交互过的所有物品.wij表示物品相似度.ruj为用户u对物品j的的兴趣强度或者权重,可以为0这里假设ruj为1, u为user1, 感兴趣物品(i)为item3,上面公式计算示例: p(user1,3) = w13 1 + w23 1 = 0.816这里只有item1和item3有共同交互用户,所以其实可以直接得出0.816皮尔逊相关系数算法余弦相似度适用于隐式反馈场景（如点击、浏览）,但在某些应用中我们还有显式评分数据（如5星评分、点赞数等）。对于这类数据，可以使用更精细的相似度计算方法。$$ w_{ij} = \frac{\sum_{u \in U_{ij}}(r_{ui} - \bar{r}_i)(r_{uj} - \bar{r}_j)}{\sqrt{\sum_{u \in U_{ij}}(r_{ui} - \bar{r}_i)^2}\sqrt{\sum_{u \in U_{ij}}(r_{uj} - \bar{r}_j)^2}} $$uij表示同时评分过i和j的用户集合rui表示用户u对i的评分-ri是物品i的平均评分基于皮尔逊系数可以预测未接触物品的评分$$ \hat{r}_{u,j} = \bar{r}_{j} + \frac{\sum_{k \in S_j} w_{jk}\,\left( r_{u,k} - \bar{r}_{k} \right)}{\sum_{k \in S_j} w_{jk}} $$
Anti-Entropy熵是物理学上的一个概念，代表杂乱无章，而反熵就是在杂乱无章中寻求一致。本质上，反熵是一种通过异步修复实现最终一致性的方法。--钝悟节点周期性随机选择节点,交换全量数据消除差异.周期不会结束.若运行时间较长会导致传递数据过大.状态Susceptible(易感): 节点还没接收到这条消息Infective(传染性的): 节点已经持有这条消息，并且会周期性地向其他节点传播全量消息过程1. 节点周期性触发节点 A 每隔固定时间 T (例如 5 秒) 执行 Anti-Entropy随机选择一个邻居节点 B (fan-out=1,此时只会选择一个邻居节点)2. 交换状态节点 A 将自己完整状态发送给 B节点 B 将自己完整状态发送给 A3. 计算差异A 接收到 B 的状态 → 找出 B 拥有但自己缺失的部分 → 更新自己B 接收到 A 的状态 → 找出 A 拥有但自己缺失的部分 → 更新自己4. 重复执行节点 A 等待下一个周期，再随机选择一个节点执行同样操作所有节点独立、周期性地执行这个流程优点最终一致性确定节点周期性全量或差异同步，理论上所有节点最终必然一致。容错能力强即使某轮传播失败，下轮周期仍会补救，适应网络不稳定。修复遗漏节点无论之前哪个节点没收到消息，周期性同步最终都会把缺失数据同步过去。适合大规模网络同步不依赖单轮成功，概率模型收敛分析简单可靠。缺点传播速度相对慢周期性触发可能导致新消息扩散比 Rumor-Mongering 慢。带宽消耗大全量或差异同步比单条消息增量传播消耗更多网络资源。实现稍复杂需要周期触发、状态摘要或差异计算(摘要和差异是现代实践,论文中没有)，以及节点间同步机制。Rumor-Mongering仅传播最新数据. 每次传播都随机选择节点进行传播, 如果节点已经持有消息(即传播失败), 或因网络超时传播失败, 计数器+1, 达到阈值后停止.由于是增量传播,数据包小,不会无限制传播.状态Susceptible(易感): 节点还没接收到这条消息Infective(传染性的): 节点已经持有这条消息，并且会周期性地向其他节点增量传播消息。Removed: 当收到被传播节点持有此消息,或传播次数超过阈值时,停止传播这条消息过程论文传播停止是因为全网络节点已感染有点难理解,这里用工程实践版的1. 收到消息节点 A 收到消息更新 infected 列表标记这个消息已知2. 选择目标节点 A 从自己的邻居列表中随机选择几个节点 (fan-out数量)3. 尝试发送消息对每个目标节点 B：如果 B 没有收到过消息 M:B 接收消息 M,并在下一轮也会对自己的邻居进行 rumor-mongering 传播节点 A 的计数器 不增加如果 B 已经持有消息 M:节点 A 的计数器 +1 (表示一次失败传播)如果 B 网络超时:节点节点 A 的计数器 +1 (表示一次失败传播)4. 检查停止条件节点 A 对消息 M 的传播计数器达到阈值 (例如 3):节点 A 停止传播消息 M节点 A 不再尝试发送 M 给其他节点5. 下一轮传播对于尚未停止的节点,重复 1-3节点每轮随机选择邻居,尝试传播消息,直到自己的计数器达到阈值优点快速传播新消息节点收到新消息就立即传播,消息扩散速度快低带宽开销（增量传播）只发送新消息,不发送全量状态实现简单逻辑只需随机选择邻居并推送消息,无需全局状态或复杂同步表概率性最终一致性随着轮次增加,高概率网络大部分节点最终会收到消息缺点潜在不一致问题节点 A 向节点 B 传播消息 M，但因为网络超时/丢包, B 没有收到 → 本轮传播失败A 后续随机选择节点传播时 可能没再次选择 BA 的失败计数器达到阈值 → 停止传播如果其他节点也因为随机选择没选择到 B,由于Rumor-mongering不会向Anti-Entropy主动周期性和邻居节点同步, 所以 B 在下次被传播前缺失了这条消息结果：网络中存在少数节点缺失该消息 → 高概率最终一致,但不是强一致性可靠性受网络不稳定影响大网络超时或节点离线可能导致消息传播失败,且节点不会自动重新尝试收敛不可预测消息最终传播到全网的时间依赖于随机性和阈值,不保证精确收敛不能保证强一致性适合快速扩散场景,但不能保证每个节点最终状态完全一致因此现代实践中先使用Rumor-mongering快速传播,再使用Anti-Entropy保证最终一致性通信模式Push节点A全量推送自己所有数据给B节点,B节点对比差异更新自己Pull节点A将自己的key,version推送给BB对比本地数据,增量推送(包含key,value,version)比A新的数据给AA对比更新自己Push-Pull节点A将自己的key,version推送给BB对比本地数据,增量推送(包含key,value,version)比A新的数据给AA对比更新自己,再增量推送(包含key,value,version)比B新的数据给BB对比更新自己注意: Anti-Entropy和Rumor-mongering可以随意选择Push,Pull,Push-Pull搭配,而不是固定的.Reference[]: https://dunwu.github.io/waterdrop/pages/88db2f5d "分布式算法 Gossip"[]: https://chatgpt.com/ "ChatGPT"
Fx is the backbone of nearly all Go services at Uber.----https://uber-go.github.io/fx/index.htmlfx和wire选择wire每次启动前需要手动使用CLI编译才能调用依赖注入.这在启动时保证了更好的性能.fx则是在启动时进行依赖注入,因此启动性能不会很好,但是在启动完成后运行时的性能是无影响的.在考虑到业务对于启动时时长并不敏感,且wire如果新增/删除依赖对象需要再次执行cli生成,最终选择了fx.基础函数介绍fx.Provide用于提供需要的依赖对象.这个函数不会一定执行,只有当有调用此依赖对象时才会被执行.package main import ( "fmt" "go.uber.org/fx" ) type A struct{} func NewA() A { fmt.Println("A is running") return A{} } type B struct { } func NewB(p A) B { fmt.Println("B is running") return B{} } func main() { app := fx.New( fx.Provide(NewA), fx.Provide(NewB), ) app.Run() } A函数提供了1,如果B不依赖A函数返回的1,那么A函数就不会执行.注意: 此时B没有被引用,因此只是将对应类型告知给了container,并没有实例化对象// 控制台输出如下,并没有打印 running [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE main.A <= main.NewA() [Fx] PROVIDE main.B <= main.NewB() [Fx] RUNNINGfx.Invoke此函数提供的对象一定会被执行,如果你有一个函数没有被依赖,但是必须要他被执行,那就使用Invoke// 此时新增C类型 type C struct { } func NewC(b B) C { fmt.Println("C is running") return C{} } // 再将main修改成这样 func main() { app := fx.New( fx.Provide( NewA, NewB, ), fx.Invoke(NewC), ) app.Run() } 如果没有Invoke C那么Provde的A,B两个函数就不会被执行. 但是现在C一定会被执行,且依赖1,2.那么A和B也就会被执行.// 控制台输出如下,可以看到是根据依赖自动实例化对象的 [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE main.A <= main.NewA() [Fx] PROVIDE main.B <= main.NewB() [Fx] INVOKE main.NewC() [Fx] BEFORE RUN provide: main.NewA() [Fx] RUN provide: main.NewA() in 0s [Fx] BEFORE RUN provide: main.NewB() [Fx] RUN provide: main.NewB() in 0s [Fx] RUNNING A is running B is running C is runningfx.Annotate和fx.As这两个函数通常是一块出现的,用于将一个函数的返回,强行转换成另一个接口package main import ( "fmt" "go.uber.org/fx" ) type Animal interface { Say() } type A struct{} func (receiver A) Say() { fmt.Println("i am a") } func NewA() A { fmt.Println("A is running") return A{} } func Speak(animal Animal) { animal.Say() } func main() { app := fx.New( fx.Provide( fx.Annotate( NewA, fx.As(new(Animal)), ), ), fx.Invoke(Speak), ) app.Run() } // 控制台输出 [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE main.Animal <= fx.Annotate(main.NewA(), fx.As(main.Animal) [Fx] INVOKE main.Speak() [Fx] BEFORE RUN provide: fx.Annotate(main.NewA(), fx.As(main.Animal) [Fx] RUN provide: fx.Annotate(main.NewA(), fx.As(main.Animal) in 10.5µs [Fx] RUNNING A is running i am afx.ResultTags和fx.ParamTags相信你已经发现了问题,如果你转换成了多个Animal那么fx就不知道该提供哪一个,因此需要这两个函数.package main import ( "fmt" "go.uber.org/fx" ) type Animal interface { Say() } type A struct{} func (receiver A) Say() { fmt.Println("i am a") } func NewA() A { fmt.Println("A is running") return A{} } type B struct { } func (receiver B) Say() { fmt.Println("i am b") } func NewB() B { return B{} } func Speak(animal Animal) { animal.Say() } func main() { app := fx.New( fx.Provide( fx.Annotate( NewA, fx.As(new(Animal)), ), fx.Annotate( NewB, fx.As(new(Animal)), ), ), fx.Invoke(Speak), ) app.Run() }// 控制台输出如下,这里"reflect".makeFuncStub 替换成"main".NewA就好理解了 cannot provide main.Animal from [0].Field0: already provided by "reflect".makeFuncStub因此我们使用fx.ResultTags和fx.ParamTags告诉container这个函数的标记,以及我需要哪个标记的依赖.package main import ( "fmt" "go.uber.org/fx" ) type Animal interface { Say() } type A struct{} func (receiver A) Say() { fmt.Println("i am a") } func NewA() A { fmt.Println("A is running") return A{} } type B struct { } func (receiver B) Say() { fmt.Println("i am b") } func NewB() B { return B{} } func Speak(animal Animal) { animal.Say() } func main() { app := fx.New( fx.Provide( fx.Annotate( NewA, fx.As(new(Animal)), fx.ResultTags(`name:"a"`), ), fx.Annotate( NewB, fx.As(new(Animal)), fx.ResultTags(`name:"b"`), ), ), fx.Invoke( fx.Annotate( Speak, fx.ParamTags(`name:"a"`), ), ), ) app.Run() }// 控制台输出如下,可以看到我们告诉fx指定Speak传递的是A [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE main.Animal[name = "a"] <= fx.Annotate(main.NewA(), fx.ResultTags(["name:\"a\""]), fx.As(main.Animal) [Fx] PROVIDE main.Animal[name = "b"] <= fx.Annotate(main.NewB(), fx.ResultTags(["name:\"b\""]), fx.As(main.Animal) [Fx] INVOKE fx.Annotate(main.Speak(), fx.ParamTags(["name:\"a\""]) [Fx] BEFORE RUN provide: fx.Annotate(main.NewA(), fx.ResultTags(["name:\"a\""]), fx.As(main.Animal) [Fx] RUN provide: fx.Annotate(main.NewA(), fx.ResultTags(["name:\"a\""]), fx.As(main.Animal) in 0s [Fx] RUNNING A is running i am a如果把speak注释的a改成b那么会输出i am bgroup:""注释如果你要注册多个,但是发现这样一个一个显示传参或声明太过麻烦.那么你可以直接使用group注释,然后将speak的接收参数设置为[]Animal即可package main import ( "fmt" "go.uber.org/fx" ) type Animal interface { Say() } type A struct{} func (receiver A) Say() { fmt.Println("i am a") } func NewA() A { fmt.Println("A is running") return A{} } type B struct { } func (receiver B) Say() { fmt.Println("i am b") } func NewB() B { return B{} } func Speak(animals []Animal) { for _, v := range animals { v.Say() } } func CovertToAnimalInterface(a interface{}) interface{} { return fx.Annotate( a, fx.As(new(Animal)), fx.ResultTags(`group:"animals"`), ) } func main() { app := fx.New( fx.Provide( CovertToAnimalInterface(NewA), CovertToAnimalInterface(NewB), ), fx.Invoke( fx.Annotate( Speak, fx.ParamTags(`group:"animals"`), ), ), ) app.Run() }// 控制台输出如下 A is running i am a i am b [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE main.Animal[group = "animals"] <= fx.Annotate(main.NewA(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) [Fx] PROVIDE main.Animal[group = "animals"] <= fx.Annotate(main.NewB(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) [Fx] INVOKE fx.Annotate(main.Speak(), fx.ParamTags(["group:\"animals\""]) [Fx] BEFORE RUN provide: fx.Annotate(main.NewA(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) [Fx] RUN provide: fx.Annotate(main.NewA(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) in 0s [Fx] BEFORE RUN provide: fx.Annotate(main.NewB(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) [Fx] RUN provide: fx.Annotate(main.NewB(), fx.ResultTags(["group:\"animals\""]), fx.As(main.Animal) in 0s [Fx] RUNNINGfx.Supply这个函数只能用于非接口的值.当一个构造函数不依赖任何其他的对象时,可以使用.但通常可以直接使用Provide.fx.Provide(func() *Config { return &Config{Name: "my-app"} }) // 和下面的是一样的. fx.Supply(&Config{Name: "my-app"})生命周期根据附加的顺序逐个启动根据附加的顺序反过来逐个关闭fx会自动传递fx.Lifecyclefunc(context.Context) error 这里fx自动传递的context具有过期时间package main import ( "context" "fmt" "go.uber.org/fx" ) func ExpLc1(lc fx.Lifecycle) { lc.Append( fx.Hook{ OnStart: func(context.Context) error { fmt.Println("lc 1 start") return nil }, OnStop: func(context.Context) error { fmt.Println("lc 1 stop") return nil }, }, ) } func ExpLc2(lc fx.Lifecycle) { lc.Append( fx.Hook{ OnStart: func(context.Context) error { fmt.Println("lc 2 start") return nil }, OnStop: func(context.Context) error { fmt.Println("lc 2 stop") return nil }, }, ) } func main() { app := fx.New( fx.Invoke( ExpLc1, ExpLc2, ), ) app.Run() } // 控制台输出 [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] INVOKE main.ExpLc1() [Fx] BEFORE RUN provide: go.uber.org/fx.New.func1() [Fx] RUN provide: go.uber.org/fx.New.func1() in 0s [Fx] INVOKE main.ExpLc2() [Fx] HOOK OnStart main.ExpLc1.func1() executing (caller: main.ExpLc1) [Fx] HOOK OnStart main.ExpLc1.func1() called by main.ExpLc1 ran successfully in 0s [Fx] HOOK OnStart main.ExpLc2.func1() executing (caller: main.ExpLc2) [Fx] HOOK OnStart main.ExpLc2.func1() called by main.ExpLc2 ran successfully in 0s [Fx] RUNNING lc 1 start lc 2 start [Fx] INTERRUPT [Fx] HOOK OnStop main.ExpLc2.func2() executing (caller: main.ExpLc2) [Fx] HOOK OnStop main.ExpLc2.func2() called by main.ExpLc2 ran successfully in 0s [Fx] HOOK OnStop main.ExpLc1.func2() executing (caller: main.ExpLc1) [Fx] HOOK OnStop main.ExpLc1.func2() called by main.ExpLc1 ran successfully in 0s lc 2 stop lc 1 stopModulesfx.Module第一个参数传递模块名称,用于日志中查询使用如果设置了fx.Private那么对象会被私有,任何非此module的对象将无法获取返回的对象package main import ( "go.uber.org/fx" ) type A struct { } func NewA() A { return A{} } type B struct { A } func NewB(a A) B { return B{a} } func main() { app := fx.New( fx.Module("name", fx.Provide( fx.Private, // 作用域仅在当前Provide NewA, ), ), fx.Invoke( NewB, ), ) app.Run() } // 控制台输出 [Fx] PROVIDE fx.Lifecycle <= go.uber.org/fx.New.func1() [Fx] PROVIDE fx.Shutdowner <= go.uber.org/fx.(*App).shutdowner-fm() [Fx] PROVIDE fx.DotGraph <= go.uber.org/fx.(*App).dotGraph-fm() [Fx] PROVIDE (PRIVATE) main.A <= main.NewA() from module "name" // 这里能看到module的第一个参数用于命名方便日志查看 [Fx] INVOKE main.NewB() [Fx] ERROR fx.Invoke(main.NewB()) called from: missing type: - main.A (did you mean to Provide it?) // module的fx.Private设置后,NewA返回的a仅在name module内能够使用,外部无法获取参数对象和结果对象参数对象目的: 为函数封装依赖，提升可读性与扩展性,通常专为某个函数定义,不与其他函数共享.type Params struct { // 命名规范: 若构造函数叫NewClient,那么参数命名为ClientParams. // 如果构造函数叫New,参数命名为Params. // 非强制,但官方推荐. fx.In // 结构体必须嵌入fx.In字段 Config ClientConfig // 所有字段必须导出 HTTPClient *http.Client Logger *zap.Logger `optional:"true"` // 可选参数,用于向后兼容,但在构造函数中必须处理nil情况 } func New(p Params) (*Client, error) { log := p.Logger if log == nil { log = zap.NewNop() } // 使用 log... }结果对象命名规范同上必须嵌入fx.Out所有字段必须导出没有 \`optional:"true"\`,如果需要可选,那么设置类型为指针,然后设置为nil为某个函数定义,不与其他函数共享.如果某个字段没有设置为指针类型,那么go会自动创建这个类型的结构体自动注入.
前置请先看完 http://www.kailing.pub/raft/index.html 后继续查看文章角色leaderfollowercandidate每个节点同时只扮演一个角色,允许切换Leader领导者,接收客户端请求,使用日志条目复制数据到Follower维护状态term: 当前任期号log[]: 日志条目数组,每个条目包含客户端指令和产生该条目的任期号commitIndex: 已被集群多数节点复制且可以提交的日志最大索引（所有≤该索引的日志均可应用到状态机）lastApplied: 已应用到状态机的日志最大索引（记录本地状态机执行到哪条日志）nextIndex[]: 对每个 follower 维护一个值，表示下一次向该 follower 发送日志的起始索引matchIndex[]: 对每个 follower 维护一个值，表示该 follower 已成功复制的最大日志索引Follower跟随者,被动接受Leader的日志复制请求和心跳包维护状态term: 当前任期号（与 leader 保持一致，或在发现更高任期时更新）log[]: 日志条目数组（与 leader 同步的日志副本）commitIndex: 已被集群多数节点复制且可以提交的日志最大索引（由 leader 通过心跳同步）lastApplied: 已应用到状态机的日志最大索引（记录本地状态机执行到哪条日志）votedFor: 当前任期内投票给的节点 ID（用于避免重复投票，保证每个任期仅投一票）超时计时器：用于检测 leader 是否失联（若超时未收到 leader 心跳，会转换为 candidate 发起选举）Candidate候选者,当Follower在一定时间没有接收到Leader心跳包后将从follower切换到candidate进行选举请求日志复制作用: 专门用于向 follower 同步新的日志条目内容: 包含 leader 的任期号（term）、前一条日志的索引和任期（用于 follower 校验匹配）、需要同步的日志条目（可能是一条或多条）、leader 当前的commitIndex（用于同步提交状态）心跳包作用：用于维持leader的地位,定时发送,防止follower因超时触发新的选举。内容：格式与日志复制请求基本一致，但不包含任何新的日志条目（可以理解为空日志的复制请求），同样会携带 leader 的term和commitIndex投票请求作用: 竞选leader内容: 等待150ms-300ms随机时间过期后为自身term+1,给自己投一票,然后向其他节点发送投票请求,多数派同意后变成leader成员变更请求作用: 动态增删节点,避免脑裂问题执行流程客户端向Leader发送请求Leader接受到请求,使用日志复制将数据同步到Follower多数派Follower处理完成响应给LeaderLeader收到多数派响应,修改自身CommitIndex,返回响应给客户端Leader等待下次心跳包进行同步提交信息如果在同步提交信息前挂掉了怎么办?此时Follower变成Candidate进行选举,注意,只有日志与多数派相同的Candidate才会被选举为Leader新的Leader当选,任期+1,立刻触发日志同步步骤如下向所有follower发送包含自身日志信息的同步请求（基于nextIndex初始化值）当follower返回响应时,NewLeader会通过matchIndex记录每个follower已复制的最大日志索引当NewLeader发现某条日志的matchIndex已满足多数派已复制（即超过半数 follower 的matchIndex≥该日志索引），就会直接将自己的commitIndex更新为该日志索引此时之前Leader未同步提交的信息将会被标记提交Reference[]: http://www.kailing.pub/raft/index.html "Raft 算法动画演示"[]: https://chatgpt.com/ "ChatGPT"[]: https://www.doubao.com/ "豆包"
介绍There is only one consensus protocol, and that’s “Paxos” — all other approaches are just broken versions of Paxos—— Mike Burrows角色proposal(提议者),发起提案acceptor(接受者),决策提案learner(学习者),当提案被通过,会被学习者学习,学习者不参与提案的发起和抉择[!IMPORTANT]一个agent允许扮演多个角色提案一次提案通过多轮决策最终决定一个唯一值.每个提案均会有instance_id,下面用i代替每个agent均会带有promised_rnd: 已承诺决策ID,下面用prnd代替accepted_rnd: 已接受决策ID,下面用arnd代替accepted_v: 已接受决策值,下面用av代替传递的rnd不仅会有round_id,还会携带proposal的唯一标识实现理想情况假设现有4个节点(agent),其中A为proposal,BCD为Acceptorproposal A向BCD发送信息: i=1,rnd=(1,A)收到多数派返回结果: i=1,prnd=(1,A),arnd=0,av=nilA发现prnd和自己的rnd相同,因此开始Phase2,发送: i=1,rnd=(1,A),v="G"接收到多数派节点返回信息: i=1,arnd=1,av="G",prnd=1理想情况下提案结束.并发写冲突X向多数派BC发送: i=1,rnd=(1,X)X收到多数派返回: i=1,prnd=(1,X),arnd=0,av=nilY向多数派CD发送: i=1,rnd=(1,Y)Y收到多数派返回: C: i=1,prnd=(1,X),arnd=0,av=nilD: i=1,prnd=(1,Y),arnd=0,av=nilY查询多数派返回的prnd != rnd,rnd+1继续发送请求 i=1,rnd=(2,Y) 给BCY收到多数派返回: i=1,prnd=(2,Y),arnd=0,av=nilX此时进行Phase2发送请求: i=1,rnd=(1,X),v="G"B节点返回: i=1,prnd=(1,X),arnd=1,av="G" B节点更新成功C节点判断prnd!=rnd返回: i=1,prnd=(2,Y),arnd=0,av=nil C节点更新失败Y进行Phase2发送请求: i=1,rnd=(2,Y),v="J"Y收到多数派返回: i=1,prnd=(2,Y),arnd=2,av="J"此时X重新进行Phase1,由于上面被拒绝,这次rnd增加为3,发送请求: i=1,rnd=(3,X)X收到多数派返回:B: i=1,prnd=(3,X),arnd=1,av="G"C: i=1,prnd=(3,X),arnd=2,av="J"X发现多数派响应中,有两个接受的值,av="G" av="J"因此,这次不能写入自己的值,而是进行修复.由于C节点的arnd最新,考虑到paxos算法的约定: 只接收rnd大于自己的请求,如果av="G"已经被多数派确认,那么av="J"一定不会被写入. 所以将av="J"确定为要修复值X向BC节点进行Phase2请求: i=1,rnd=(3,X),v="J"X接收到多数派响应: i=1,prnd=(3,X),arnd=3,av="J"提案结束最终值J被确定极端情况--rnd无限提升X向BC发送i=1,rnd=1X收到多数派成功返回Y向CD发送i=1,rnd=1失败(这里没画出来),提升rnd后发送i=1,rnd=2Y收到多数派成功返回X向BC发送i=1,rnd=1,v="X",B节点成功,C节点失败X提升rnd,向BC发送i=1,rnd=3,收到成功响应Y向CD发送i=1,rnd=2,v="Y",D节点成功,C节点失败Y提升rnd .....无限重试...Muti-Paxos将多个提案的Phase1合并到一个请求中发送Fast-Paxos逻辑一步到位直接发送Phase2请求,其中rnd=0,这样可以保证冲突时安全退回,因为classic-paxos的rnd一定>=1如果Fast-Paxos冲突,退化到Classic-Paxos处理为什么增加多数派数量引用xp大佬的图除X,Y外,中间五个节点从左到右命名为ABCDEX使用rnd=0占用了ABC.因为rnd同为0,Y只占据了DE.Y退化至classic-paxos.此时CDE的arnd都为0,Y无法判断哪个值被多数派接受,如果选择CDE则最终值为y0,但实际x0为多数派接受.此时应当修改多数派从n/2+1改为n*(3/4)+1,这样接受X的多数派数量为4个,Y不论如何去选都至少有两个节点av为x0,通过至少两个节点有相同的值Y可以确定被接受的值.Reference[]: https://blog.openacid.com/algo/paxos "可靠分布式系统-paxos的直观解释"[]: https://www.doubao.com/ "豆包"[]: https://github.com/Trekkiii/paxos_raft_protocol "paxos_raft_protocol"