Golang事件驱动架构中的中介者模式和简化CQRS模式处理插件库Go-MediatR的使用

这个包是Golang中的中介者模式(Mediator Pattern)实现，灵感来源于.NET中优秀的jbogard/mediatr库。

安装

go get github.com/mehdihadeli/go-mediatr

特性

✅ 处理Request/Response消息，将消息传递给单个处理程序（命令、查询）

✅ 处理Notification消息，将消息传递给多个处理程序（事件）

✅ 管道行为(Pipeline Behaviours)用于在执行处理程序前后处理一些横切关注点

策略

MediatR有两种消息分发策略：

1. Request/Response消息，分发给单个处理程序
2. Notification消息，分发给所有（多个）处理程序，它们没有任何响应

Request/Response策略

request/response消息只有一个处理程序，可以在CQRS模式中处理命令和查询场景。

创建Request/Response消息

创建一个请求（命令或查询），它只有一个处理程序，我们可以创建一个命令消息或查询消息作为request：

// Command (Request)
type CreateProductCommand struct {
    ProductID   uuid.UUID `validate:"required"`
    Name        string    `validate:"required,gte=0,lte=255"`
    Description string    `validate:"required,gte=0,lte=5000"`
    Price       float64   `validate:"required,gte=0"`
    CreatedAt   time.Time `validate:"required"`
}

// Query (Request)
type GetProdctByIdQuery struct {
    ProductID uuid.UUID `validate:"required"`
}

对于这些请求的响应，我们可以创建响应消息作为response：

// Command (Response)
type CreateProductCommandResponse struct {
    ProductID uuid.UUID `json:"productId"`
}

// Query (Response)
type GetProductByIdQueryResponse struct {
    ProductID   uuid.UUID `json:"productId"`
    Name        string    `json:"name"`
    Description string    `json:"description"`
    Price       float64   `json:"price"`
    CreatedAt   time.Time `json:"createdAt"`
}

创建Request处理程序

为了处理我们的请求，我们应该为每个请求创建一个单一请求处理程序。每个处理程序应该实现RequestHandler接口。

type RequestHandler[TRequest any, TResponse any] interface {
    Handle(ctx context.Context, request TRequest) (TResponse, error)
}

这里我们为请求创建请求处理程序（命令处理程序和查询处理程序），实现上述接口：

// Command Handler
type CreateProductCommandHandler struct {
    productRepository *repository.InMemoryProductRepository
}

func NewCreateProductCommandHandler(productRepository *repository.InMemoryProductRepository) *CreateProductCommandHandler {
    return &CreateProductCommandHandler{productRepository: productRepository}
}

func (c *CreateProductCommandHandler) Handle(ctx context.Context, command *CreateProductCommand) (*creatingProductDtos.CreateProductCommandResponse, error) {
    product := &models.Product{
        ProductID:   command.ProductID,
        Name:        command.Name,
        Description: command.Description,
        Price:       command.Price,
        CreatedAt:   command.CreatedAt,
    }

    createdProduct, err := c.productRepository.CreateProduct(ctx, product)
    if err != nil {
        return nil, err
    }

    response := &creatingProductDtos.CreateProductCommandResponse{ProductID: createdProduct.ProductID}

    return response, nil
}

// Query Handler
type GetProductByIdQueryHandler struct {
    productRepository *repository.InMemoryProductRepository
}

func NewGetProductByIdQueryHandler(productRepository *repository.InMemoryProductRepository) *GetProductByIdQueryHandler {
    return &GetProductByIdQueryHandler{productRepository: productRepository}
}

func (c *GetProductByIdQueryHandler) Handle(ctx context.Context, query *GetProductByIdQuery) (*gettingProductDtos.GetProdctByIdQueryResponse, error) {
    product, err := c.productRepository.GetProductById(ctx, query.ProductID)
    if err != nil {
        return nil, err
    }

    response := &gettingProductDtos.GetProdctByIdQueryResponse{
        ProductID:   product.ProductID,
        Name:        product.Name,
        Description: product.Description,
        Price:       product.Price,
        CreatedAt:   product.CreatedAt,
    }

    return response, nil
}

注意：在我们不需要从请求处理程序获得响应的情况下，可以使用Unit类型，实际上是一个空结构体。

将Request处理程序注册到MediatR

在发送或分发我们的请求之前，我们应该将请求处理程序注册到MediatR。

这里我们将请求处理程序（命令处理程序和查询处理程序）注册到MediatR：

// Registering `createProductCommandHandler` request handler for `CreateProductCommand` request to the MediatR
mediatr.RegisterHandler[*creatingProduct.CreateProductCommand, *creatingProductsDtos.CreateProductCommandResponse](createProductCommandHandler)

// Registering `getProductByIdQueryHandler` request handler for `GetProductByIdQuery` request to the MediatR
mediatr.RegisterHandler[*gettingProduct.GetProductByIdQuery, *gettingProductDtos.GetProdctByIdQueryResponse](getProductByIdQueryHandler)

发送Request到MediatR

最后，通过中介发送消息。

这里我们将请求发送到MediatR，将它们分发给请求处理程序（命令处理程序和查询处理程序）：

// Sending `CreateProductCommand` request to mediatr for dispatching to the `CreateProductCommandHandler` request handler
command := &CreateProductCommand{
    ProductID:   uuid.NewV4(),
    Name:        request.name,
    Description: request.description,
    Price:       request.price,
    CreatedAt:   time.Now(),
}

mediatr.Send[*CreateProductCommand, *creatingProductsDtos.CreateProductCommandResponse](ctx, command)

// Sending `GetProductByIdQuery` request to mediatr for dispatching to the `GetProductByIdQueryHandler` request handler
query := &GetProdctByIdQuery{
    ProductID:   uuid.NewV4()
}

mediatr.Send[*GetProductByIdQuery, *gettingProductsDtos.GetProductByIdQueryResponse](ctx, query)

Notification策略

notification消息可以有多个处理程序，并且没有任何响应，它可以处理事件驱动架构中的事件通知或通知。

创建Notification消息

创建一个通知（事件），它有多个处理程序并且没有任何响应，我们可以创建一个事件通知作为notification：

// Event (Notification)
type ProductCreatedEvent struct {
    ProductID uuid.UUID   `json:"productId"`
    Name        string    `json:"name"`
    Description string    `json:"description"`
    Price       float64   `json:"price"`
    CreatedAt   time.Time `json:"createdAt"`
}

这个事件没有任何响应。

创建Notification处理程序

为了处理我们的通知，我们可以为每个通知事件创建多个通知处理程序。每个处理程序应该实现NotificationHandler接口。

type NotificationHandler[TNotification any] interface {
    Handle(ctx context.Context, notification TNotification) error
}

这里我们为通知创建多个通知事件处理程序，实现上述接口：

// Notification Event Handler1
type ProductCreatedEventHandler1 struct {
}

func (c *ProductCreatedEventHandler1) Handle(ctx context.Context, event *ProductCreatedEvent) error {
//Do something with the event here !
    return nil
}

// Notification Event Handler2
type ProductCreatedEventHandler2 struct {
}

func (c *ProductCreatedEventHandler2) Handle(ctx context.Context, event *ProductCreatedEvent) error {
//Do something with the event here !
    return nil
}

将Notification处理程序注册到MediatR

在发布我们的通知之前，我们应该将通知处理程序注册到MediatR。

这里我们将通知处理程序注册到MediatR：

// Registering `notificationHandler1`, `notificationHandler2` notification handler for `ProductCreatedEvent` notification event to the MediatR
notificationHandler1 := &ProductCreatedEventHandler1{}
notificationHandler2 := &ProductCreatedEventHandler2{}

mediatr.RegisterNotificationHandlers[*events.ProductCreatedEvent](notificationHandler1, notificationHandler2)

发布Notification到MediatR

最后，通过中介发布通知事件。

这里我们将通知发布到MediatR，将它们分发给通知处理程序：

// Publishing `ProductCreatedEvent` notification to mediatr for dispatching to the `ProductCreatedEventHandler1`, `ProductCreatedEventHandler2` notification handlers
productCreatedEvent := 	&ProductCreatedEvent {
    ProductID:   createdProduct.ProductID,
    Name:        createdProduct.Name,
    Price:       createdProduct.Price,
    CreatedAt:   createdProduct.CreatedAt,
    Description: createdProduct.Description,
}

mediatr.Publish[*events.ProductCreatedEvent](ctx, productCreatedEvent)

使用管道行为(Pipeline Behaviors)

有时我们需要在运行请求处理程序之前或之后添加一些横切关注点，如日志记录、指标、断路器、重试等。在这种情况下，我们可以使用PipelineBehavior。它实际上就像一个中间件或装饰器模式。

这些行为将在调用mediatr上的Send方法发送请求时，在执行请求处理程序之前或之后执行。

创建管道行为

为了创建一个管道行为，我们应该实现PipelineBehavior接口：

type PipelineBehavior interface {
    Handle(ctx context.Context, request interface{}, next RequestHandlerFunc) (interface{}, error)
}

request参数是通过mediatr的Send方法传入的请求对象，而next参数是行为链中下一个动作的延续，其类型为RequestHandlerFunc。

这是一个管道行为的例子：

type RequestLoggerBehaviour struct {
}

func (r *RequestLoggerBehaviour) Handle(ctx context.Context, request interface{}, next mediatr.RequestHandlerFunc) (interface{}, error) {
    log.Printf("logging some stuff before handling the request")

    response, err := next()
    if err != nil {
        return nil, err
    }

    log.Println("logging some stuff after handling the request")

    return response, nil
}

在我们定义的行为中，我们需要调用next参数，它会调用行为链中的下一个动作，如果没有其他行为，next将调用我们的实际请求处理程序并返回响应。我们可以在调用行为链中的下一个动作之前或之后做一些事情。

将管道行为注册到MediatR

为了将我们的管道行为注册到MediatR，我们应该使用RegisterPipelineBehaviors方法：

loggerPipeline := &behaviours.RequestLoggerBehaviour{}
err = mediatr.RegisterRequestPipelineBehaviors(loggerPipeline)