微服务中的事务：SAGA 模式与编排的一部分-Golang-PHP中文网

首页

后端开发

Golang

微服务中的事务：SAGA 模式与编排的一部分

Barbara Streisand

Jan 23, 2025 am 02:05 AM

在本系列的第一篇文章中，我们介绍了SAGA 模式，并演示了最小的编排如何使用中央编排器管理分布式事务。

让我们面对现实吧！这次，我们将深入探讨编排方法，其中服务通过自主发出和消费事件来协调工作流程。

为了使其切实可行，我们将使用 Go 和 RabbitMQ 实现多服务医疗保健工作流程。每个服务都有自己的 main.go，使其易于扩展、测试和独立运行。

什么是SAGA编排？

编排依赖于去中心化的沟通。每个服务都会侦听事件并通过发出新事件来触发后续步骤。没有中央协调者；该流程源自各个服务的交互。

主要优点：

解耦服务：每个服务独立运行。
可扩展性：事件驱动系统有效地处理高负载。
灵活性：添加新服务不需要更改工作流程逻辑。

挑战：

调试复杂性：跨多个服务跟踪事件可能很棘手。（我会写一篇专门讨论这个话题的文章，敬请期待！）
基础设施设置：服务需要强大的消息代理（例如 RabbitMQ）来连接所有点。
事件风暴：设计不当的工作流程可能会导致系统因事件而不堪重负。

实例：医疗保健工作流程

让我们回顾一下第一篇文章中的医疗保健工作流程：

患者服务：验证患者详细信息和保险范围。
调度服务： 安排程序。
库存服务：储备医疗用品。
计费服务：处理计费。

每项服务将：

使用 RabbitMQ 监听特定事件。
发出新事件来触发后续步骤。

使用 Docker 设置 RabbitMQ

我们将使用 RabbitMQ 作为事件队列。使用 Docker 在本地运行它：

docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

登录后复制

访问RabbitMQ管理界面：http://localhost:15672（用户名：guest，密码：guest）。

交换、队列和绑定设置

我们需要配置 RabbitMQ 来适应我们的事件。以下是用于设置 RabbitMQ 基础设施的 init.go 文件示例：

package main

import (
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    err = ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare an exchange: %v", err)
    }

    _, err = ch.QueueDeclare("PatientVerified", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    err = ch.QueueBind("PatientVerified", "PatientVerified", "events", false, nil)
    if err != nil {
        log.Fatalf("Failed to bind a queue: %v", err)
    }
}

登录后复制

完整代码在这里！

注意：在生产环境中，您可能希望使用 GitOps 方法（例如，使用 Terraform）来管理此设置，或者让每个服务动态处理自己的队列。

实施：服务文件

每个服务都有自己的 main.go。我们还将包括优雅地处理失败的补偿措施。

1.病人服务

该服务验证患者详细信息并发出 PatientVerified 事件。如果发生下游故障，它还会通过通知患者进行补偿。

docker run --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:4.0.5-management

登录后复制

2.调度服务

此服务侦听 PatientVerified 并发出 procedureScheduled。如果发生下游故障，它会通过取消程序来进行补偿。

package main

import (
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    err = ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare an exchange: %v", err)
    }

    _, err = ch.QueueDeclare("PatientVerified", true, false, false, false, nil)
    if err != nil {
        log.Fatalf("Failed to declare a queue: %v", err)
    }

    err = ch.QueueBind("PatientVerified", "PatientVerified", "events", false, nil)
    if err != nil {
        log.Fatalf("Failed to bind a queue: %v", err)
    }
}

登录后复制

附加服务

包括库存服务和计费服务实现，遵循与上面相同的结构。每个服务都会监听前一个事件并发出下一个事件，确保针对失败的补偿逻辑到位。

完整代码在这里！

运行工作流程

启动 RabbitMQ：

// patient/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[PatientService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "ProcedureScheduleCancelled")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[PatientService] Processing event: ProcedureScheduleCancelled")
            if err := notifyProcedureScheduleCancellation(); err != nil {
                log.Fatalf("Failed to notify patient: %v", err)
            }
        }
    }()

    common.PublishEvent(ch, "events", "PatientVerified", "Patient details verified")
    fmt.Println("[PatientService] Event published: PatientVerified")

    select {}
}

func notifyProcedureScheduleCancellation() error {
    fmt.Println("Compensation: Notify patient of procedure cancellation.")
    return nil
}

登录后复制

运行每个服务：
打开单独的终端并运行：

// scheduler/main.go
package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
    "github.com/thegoodapi/saga_tutorial/choreography/common"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    go func() {
        fmt.Println("[SchedulerService] Waiting for events...")
        msgs, err := common.ConsumeEvent(ch, "PatientVerified")
        if err != nil {
            log.Fatalf("Failed to consume event: %v", err)
        }

        for range msgs {
            fmt.Println("[SchedulerService] Processing event: PatientVerified")
            if err := scheduleProcedure(); err != nil {
                common.PublishEvent(ch, "events", "ProcedureScheduleFailed", "Failed to schedule procedure")
                fmt.Println("[SchedulerService] Compensation triggered: ProcedureScheduleFailed")
            } else {
                common.PublishEvent(ch, "events", "ProcedureScheduled", "Procedure scheduled successfully")
                fmt.Println("[SchedulerService] Event published: ProcedureScheduled")
            }
        }
    }()

    select {}
}

func scheduleProcedure() error {
    fmt.Println("Step 2: Scheduling procedure...")
    return nil // or simulate a failure
}

登录后复制

观察输出：
每个服务按顺序处理事件，记录工作流程进度。