术语
Command
该命令是一个简单的数据结构，表示执行某些操作的请求。
Command Bus
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/command_bus.go
// ...
// CommandBus 将命令(commands)传输到命令处理程序(command handlers)。
type CommandBus struct {
// ...
Command Processor
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/command_processor.go
// ...
// CommandProcessor 决定哪个 CommandHandler 应该处理这个命令
received from the command bus.
type CommandProcessor struct {
// ...
Command Handler
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/command_processor.go
// ...
// CommandHandler 接收由 NewCommand 定义的命令，并使用 Handle 方法处理它。
// 如果使用 DDD, CommandHandler 可以修改并持久化聚合。
//
// 与 EvenHandler 相反，每个命令必须只有一个 CommandHandler。
//
// 在处理消息期间使用 CommandHandler 的一个实例。
// 当同时发送多个命令时，Handle 方法可以同时执行多次。
// 因此，Handle 方法必须是线程安全的！
type CommandHandler interface {
// ...
Event
该事件表示已经发生的事情。 事件是不可变的。
Event Bus
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/event_bus.go
// ...
// EventBus 将事件传输到事件处理程序。
type EventBus struct {
// ...
Event Processor
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/event_processor.go
// ...
// EventProcessor 确定哪个 EventHandler 应该处理从事件总线接收到的事件。
type EventProcessor struct {
// ...
Event Handler
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/event_processor.go
// ...
// EventHandler 接收由 NewEvent 定义的事件，并使用其 Handle 方法对其进行处理。
// 如果使用 DDD，CommandHandler 可以修改并保留聚合。
// 它还可以调用流程管理器、saga 或仅仅构建一个读取模型。
// 与 CommandHandler 相比，每个 Event 可以具有多个 EventHandler。
//
// 在处理消息时使用 EventHandler 的一个实例。
// 当同时传递多个事件时，Handle 方法可以同时执行多次。
// 因此，Handle 方法必须是线程安全的！
type EventHandler interface {
// ...
CQRS Facade
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/cqrs.go
// ...
// Facade 是用于创建 Command 和 Event buses 及 processors 的 facade。
// 创建它是为了在以标准方式使用 CQRS 时避免使用 boilerplate。
// 您还可以手动创建 buses 和 processors，并从 NewFacade 中获得灵感。
type Facade struct {
// ...
Command and Event Marshaler
完整源码：
github.com/ThreeDotsLabs/watermill/components/cqrs/marshaler.go
// ...
// 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
}
// ...
用法
Example domain（领域模型示例）
作为示例，我们将使用一个简单的 domain，它负责处理酒店的房间预订。
我们将使用 Event Storming 表示法来展示这个 domain 的模型。
Legend:
blue（蓝色）便利贴是命令
orange（橙色）便利贴是事件
green（绿色）便利贴是读取模型，从事件异步生成
violet（紫色）便利贴是策略，由事件触发并产生命令
pink（粉色）便利贴是热点（hot-spots）；我们标记经常发生问题的地方
domain（领域模型）很简单：
客人可以预订房间（book a room）。
每当预订房间时，我们都会为客人订购啤酒（Whenever a room is booked, we order a beer）（因为我们爱客人）。
我们知道有时候啤酒不够（not enough beers）。
我们根据预订生成财务报告（financial report）。
Sending a command（发送命令）
首先，我们需要模拟访客的动作。
完整源码：
github.com/ThreeDotsLabs/watermill/_examples/basic/5-cqrs-protobuf/main.go
// ...
bookRoomCmd := &BookRoom{
RoomId: fmt.Sprintf("%d", i),
GuestName: "John",
StartDate: startDate,
EndDate: endDate,
}
if err := commandBus.Send(context.Background(), bookRoomCmd); err != nil {
panic(err)
}
// ...
Command handler
BookRoomHandler 将处理我们的命令。
完整源码：
github.com/ThreeDotsLabs/watermill/_examples/basic/5-cqrs-protobuf/main.go
// ...
// 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 命令。
// ...
Event handler
如前所述，我们希望每次预订房间时都点一杯啤酒（“每次预订房间时”便笺）。我们通过使 OrderBeer 命令来实现。
完整源码：
github.com/ThreeDotsLabs/watermill/_examples/basic/5-cqrs-protobuf/main.go
// ...
// 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。
// ...
OrderBeerHandler 与 BookRoomHandler 非常相似。唯一的区别是，当啤酒不够时，它有时会返回一个错误，这将导致重新交付命令。您可以在示例源代码中找到整个实现。
使用事件处理程序构建读取模型
完整源码：
github.com/ThreeDotsLabs/watermill/_examples/basic/5-cqrs-protobuf/main.go
// ...
// 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() {
// ...
将其连接起来——CQRS facade
我们拥有构建 CQRS 应用程序的所有块。 现在，我们需要使用某种胶水将其连接起来。
我们将使用最简单的内存消息传递基础设施: GoChannel。
在后台，CQRS 正在使用 Watermill 的消息路由器。 如果您不熟悉它，并且想了解它的工作原理，则应查阅《入门指南》。 它还将向您展示如何使用一些标准的消息传递模式，例如 metrics，poison queue，throttling，correlation 以及每个消息驱动的应用程序使用的其他工具。那些内置于 Watermill 中。
让我们回到 CQRS。如您所知，CQRS 是由多个组件构建的，如命令（Command）或事件总线（Event buses）、处理程序（handlers）、处理器（processors）等。为了简化所有这些构建块的创建，我们创建了 cqrs.Facade，它创建所有这些。
完整源码：
github.com/ThreeDotsLabs/watermill/_examples/basic/5-cqrs-protobuf/main.go
// ...
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)
}
}
// ...
就这样。 我们有一个正在运行的 CQRS 应用程序。
龙华大道1号 http://www.kinghill.cn/Dynamics/2106.html
有疑问加站长微信联系（非本文作者）
