Golang轻量级内部通信消息总线插件库Bus的使用

Bus是一个用于内部通信的极简事件/消息总线实现，灵感来源于Elixir语言的event_bus包。

安装

通过go包安装：

go get github.com/mustafaturan/bus/v3

完整示例

下面是一个完整的Bus使用示例：

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/mustafaturan/bus/v3"
	"github.com/mustafaturan/monoton/v2"
	"github.com/mustafaturan/monoton/v2/sequencer"
)

func main() {
	// 初始化Bus
	b := NewBus()
	
	// 注册订单处理handler
	handler := bus.Handler{
		Handle: func(ctx context.Context, e bus.Event) {
			// 异步处理订单事件
			go func() {
				fmt.Printf("处理订单事件: %+v\n", e)
				// 这里可以添加具体的业务逻辑
			}()
		},
		Matcher: "order.*", // 匹配所有order开头的topic
	}
	b.RegisterHandler("order-handler", handler)

	// 模拟订单事件
	ctx := context.Background()
	ctx = context.WithValue(ctx, bus.CtxKeyTxID, "tx-123456")
	ctx = context.WithValue(ctx, bus.CtxKeySource, "order-service")

	order := map[string]string{
		"orderID":     "123456",
		"orderAmount": "112.20",
		"currency":    "USD",
	}

	// 发送订单接收事件
	err := b.Emit(ctx, "order.received", order)
	if err != nil {
		fmt.Println("发送事件失败:", err)
	}

	// 等待handler处理
	time.Sleep(100 * time.Millisecond)
}

func NewBus() *bus.Bus {
	// 配置ID生成器
	node := uint64(1)
	initialTime := uint64(1577865600000) // 设置2020-01-01 PST为初始时间
	m, err := monoton.New(sequencer.NewMillisecond(), node, initialTime)
	if err != nil {
		panic(err)
	}

	// 初始化ID生成器
	var idGenerator bus.Next = m.Next

	// 创建新的Bus实例
	b, err := bus.NewBus(idGenerator)
	if err != nil {
		panic(err)
	}

	// 注册topic
	b.RegisterTopics("order.received", "order.fulfilled")

	return b
}

核心功能

1. 配置Bus

Bus需要一个唯一的ID生成器来为事件分配ID：

import (
    "github.com/mustafaturan/bus/v3"
    "github.com/mustafaturan/monoton/v2"
    "github.com/mustafaturan/monoton/v2/sequencer"
)

func NewBus() *bus.Bus {
    // 配置ID生成器
    node := uint64(1)
    initialTime := uint64(1577865600000) // 设置2020-01-01 PST为初始时间
    m, err := monoton.New(sequencer.NewMillisecond(), node, initialTime)
    if err != nil {
        panic(err)
    }

    // 初始化ID生成器
    var idGenerator bus.Next = m.Next

    // 创建新的Bus实例
    b, err := bus.NewBus(idGenerator)
    if err != nil {
        panic(err)
    }

    // 注册topic
    b.RegisterTopics("order.received", "order.fulfilled")

    return b
}

2. 注册事件Topic

在发送事件到topic之前，需要先注册topic名称：

// 注册topic
b.RegisterTopics("order.received", "order.fulfilled")

3. 注册事件处理器

要接收topic事件，需要注册处理器：

handler := bus.Handler{
    Handle: func(ctx context.Context, e bus.Event) {
        // 处理事件
        // 注意：强烈建议以异步方式处理事件
    },
    Matcher: ".*", // 匹配所有topic
}
b.RegisterHandler("handler的唯一key", handler)

4. 发送事件

发送事件的基本方式：

// 如果txID为空，bus会使用ID生成器生成一个
ctx := context.Background()
ctx = context.WithValue(ctx, bus.CtxKeyTxID, "some-transaction-id-if-exists")
// 可选来源
ctx = context.WithValue(ctx, bus.CtxKeySource, "source-of-the-event")

// 事件topic名称(必须提前注册)
topic := "order.received"

// 事件数据
order := make(map[string]string)
order["orderID"] = "123456"
order["orderAmount"] = "112.20"
order["currency"] = "USD"

// 发送事件
err := b.Emit(ctx, topic, order)

if err != nil {
    // 处理错误
    fmt.Println(err)
}

// 使用选项发送事件
err := b.EmitWithOpts(ctx, topic, order, bus.WithTxID("some-tx-id"))

if err != nil {
    // 处理错误
    fmt.Println(err)
}

5. 事件结构

处理器接收的事件结构如下：

// 事件数据结构
type Event struct {
    ID         string      // 标识符
    TxID       string      // 事务标识符
    Topic      string      // topic名称
    Source     string      // 事件来源
    OccurredAt time.Time   // 创建时间(纳秒)
    Data       interface{} // 实际事件数据
}

性能基准

基准测试结果：

goos: darwin
goarch: amd64
pkg: github.com/mustafaturan/bus/v3
cpu: Intel(R) Core(TM) i5-6267U CPU @ 2.90GHz
BenchmarkEmit-4                      10000000       180.5 ns/op       8 B/op       0 allocs/op
BenchmarkEmitWithoutTxID-4           10000000       244.6 ns/op      72 B/op       2 allocs/op
BenchmarkEmitWithOpts-4              10000000       280.8 ns/op     112 B/op       4 allocs/op
BenchmarkEmitWithOptsUnspecified-4   10000000       169.4 ns/op       8 B/op       0 allocs/op
PASS
ok      github.com/mustafaturan/bus/v3       8.884s

许可证

Apache License 2.0