利用 Watermill 实现 Golang CQRS

// ...
// CommandBus 将命令(commands)传输到命令处理程序(command handlers)。
type CommandBus struct {
// ...

// ...
// CommandProcessor 决定哪个 CommandHandler 应该处理这个命令
received from the command bus.
type CommandProcessor struct {
// ...

// ...
// CommandHandler 接收由 NewCommand 定义的命令，并使用 Handle 方法处理它。
// 如果使用 DDD, CommandHandler 可以修改并持久化聚合。
//
// 与 EvenHandler 相反，每个命令必须只有一个 CommandHandler。
//
// 在处理消息期间使用 CommandHandler 的一个实例。
// 当同时发送多个命令时，Handle 方法可以同时执行多次。
// 因此，Handle 方法必须是线程安全的！
type CommandHandler interface {
// ...

// ...
// EventBus 将事件传输到事件处理程序。
type EventBus struct {
// ...

// ...
// EventProcessor 确定哪个 EventHandler 应该处理从事件总线接收到的事件。
type EventProcessor struct {
// ...

// ...
// EventHandler 接收由 NewEvent 定义的事件，并使用其 Handle 方法对其进行处理。
// 如果使用 DDD，CommandHandler 可以修改并保留聚合。
// 它还可以调用流程管理器、saga 或仅仅构建一个读取模型。
// 与 CommandHandler 相比，每个 Event 可以具有多个 EventHandler。
//
// 在处理消息时使用 EventHandler 的一个实例。
// 当同时传递多个事件时，Handle 方法可以同时执行多次。
// 因此，Handle 方法必须是线程安全的！
type EventHandler interface {
// ...

// ...
// Facade 是用于创建 Command 和 Event buses 及 processors 的 facade。
// 创建它是为了在以标准方式使用 CQRS 时避免使用 boilerplate。
// 您还可以手动创建 buses 和 processors，并从 NewFacade 中获得灵感。
type Facade struct {
// ...

// ...
// CommandEventMarshaler 将命令和事件 marshal 给 Watermill 的消息，反之亦然。
// 该命令的有效载荷需要 marshal 至 []bytes。
type CommandEventMarshaler interface {
    // Marshal marshal 命令或事件给 Watermill 的消息。
   Marshal(v interface{}) (*message.Message, error)

    // Unmarshal Unmarshal watermill的信息给 v Command 或 Event。
   Unmarshal(msg *message.Message, v interface{}) (err error)

   // Name 返回命令或事件的名称。
   // Name 用于确定接收到的命令或事件是我们想要处理的事件。
   Name(v interface{}) string

   // NameFromMessage 从 Watermill 的消息(由 Marshal 生成)中返回命令或事件的名称。
   // 
   //
   // 当我们将 Command 或 Event marshal 到 Watermill 的消息中时，
   // 我们应该使用 NameFromMessage 而不是 Name 来避免不必要的 unmarshaling。
   NameFromMessage(msg *message.Message) string
}
// ...

// ...
       bookRoomCmd := &BookRoom{
            RoomId:    fmt.Sprintf("%d", i),
            GuestName: "John",
            StartDate: startDate,
            EndDate:   endDate,
        }
        if err := commandBus.Send(context.Background(), bookRoomCmd); err != nil {
            panic(err)
        }
// ...

// ...
// BookRoomHandler 是一个命令处理程序，它处理 BookRoom 命令并发出 RoomBooked。
//
// 在 CQRS 中，一个命令只能由一个处理程序处理。
// 将具有此命令的另一个处理程序添加到命令处理器时，将返回错误。
type BookRoomHandler struct {
    eventBus *cqrs.EventBus
}

func (b BookRoomHandler) HandlerName() string {
    return "BookRoomHandler"
}

// NewCommand 返回该 handle 应该处理的命令类型。它必须是一个指针。
func (b BookRoomHandler) NewCommand() interface{} {
    return &BookRoom{}
}

func (b BookRoomHandler) Handle(ctx context.Context, c interface{}) error {
    // c 始终是 `NewCommand` 返回的类型，因此强制转换始终是安全的
   cmd := c.(*BookRoom)

   // 一些随机的价格，在生产中你可能会用更明智的方式计算
   price := (rand.Int63n(40) + 1) * 10

    log.Printf(
        "Booked %s for %s from %s to %s",
        cmd.RoomId,
        cmd.GuestName,
        time.Unix(cmd.StartDate.Seconds, int64(cmd.StartDate.Nanos)),
        time.Unix(cmd.EndDate.Seconds, int64(cmd.EndDate.Nanos)),
    )

   // RoomBooked 将由 OrderBeerOnRoomBooked 事件处理程序处理，
   // 将来，RoomBooked 可能由多个事件处理程序处理
   if err := b.eventBus.Publish(ctx, &RoomBooked{
        ReservationId: watermill.NewUUID(),
        RoomId:        cmd.RoomId,
        GuestName:     cmd.GuestName,
        Price:         price,
        StartDate:     cmd.StartDate,
        EndDate:       cmd.EndDate,
    }); err != nil {
        return err
    }

    return nil
}

// OrderBeerOnRoomBooked 是事件处理程序，它处理 RoomBooked 事件并发出 OrderBeer 命令。
// ...

// ...
// OrderBeerOnRoomBooked 是事件处理程序，它处理 RoomBooked 事件并发出 OrderBeer 命令。
type OrderBeerOnRoomBooked struct {
    commandBus *cqrs.CommandBus
}

func (o OrderBeerOnRoomBooked) HandlerName() string {
   // 此名称传递给 EventsSubscriberConstructor 并用于生成队列名称
   return "OrderBeerOnRoomBooked"
}

func (OrderBeerOnRoomBooked) NewEvent() interface{} {
    return &RoomBooked{}
}

func (o OrderBeerOnRoomBooked) Handle(ctx context.Context, e interface{}) error {
    event := e.(*RoomBooked)

    orderBeerCmd := &OrderBeer{
        RoomId: event.RoomId,
        Count:  rand.Int63n(10) + 1,
    }

    return o.commandBus.Send(ctx, orderBeerCmd)
}

// OrderBeerHandler 是命令处理程序，它处理 OrderBeer 命令并发出 BeerOrdered。
// ...

// ...
// BookingsFinancialReport 是一个读取模型，用于计算我们可以从预订中赚取多少钱。
// 与 OrderBeerOnRoomBooked 一样，它侦听 RoomBooked 事件。
//
// 此实现只是写入内存。在生产中，您可能会使用一些持久性存储。
type BookingsFinancialReport struct {
    handledBookings map[string]struct{}
    totalCharge     int64
    lock            sync.Mutex
}

func NewBookingsFinancialReport() *BookingsFinancialReport {
    return &BookingsFinancialReport{handledBookings: map[string]struct{}{}}
}

func (b BookingsFinancialReport) HandlerName() string {
   // 此名称传递给 EventsSubscriberConstructor 并用于生成队列名称
   return "BookingsFinancialReport"
}

func (BookingsFinancialReport) NewEvent() interface{} {
    return &RoomBooked{}
}

func (b *BookingsFinancialReport) Handle(ctx context.Context, e interface{}) error {
   // Handle 可以被并发调用，因此它必须是线程安全的。
   b.lock.Lock()
    defer b.lock.Unlock()

    event := e.(*RoomBooked)

   // 当我们使用不提供一次精确交付语义的 Pub/Sub 时，我们需要对消息进行重复数据删除。
   // GoChannel Pub/Sub 提供了精确的一次交付，
   // 但是让我们为其他 Pub/Sub 实现准备好这个示例。
   if _, ok := b.handledBookings[event.ReservationId]; ok {
        return nil
    }
    b.handledBookings[event.ReservationId] = struct{}{}

    b.totalCharge += event.Price

    fmt.Printf(">>> Already booked rooms for $%d\n", b.totalCharge)
    return nil
}

var amqpAddress = "amqp://guest:guest@rabbitmq:5672/"

func main() {
// ...

// ...
func main() {
    logger := watermill.NewStdLogger(false, false)
    cqrsMarshaler := cqrs.ProtobufMarshaler{}

    // 您可以从此处使用任何 Pub/Sub 实现：https://watermill.io/docs/pub-sub-implementations/
   // 详细的 RabbitMQ 实现: https://watermill.io/docs/pub-sub-implementations/#rabbitmq-amqp
   // 命令将被发送到队列，因为它们需要被使用一次。
   commandsAMQPConfig := amqp.NewDurableQueueConfig(amqpAddress)
    commandsPublisher, err := amqp.NewPublisher(commandsAMQPConfig, logger)
    if err != nil {
        panic(err)
    }
    commandsSubscriber, err := amqp.NewSubscriber(commandsAMQPConfig, logger)
    if err != nil {
        panic(err)
    }

   // 事件将被发布到配置了 PubSu b的 Rabbit，因为它们可能被多个使用者使用。
   // (在这种情况下，BookingsFinancialReport 和 OrderBeerOnRoomBooked).
   eventsPublisher, err := amqp.NewPublisher(amqp.NewDurablePubSubConfig(amqpAddress, nil), logger)
    if err != nil {
        panic(err)
    }

    // CQRS 建立在消息路由器上。详细文档：https://watermill.io/docs/messages-router/
   router, err := message.NewRouter(message.RouterConfig{}, logger)
    if err != nil {
        panic(err)
    }

   // 简单的中间件，可以从事件或命令处理程序中 recover panics。
   // 您可以在文档中找到有关路由器中间件的更多信息：
   // https://watermill.io/docs/messages-router/#middleware
   //
   // 您可以在 message/router/middleware 中找到可用的中间件列表。
   router.AddMiddleware(middleware.Recoverer)

    // cqrs.Facade是命令和事件总线与处理器的 facade。
   // 您可以使用 facade，或者手动创建总线和处理器(您可以使用 cqrs.NewFacade 激发灵感)
   cqrsFacade, err := cqrs.NewFacade(cqrs.FacadeConfig{
        GenerateCommandsTopic: func(commandName string) string {
            // 我们正在使用RabbitMQ队列配置，因此我们需要按命令类型指定主题 topic
           return commandName
        },
        CommandHandlers: func(cb *cqrs.CommandBus, eb *cqrs.EventBus) []cqrs.CommandHandler {
            return []cqrs.CommandHandler{
                BookRoomHandler{eb},
                OrderBeerHandler{eb},
            }
        },
        CommandsPublisher: commandsPublisher,
        CommandsSubscriberConstructor: func(handlerName string) (message.Subscriber, error) {
            // 我们可以重用订阅者(subscriber)，因为所有命令都有各自的主题(topics)
           return commandsSubscriber, nil
        },
        GenerateEventsTopic: func(eventName string) string {
            // 因为我们使用的是PubSub RabbitMQ配置，所以我们可以对所有事件使用一个主题(topic)
           return "events"

            // 我们还可以按事件类型使用主题(topic)
           // return eventName
       },
        EventHandlers: func(cb *cqrs.CommandBus, eb *cqrs.EventBus) []cqrs.EventHandler {
            return []cqrs.EventHandler{
                OrderBeerOnRoomBooked{cb},
                NewBookingsFinancialReport(),
            }
        },
        EventsPublisher: eventsPublisher,
        EventsSubscriberConstructor: func(handlerName string) (message.Subscriber, error) {
            config := amqp.NewDurablePubSubConfig(
                amqpAddress,
                amqp.GenerateQueueNameTopicNameWithSuffix(handlerName),
            )

            return amqp.NewSubscriber(config, logger)
        },
        Router:                router,
        CommandEventMarshaler: cqrsMarshaler,
        Logger:                logger,
    })
    if err != nil {
        panic(err)
    }

    // 每秒发布 BookRoom 命令以模拟传入流量
   go publishCommands(cqrsFacade.CommandBus())

    // 处理器(processors)是基于路由器(router)的，所以当路由器启动时，处理器就会工作
   if err := router.Run(context.Background()); err != nil {
        panic(err)
    }
}
// ...