怎样用Golang实现分布式事务讲解Saga模式与DTF框架实践

风浩小哥_5926

发布时间：2025-07-10 11:24:03

749人浏览过

来源于php中文网

原创

在golang中实现分布式事务，可使用saga模式或dtf框架。1.saga模式通过拆分事务并定义补偿操作来保证一致性，适用于业务流程长、参与者多的场景，优点是实现简单，缺点是需手动编写补偿逻辑且可能最终一致；2.dtf框架提供结构化管理，支持tcc等模式，适合对一致性要求高的场景，优点是具备acid特性，缺点是复杂度高、侵入性强；3.选择依据：若允许最终一致性且希望低侵入，选saga模式；若需强一致性并接受高复杂度，选dtf框架；4.实现时需处理网络异常，采用重试、超时、幂等性及补偿机制；5.性能优化包括减少参与者、异步提交、本地事务和数据库优化。

怎样用Golang实现分布式事务讲解Saga模式与DTF框架实践

用Golang实现分布式事务，核心在于保证多个服务操作要么全部成功，要么全部失败。Saga模式和DTF框架都是解决这个问题的有效方案，它们通过不同的策略来协调各个服务的本地事务，最终达到数据一致性。

Saga模式和DTF框架实践

Saga模式是一种补偿型的分布式事务方案，它将一个大的事务拆分成多个本地事务，每个本地事务对应一个服务的操作。如果其中一个本地事务失败，Saga模式会执行一系列补偿操作，撤销之前成功的本地事务，从而保证数据的一致性。

立即学习“go语言免费学习笔记（深入）”；

DTF（Distributed Transaction Framework）框架则提供了更加结构化的方式来管理分布式事务。它通常包含事务协调器、参与者等组件，协调器负责协调各个参与者的本地事务，并根据事务的结果进行提交或回滚。

如何选择Saga模式还是DTF框架？

选择Saga模式还是DTF框架，取决于你的业务场景和技术栈。

Saga模式： 适用于业务流程比较长、事务参与者较多的场景。它的优点是实现简单、侵入性小，缺点是需要手动编写补偿逻辑，且可能存在数据最终一致性的问题。
DTF框架： 适用于对事务一致性要求较高、业务流程相对简单的场景。它的优点是提供了事务的ACID特性，缺点是实现复杂、侵入性较大。

一个简单的选择依据是：如果你的业务允许最终一致性，并且你希望尽量减少对现有服务的侵入，那么Saga模式可能更适合你。如果你的业务需要强一致性，并且你愿意为了事务的可靠性付出更高的代价，那么DTF框架可能更适合你。

Saga模式的Golang实现示例

Saga模式的核心在于定义好每个本地事务的补偿操作。下面是一个简单的示例，演示如何使用Golang实现一个Saga模式：

git摸鱼周报自动化

生成周报：读取本机 Git 仓库最近一周的提交记录，按项目分类提炼总结，生成标准周报格式，通过微信推送并返回给用户。触发词：周报、weekly report、git总结、提交记录总结、工作周报。

下载

package main

import (
    "fmt"
    "log"
    "time"
)

// 定义一个简单的服务接口
type Service interface {
    DoSomething() error
    Compensate() error
}

// 模拟服务A
type ServiceA struct {
    data string
}

func (s *ServiceA) DoSomething() error {
    fmt.Println("Service A: Doing something...")
    // 模拟成功的情况
    s.data = "Service A data"
    return nil
}

func (s *ServiceA) Compensate() error {
    fmt.Println("Service A: Compensating...")
    s.data = "" // 撤销操作
    return nil
}

// 模拟服务B
type ServiceB struct {
    data int
}

func (s *ServiceB) DoSomething() error {
    fmt.Println("Service B: Doing something...")
    // 模拟成功的情况
    s.data = 123
    return nil
}

func (s *ServiceB) Compensate() error {
    fmt.Println("Service B: Compensating...")
    s.data = 0 // 撤销操作
    return nil
}

// Saga orchestrator
type SagaOrchestrator struct {
    services []Service
    completed []bool
}

func NewSagaOrchestrator(services []Service) *SagaOrchestrator {
    return &SagaOrchestrator{
        services:  services,
        completed: make([]bool, len(services)),
    }
}

func (s *SagaOrchestrator) Run() error {
    for i, service := range s.services {
        err := service.DoSomething()
        if err != nil {
            log.Printf("Service %d failed: %v\n", i, err)
            return s.compensate(i)
        }
        s.completed[i] = true
    }
    return nil
}

func (s *SagaOrchestrator) compensate(failedIndex int) error {
    fmt.Println("Starting compensation...")
    for i := failedIndex; i >= 0; i-- {
        if s.completed[i] {
            err := s.services[i].Compensate()
            if err != nil {
                log.Printf("Compensation for service %d failed: %v\n", i, err)
                // 这里可以考虑重试补偿操作，或者记录日志并人工介入
                return err
            }
            s.completed[i] = false
        }
    }
    return nil
}

func main() {
    serviceA := &ServiceA{}
    serviceB := &ServiceB{}

    saga := NewSagaOrchestrator([]Service{serviceA, serviceB})

    err := saga.Run()
    if err != nil {
        log.Fatalf("Saga failed: %v\n", err)
    } else {
        fmt.Println("Saga completed successfully!")
    }

    time.Sleep(time.Second) // 模拟等待
}

这段代码定义了两个简单的服务 ServiceA 和 ServiceB，每个服务都有 DoSomething 和 Compensate 两个方法。SagaOrchestrator 负责协调这两个服务的操作，如果其中一个服务失败，它会调用之前所有成功服务的 Compensate 方法，从而保证数据的一致性。

这个示例只是一个简单的演示，实际应用中需要考虑更多的情况，例如：

幂等性： 确保 DoSomething 和 Compensate 方法都是幂等的，即多次执行的结果和执行一次的结果相同。
重试机制： 在服务失败时，需要进行重试，避免因为网络抖动等临时性错误导致事务失败。
日志记录： 记录所有事务操作的日志，方便排查问题。

DTF框架的Golang实践：选择与使用

虽然从零开始构建一个DTF框架比较复杂，但我们可以选择一些现有的开源框架来简化开发。目前Golang生态中成熟的DTF框架相对较少，但可以考虑基于一些现有的分布式事务解决方案，例如：

Seata： 虽然Seata主要支持Java，但可以通过其提供的AT模式（Auto Transaction Mode）来实现基于数据库的分布式事务。Golang可以通过gRPC等方式与Seata Server进行交互。
Dtm： Dtm是一个开源的、支持多种事务模式的分布式事务管理器，它支持TCC、Saga、XA等多种事务模式，并且提供了Golang的SDK。

使用DTM框架的TCC模式，你需要定义Try、Confirm、Cancel三个阶段的操作。

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/dtm-labs/dtmcli"
    "github.com/dtm-labs/dtmcli/dtmimp"
    "github.com/dtm-labs/dtmgrpc"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

const (
    TransOutSvc = "localhost:8081" // 转出服务地址
    TransInSvc  = "localhost:8082"  // 转入服务地址
    DtmSvr      = "localhost:36800"  // dtm服务地址
)

var (
    TransOutPrepareURL string
    TransOutCommitURL  string
    TransOutRollbackURL string
    TransInPrepareURL  string
    TransInCommitURL  string
    TransInRollbackURL string
)

func init() {
    TransOutPrepareURL = dtmgrpc.MustGetTccGRPCURL(TransOutSvc, "/TransOutService.TransOut/Prepare")
    TransOutCommitURL = dtmgrpc.MustGetTccGRPCURL(TransOutSvc, "/TransOutService.TransOut/Commit")
    TransOutRollbackURL = dtmgrpc.MustGetTccGRPCURL(TransOutSvc, "/TransOutService.TransOut/Rollback")

    TransInPrepareURL = dtmgrpc.MustGetTccGRPCURL(TransInSvc, "/TransInService.TransIn/Prepare")
    TransInCommitURL = dtmgrpc.MustGetTccGRPCURL(TransInSvc, "/TransInService.TransIn/Commit")
    TransInRollbackURL = dtmgrpc.MustGetTccGRPCURL(TransInSvc, "/TransInService.TransIn/Rollback")
}

func main() {
    gid, err := dtmcli.NewTccGlobalTransaction(DtmSvr, func(tcc *dtmcli.TCC) error {
        // 转出
        err := tcc.CallGRPC(TransOutSvc, TransOutPrepareURL, TransOutCommitURL, TransOutRollbackURL, &TransRequest{Amount: 10})
        if err != nil {
            return err
        }
        // 转入
        err = tcc.CallGRPC(TransInSvc, TransInPrepareURL, TransInCommitURL, TransInRollbackURL, &TransRequest{Amount: 10})
        return err
    })
    if err != nil {
        panic(err)
    }
    log.Printf("tcc transaction: %s", gid)
}

// TransRequest 定义请求结构
type TransRequest struct {
    Amount int
}

// 定义grpc服务
type TransOutService struct {
    UnimplementedTransOutServiceServer
}

func (s *TransOutService) Prepare(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransOut Prepare: %v", req)
    // 预留资源，冻结账户金额
    return &Response{Result: "ok"}, nil
}

func (s *TransOutService) Commit(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransOut Commit: %v", req)
    // 真正扣减账户金额
    return &Response{Result: "ok"}, nil
}

func (s *TransOutService) Rollback(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransOut Rollback: %v", req)
    // 释放预留资源，解冻账户金额
    return &Response{Result: "ok"}, nil
}

// 定义grpc服务
type TransInService struct {
    UnimplementedTransInServiceServer
}

func (s *TransInService) Prepare(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransIn Prepare: %v", req)
    // 预留资源
    return &Response{Result: "ok"}, nil
}

func (s *TransInService) Commit(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransIn Commit: %v", req)
    // 真正增加账户金额
    return &Response{Result: "ok"}, nil
}

func (s *TransInService) Rollback(ctx context.Context, req *TransRequest) (*Response, error) {
    log.Printf("TransIn Rollback: %v", req)
    // 释放预留资源
    return &Response{Result: "ok"}, nil
}

// 启动grpc服务
func startGrpcServer(port string, service interface{}, register func(grpc.ServiceRegistrar, interface{})) {
    lis, err := net.Listen("tcp", ":"+port)
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    s := grpc.NewServer()
    register(s, service)
    log.Printf("server listening at %v", lis.Addr())
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

func main() {
    // 启动转出服务
    go func() {
        startGrpcServer("8081", &TransOutService{}, RegisterTransOutServiceServer)
    }()

    // 启动转入服务
    go func() {
        startGrpcServer("8082", &TransInService{}, RegisterTransInServiceServer)
    }()
    time.Sleep(time.Second) // 确保服务启动
    // 初始化 DTM 客户端
    dtmServer := "localhost:36800" // DTM 服务器地址
    conn, err := grpc.Dial(dtmServer, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
    if err != nil {
        log.Fatalf("Failed to connect to DTM server: %v", err)
    }
    defer conn.Close()

    // 创建 DTM 客户端
    dtmClient := NewDtmClient(conn)

    // 创建 TCC 事务
    gid := dtmimp.GetNewGid()
    tcc := dtmcli.NewTCC(dtmClient, gid).
        Add(TransOutPrepareURL, TransOutCommitURL, TransOutRollbackURL, &TransRequest{Amount: 10}).
        Add(TransInPrepareURL, TransInCommitURL, TransInRollbackURL, &TransRequest{Amount: 10})

    // 提交 TCC 事务
    err = tcc.Call(func(tcc *dtmcli.TCC) error {
        return nil // 成功回调，这里可以添加一些额外的业务逻辑
    })
    if err != nil {
        log.Fatalf("TCC transaction failed: %v", err)
    }

    log.Printf("TCC transaction completed successfully with GID: %s", gid)
}

这段代码演示了如何使用DTM框架的TCC模式来实现一个跨服务的转账操作。你需要分别实现转出服务和转入服务的Prepare、Commit、Rollback三个阶段的操作，并在DTM框架中注册这些操作。

如何处理分布式事务中的网络异常？

网络异常是分布式事务中最常见的问题之一。为了解决这个问题，我们需要采取一些容错措施，例如：

重试机制： 在服务调用失败时，进行重试，避免因为网络抖动等临时性错误导致事务失败。
超时机制： 为每个服务调用设置超时时间，避免因为服务长时间无响应导致事务阻塞。
幂等性： 确保所有操作都是幂等的，即多次执行的结果和执行一次的结果相同。这样即使因为网络异常导致操作被重复执行，也不会影响数据的一致性。
补偿机制： 如果事务最终失败，需要进行补偿操作，撤销之前成功的操作，从而保证数据的一致性。

这些容错措施可以有效地提高分布式事务的可靠性。

分布式事务的性能优化策略

分布式事务的性能通常比单机事务要差，因此我们需要采取一些优化策略来提高性能。

减少事务的参与者： 事务的参与者越多，事务的开销就越大。因此，我们应该尽量减少事务的参与者，将相关的操作放在同一个服务中执行。
使用异步提交： 对于一些对一致性要求不高的场景，可以使用异步提交的方式来提高性能。异步提交是指在事务提交后，立即返回给客户端，然后由后台线程异步地将事务提交到数据库。
使用本地事务： 尽量使用本地事务，避免使用全局事务。全局事务的开销比本地事务要大得多。
优化数据库： 优化数据库的性能，例如使用索引、分区等技术，可以提高事务的执行速度。

这些优化策略可以有效地提高分布式事务的性能。选择合适的事务模式，并结合具体的业务场景进行优化，才能达到最佳的性能。

怎样在SQL Server中用表值参数TVP实现高性能批量修改？

怎样在SQL存储过程中动态决定是否开启显式事务

如何处理Golang API版本升级

Go语言文件写入时路径错误的排查方法

Golang分布式任务怎么在Kubernetes中完成高可用部署