Go-CommandBus: 轻量级可插拔命令总线实现

Go-CommandBus 是一个轻量级的命令总线实现，它允许你将命令与处理程序解耦，支持中间件和插件机制。下面我将详细介绍其使用方法和示例代码。

基本概念

命令总线模式包含三个主要部分：

1. Command - 包含执行操作所需数据的结构体
2. CommandHandler - 处理特定命令的函数
3. CommandBus - 负责将命令路由到相应处理程序的中央调度器

安装

go get github.com/vardius/go-commandbus

基本使用示例

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/vardius/go-commandbus"
)

// 定义命令
type CreateUserCommand struct {
	ID       string
	Username string
	Email    string
}

type DeleteUserCommand struct {
	ID string
}

func main() {
	// 创建命令总线实例
	bus := commandbus.New()

	// 注册命令处理程序
	if err := bus.Register(
		"CreateUserCommand", // 命令名称
		func(ctx context.Context, command interface{}) error {
			// 类型断言获取具体命令
			cmd := command.(CreateUserCommand)
			
			fmt.Printf("创建用户: ID=%s, Username=%s, Email=%s\n", 
				cmd.ID, cmd.Username, cmd.Email)
			
			// 这里可以添加实际业务逻辑
			return nil
		},
	); err != nil {
		log.Fatal(err)
	}

	if err := bus.Register(
		"DeleteUserCommand",
		func(ctx context.Context, command interface{}) error {
			cmd := command.(DeleteUserCommand)
			fmt.Printf("删除用户: ID=%s\n", cmd.ID)
			return nil
		},
	); err != nil {
		log.Fatal(err)
	}

	// 发送命令
	ctx := context.Background()
	
	// 创建用户命令
	if err := bus.Dispatch(ctx, "CreateUserCommand", CreateUserCommand{
		ID:       "123",
		Username: "john_doe",
		Email:    "john@example.com",
	}); err != nil {
		log.Fatal(err)
	}

	// 删除用户命令
	if err := bus.Dispatch(ctx, "DeleteUserCommand", DeleteUserCommand{
		ID: "123",
	}); err != nil {
		log.Fatal(err)
	}
}

中间件支持

Go-CommandBus 支持中间件，可以在命令处理前后执行额外逻辑：

func loggingMiddleware(next commandbus.HandlerFunc) commandbus.HandlerFunc {
	return func(ctx context.Context, command interface{}) error {
		// 命令处理前
		fmt.Printf("处理命令前: %T\n", command)
		
		// 调用下一个中间件或处理程序
		err := next(ctx, command)
		
		// 命令处理后
		if err != nil {
			fmt.Printf("处理命令出错: %v\n", err)
		} else {
			fmt.Println("命令处理成功")
		}
		
		return err
	}
}

func main() {
	bus := commandbus.New(
		commandbus.WithMiddleware(loggingMiddleware),
	)
	
	// 注册处理程序和发送命令的代码同上...
}

插件机制

Go-CommandBus 支持插件机制，可以扩展功能：

type MetricsPlugin struct{}

func (p *MetricsPlugin) OnCommandReceived(ctx context.Context, command interface{}) {
	fmt.Println("收到命令:", command)
}

func (p *MetricsPlugin) OnCommandDispatched(ctx context.Context, command interface{}) {
	fmt.Println("命令已分发:", command)
}

func (p *MetricsPlugin) OnCommandHandled(ctx context.Context, command interface{}, err error) {
	if err != nil {
		fmt.Println("命令处理失败:", command, "错误:", err)
	} else {
		fmt.Println("命令处理成功:", command)
	}
}

func main() {
	bus := commandbus.New(
		commandbus.WithPlugin(&MetricsPlugin{}),
	)
	
	// 注册处理程序和发送命令的代码同上...
}

高级特性

1. 并发控制

bus := commandbus.New(
	commandbus.WithCommandHandlersPoolSize(10), // 限制并发处理程序数量
)

2. 超时控制

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

if err := bus.Dispatch(ctx, "CreateUserCommand", CreateUserCommand{
	ID: "456",
	Username: "timeout_test",
	Email: "test@example.com",
}); err != nil {
	if errors.Is(err, context.DeadlineExceeded) {
		fmt.Println("命令处理超时")
	} else {
		fmt.Println("其他错误:", err)
	}
}

最佳实践

1. 命令设计：

   • 命令应该是简单的数据传输对象(DTO)
   • 包含执行操作所需的最小数据集
   • 命名以"Command"结尾

2. 处理程序设计：

   • 每个处理程序只负责一个命令类型
   • 保持处理程序简洁，复杂逻辑委托给服务层

3. 错误处理：

   • 使用自定义错误类型区分不同类型的错误
   • 考虑实现重试机制

4. 测试：

   func TestUserCreation(t *testing.T) {
       bus := commandbus.New()
       var called bool
       
       bus.Register("CreateUserCommand", func(ctx context.Context, cmd interface{}) error {
           called = true
           return nil
       })
       
       err := bus.Dispatch(context.Background(), "CreateUserCommand", CreateUserCommand{})
       if err != nil {
           t.Fatalf("预期无错误，得到: %v", err)
       }
       
       if !called {
           t.Error("处理程序未被调用")
       }
   }
   

总结

Go-CommandBus 提供了以下优势：

• 解耦：将命令发送者与处理程序分离
• 可扩展：通过中间件和插件轻松扩展功能
• 可测试：命令和处理程序可以独立测试
• 灵活性：支持同步和异步处理模式

对于需要清晰分离关注点、支持插件架构的Go应用程序，Go-CommandBus是一个轻量级且强大的解决方案。