开发工具【免费下载链接】lancetA comprehensive, efficient, and reusable util function library of Go.项目地址https://gitcode.com/gh_mirrors/la/lancet点击查看免费下载lancet 是 Go 生态中一套全面、高效、可复用的工具函数库模块路径github.com/duke-git/lancet/v2其中eventbus子包提供了一套基于 Go 泛型的事件总线实现用于在单个应用内部完成事件的处理与解耦发布者只需向某个 topic 投递事件负载订阅者即可按需接收、过滤并处理。读完本文你将掌握NewEventBus、Subscribe、Publish、Unsubscribe、ClearListeners、GetListenersCount等全部 API 的用法理解优先级排序、事件过滤器、异步派发与 panic 恢复的底层实现并能直接在项目里搭建一套类型安全的进程内事件通信方案。本文主体内容对应官方 API 文档 docs/en/api/packages/eventbus.md并以仓库源码 eventbus/eventbus.go 及其测试 eventbus/eventbus_test.go、eventbus/eventbus_example_test.go 为依据进行源码级展开。一、eventbus 包概览类型安全的事件解耦eventbus包实现了「应用内部的事件总线」与分布式消息队列如 Kafka、RabbitMQ不同它不涉及网络传输与持久化只负责在进程内的模块之间传递事件。它解决的核心问题是让事件的「发布者」与「订阅者」不直接互相依赖通过 topic主题解耦同时借助 Go 泛型让每个事件负载Payload携带明确的数据类型避免interface{}带来的运行时类型断言风险。从源码结构看该包由三个核心类型构成见 eventbus/eventbus.go类型定义位置职责Event[T any]eventbus/eventbus.go#L14-L17一次事件投递的载体包含Topic主题与Payload负载泛型类型TEventListener[T any]eventbus/eventbus.go#L28-L33一次订阅的登记项保存监听函数、优先级、是否异步、过滤器EventBus[T any]eventbus/eventbus.go#L20-L25事件总线本体内部用sync.Map按 topic 存放监听器列表并用sync.RWMutex保护读写EventBus内部的数据结构非常轻量// eventbus/eventbus.go type EventBus[T any] struct { listeners sync.Map // topic - []*EventListener[T] mu sync.RWMutex // 保护 listeners 的并发访问 errorHandler func(topic string, err error) // 全局错误处理函数 }sync.Map用于承载「topic → 监听器切片」的映射sync.RWMutex则确保订阅、取消订阅与发布之间的互斥从而让EventBus可以安全地被多个 goroutine 共享——这是进程内事件总线在 Web 服务、并发任务场景下能落地的前提。二、安装与导入eventbus是 lancet 的子包直接通过 Go Modules 引入即可。当前仓库 go.mod 声明模块名为github.com/duke-git/lancet/v2最低要求Go 1.18泛型能力:import ( github.com/duke-git/lancet/v2/eventbus )对应在终端中获取依赖go get github.com/duke-git/lancet/v2/eventbuslancet 每个子包都是独立的 Go 包按需导入不会拖入无关代码。使用前请确认你的 Go 工具链版本满足go 1.18及以上。三、核心 API 详解eventbus包对外共暴露 10 个 API1 个构造函数 9 个方法。下面按「创建 → 订阅 → 发布 → 管理 → 查询」的生命周期逐一展开每个 API 都给出签名、语义、源码实现要点与可直接运行的示例。3.1 NewEventBus创建事件总线实例签名func NewEventBus[T any]() *EventBus[T]创建一个空的事件总线实例T是事件负载的类型。源码实现eventbus/eventbus.go#L37-L41只是初始化一个空的sync.Mapfunc NewEventBus[T any]() *EventBus[T] { return EventBus[T]{ listeners: sync.Map{}, } }由于T在编译期就确定同一个应用可以按领域划分创建多个总线例如*EventBus[OrderEvent]管订单域、*EventBus[UserEvent]管用户域类型互不串扰。示例订阅并发布一个int负载的事件验证监听器收到数据。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() receivedData : 0 listener : func(eventData int) { receivedData eventData } eb.Subscribe(event1, listener, false, 0, nil) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) fmt.Println(receivedData) // Output: // 1 }3.2 Subscribe订阅事件支持异步、优先级、过滤器签名func (eb *EventBus[T]) Subscribe(topic string, listener func(eventData T), async bool, priority int, filter func(eventData T) bool)这是事件总线最核心的方法也是参数最多的一个。各参数语义如下参数类型含义与默认行为topicstring订阅的事件主题与Publish时的Event.Topic精确匹配listenerfunc(eventData T)监听函数收到事件负载后执行的具体逻辑asyncbool是否异步执行。true时Publish会为每个监听器启动独立 goroutine不阻塞发布者false时在发布者调用栈内同步执行priorityint监听器优先级数值越大越先执行。同一 topic 下的监听器按优先级降序排列filterfunc(eventData T) bool事件过滤器返回true才把事件交给该监听器传nil表示不过滤源码实现eventbus/eventbus.go#L45-L65做了三件事构造EventListener、追加到该 topic 的监听器切片、按优先级降序排序func (eb *EventBus[T]) Subscribe(topic string, listener func(eventData T), async bool, priority int, filter func(eventData T) bool) { eb.mu.Lock() defer eb.mu.Unlock() el : EventListener[T]{ priority: priority, listener: listener, async: async, filter: filter, } listenersInterface, _ : eb.listeners.LoadOrStore(topic, []*EventListener[T]{}) listeners : listenersInterface.([]*EventListener[T]) listeners append(listeners, el) sort.Slice(listeners, func(i, j int) bool { return listeners[i].priority listeners[j].priority }) eb.listeners.Store(topic, listeners) }注意两个细节一是Subscribe全程持有写锁eb.mu.Lock()与Publish的读锁互斥保证「订阅/取消订阅」与「遍历派发」不会并发踩踏二是同 topic 多个监听器的执行顺序由 priority 决定priority 相等时按订阅先后顺序排列sort.Slice是稳定排序。TestEventBus_Subscribe_withPriorityeventbus/eventbus_test.go#L68-L89验证了 priority 为 2 的监听器先于 priority 为 1 的执行输出[2 1]。示例使用过滤器只接收负载为 1 的事件。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() receivedData : 0 listener : func(eventData int) { receivedData eventData } filter : func(eventData int) bool { return eventData 1 } eb.Subscribe(event1, listener, false, 0, filter) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 2}) fmt.Println(receivedData) // Output: // 1 }发布负载 2 时filter 返回false监听器被跳过因此receivedData最终仍为 1。3.3 Unsubscribe取消订阅签名func (eb *EventBus[T]) Unsubscribe(topic string, listener func(eventData T))按 topic 与监听函数取消订阅。源码实现eventbus/eventbus.go#L69-L89的关键在于通过函数指针匹配监听器Go 中函数值无法直接用比较除非是nil因此实现用fmt.Sprintf(%p, listener)取得函数地址字符串再与已注册的每个EventListener.listener的地址逐一比对剔除匹配项func (eb *EventBus[T]) Unsubscribe(topic string, listener func(eventData T)) { eb.mu.Lock() defer eb.mu.Unlock() listenersInterface, ok : eb.listeners.Load(topic) if !ok { return } listeners : listenersInterface.([]*EventListener[T]) listenerPtr : fmt.Sprintf(%p, listener) var updatedListeners []*EventListener[T] for _, l : range listeners { if fmt.Sprintf(%p, l.listener) ! listenerPtr { updatedListeners append(updatedListeners, l) } } eb.listeners.Store(topic, updatedListeners) }可以推断这种按函数指针的匹配方式要求「取消订阅时传入的必须是订阅时同一个函数值」若传入匿名函数字面量每次调用都会生成不同的函数地址将无法命中。topic 不存在时Unsubscribe静默返回不会报错。示例取消订阅后再发布事件监听器不再收到任何数据。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() receivedData : 0 listener : func(eventData int) { receivedData eventData } eb.Subscribe(event1, listener, false, 0, nil) eb.Unsubscribe(event1, listener) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) fmt.Println(receivedData) // Output: // 0 }3.4 Publish发布事件签名func (eb *EventBus[T]) Publish(event eventbus.Event[T])将Event[T]含Topic与Payload投递给该 topic 下所有监听器。源码实现eventbus/eventbus.go#L93-L115按优先级顺序遍历监听器切片先过 filter再按 async 标记选择同步调用或启动 goroutine 派发func (eb *EventBus[T]) Publish(event Event[T]) { eb.mu.RLock() defer eb.mu.RUnlock() listenersInterface, exists : eb.listeners.Load(event.Topic) if !exists { return } listeners : listenersInterface.([]*EventListener[T]) for _, listener : range listeners { if listener.filter ! nil !listener.filter(event.Payload) { continue } if listener.async { go eb.publishToListener(listener, event) } else { eb.publishToListener(listener, event) } } }几点实现细节值得注意Publish只持有读锁允许多个发布者并发发布写锁仅在订阅、取消订阅、清空监听器时被短暂持有。topic 下没有任何监听器时Publish直接返回事件被静默丢弃。异步监听器使用go关键字为每个监听器单独启动 goroutine因此同一个Publish里的多个异步监听器是并行执行的发布者不会等待它们完成。每个监听器最终都经由publishToListenereventbus/eventbus.go#L117-L125调用该函数内置了deferrecover的 panic 恢复逻辑异常不会击穿发布者 goroutine。示例同步监听器按顺序输出负载。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() eb.Subscribe(event1, func(eventData int) { fmt.Println(eventData) }, false, 0, nil) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) // Output: // 1 }3.5 SetErrorHandler设置错误处理函数签名以当前仓库源码为准func (eb *EventBus[T]) SetErrorHandler(handler func(topic string, err error))设置全局错误处理函数。当监听器执行过程中发生 panic 时publishToListener中的recover会捕获它并把「事件主题 错误信息」交给该处理函数见 eventbus/eventbus.go#L117-L131func (eb *EventBus[T]) publishToListener(listener *EventListener[T], event Event[T]) { defer func() { if r : recover(); r ! nil eb.errorHandler ! nil { eb.errorHandler(event.Topic, fmt.Errorf(%v, r)) } }() listener.listener(event.Payload) } func (eb *EventBus[T]) SetErrorHandler(handler func(topic string, err error)) { eb.errorHandler handler }⚠️文档与源码的差异提示官方文档 docs/en/api/packages/eventbus.md 中本节示例写的是单参数签名func(err error)而当前仓库源码eventbus/eventbus.go#L129-L131及示例测试ExampleEventBus_SetErrorHandlereventbus/eventbus_example_test.go#L189-L204均为双参数func(topic string, err error)。请以源码为准处理函数应接收两个参数否则编译不通过。示例监听器内故意 panic由错误处理函数捕获并输出。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() eb.SetErrorHandler(func(topic string, err error) { fmt.Println(topic, err) }) eb.Subscribe(event1, func(eventData int) { panic(error) }, false, 0, nil) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) // Output: // event1 error }对应的单元测试TestEventBus_ErrorHandlereventbus/eventbus_test.go#L111-L127同时断言了 topic 与错误文本。3.6 ClearListeners清空全部监听器签名func (eb *EventBus[T]) ClearListeners()清空所有 topic 下的所有监听器。源码实现eventbus/eventbus.go#L135-L140直接用一个全新的sync.Map{}替换旧实例func (eb *EventBus[T]) ClearListeners() { eb.mu.Lock() defer eb.mu.Unlock() eb.listeners sync.Map{} }示例清空后无论向哪个 topic 发布监听器都不会再收到数据。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() receivedData : 0 listener : func(eventData int) { receivedData eventData } eb.Subscribe(event1, listener, false, 0, nil) eb.Subscribe(event2, listener, false, 0, nil) eb.ClearListeners() eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) eb.Publish(eventbus.Event[int]{Topic: event2, Payload: 2}) fmt.Println(receivedData) // Output: // 0 }3.7 ClearListenersByTopic按主题清空监听器签名func (eb *EventBus[T]) ClearListenersByTopic(topic string)只清空指定 topic 下的监听器其他 topic 不受影响。实现eventbus/eventbus.go#L144-L149即sync.Map.Delete(topic)func (eb *EventBus[T]) ClearListenersByTopic(topic string) { eb.mu.Lock() defer eb.mu.Unlock() eb.listeners.Delete(topic) }示例只清空event1event2的监听器依然生效。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() receivedData : 0 listener : func(eventData int) { receivedData eventData } eb.Subscribe(event1, listener, false, 0, nil) eb.Subscribe(event2, listener, false, 0, nil) eb.ClearListenersByTopic(event1) eb.Publish(eventbus.Event[int]{Topic: event1, Payload: 1}) eb.Publish(eventbus.Event[int]{Topic: event2, Payload: 2}) fmt.Println(receivedData) // Output: // 2 }3.8 GetListenersCount查询单个主题的监听器数量签名func (eb *EventBus[T]) GetListenersCount(topic string) int返回指定 topic 下注册的监听器数量topic 不存在时返回 0。实现eventbus/eventbus.go#L153-L164在读锁保护下读取切片长度func (eb *EventBus[T]) GetListenersCount(topic string) int { eb.mu.RLock() defer eb.mu.RUnlock() listenersInterface, ok : eb.listeners.Load(topic) if !ok { return 0 } listeners : listenersInterface.([]*EventListener[T]) return len(listeners) }示例分别在event1、event2上各注册一个监听器。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() eb.Subscribe(event1, func(eventData int) {}, false, 0, nil) eb.Subscribe(event2, func(eventData int) {}, false, 0, nil) count : eb.GetListenersCount(event1) fmt.Println(count) // Output: // 1 }3.9 GetAllListenersCount查询全部监听器数量签名func (eb *EventBus[T]) GetAllListenersCount() int统计所有 topic 下监听器数量的总和。实现eventbus/eventbus.go#L168-L179遍历sync.Map累加每个 topic 的切片长度func (eb *EventBus[T]) GetAllListenersCount() int { eb.mu.RLock() defer eb.mu.RUnlock() count : 0 eb.listeners.Range(func(key, value interface{}) bool { count len(value.([]*EventListener[T])) return true }) return count }示例两个 topic 各注册一个监听器总数即为 2。import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() eb.Subscribe(event1, func(eventData int) {}, false, 0, nil) eb.Subscribe(event2, func(eventData int) {}, false, 0, nil) count : eb.GetAllListenersCount() fmt.Println(count) // Output: // 2 }3.10 GetEvents获取全部事件主题签名func (eb *EventBus[T]) GetEvents() []string返回当前总线中所有已注册有过订阅的 topic 列表。实现eventbus/eventbus.go#L183-L195通过sync.Map.Range收集所有 keyfunc (eb *EventBus[T]) GetEvents() []string { eb.mu.RLock() defer eb.mu.RUnlock() var events []string eb.listeners.Range(func(key, value interface{}) bool { events append(events, key.(string)) return true }) return events }⚠️返回顺序不定sync.Map.Range的遍历顺序是随机的因此GetEvents返回的 topic 顺序不保证稳定。文档示例docs/en/api/packages/eventbus.md中输出event2、event1只是某次运行的结果官方示例测试ExampleEventBus_GetEventseventbus/eventbus_example_test.go#L221-L237在断言前显式调用了sort.Strings(events)测试TestEventBus_GetEventseventbus/eventbus_test.go#L208-L222同样先排序再比较。实际使用中如需稳定顺序请自行排序。示例import ( fmt github.com/duke-git/lancet/v2/eventbus ) func main() { eb : eventbus.NewEventBus[int]() eb.Subscribe(event1, func(eventData int) {}, false, 0, nil) eb.Subscribe(event2, func(eventData int) {}, false, 0, nil) events : eb.GetEvents() for _, event : range events { fmt.Println(event) } // Output: // event2 // event1 }四、进阶实战优先级 过滤器 异步的组合Subscribe的async、priority、filter三个参数可以自由组合满足常见业务编排需求。以下是一个把三者结合使用的完整示例模拟「订单创建后广播通知」场景package main import ( fmt sync time github.com/duke-git/lancet/v2/eventbus ) type OrderEvent struct { OrderID string Amount float64 VIP bool } func main() { eb : eventbus.NewEventBus[OrderEvent]() var wg sync.WaitGroup // 1) 优先级最高审计日志同步执行先落库 eb.Subscribe(order.created, func(e OrderEvent) { fmt.Printf([audit] order %s amount%.2f\n, e.OrderID, e.Amount) }, false, 100, nil) // 2) 优先级中等库存扣减且只关心金额 0 的订单过滤器 eb.Subscribe(order.created, func(e OrderEvent) { fmt.Printf([stock] deduct for %s\n, e.OrderID) }, false, 50, func(e OrderEvent) bool { return e.Amount 0 }) // 3) 优先级最低VIP 短信通知异步执行不阻塞主流程 wg.Add(1) eb.Subscribe(order.created, func(e OrderEvent) { defer wg.Done() time.Sleep(50 * time.Millisecond) // 模拟发短信耗时 fmt.Printf([sms] notify vip %s\n, e.OrderID) }, true, 10, func(e OrderEvent) bool { return e.VIP }) // 发布事件 eb.Publish(eventbus.Event[OrderEvent]{ Topic: order.created, Payload: OrderEvent{OrderID: A1001, Amount: 88.5, VIP: true}, }) wg.Wait() // 等待异步监听器完成 fmt.Println(listeners:, eb.GetListenersCount(order.created)) fmt.Println(topics:, len(eb.GetEvents())) // 可能的输出异步部分顺序不定 // [audit] order A1001 amount88.50 // [stock] deduct for A1001 // [sms] notify vip A1001 // listeners: 3 // topics: 1 }这个例子同时体现了三个能力优先级保证审计日志最先执行过滤器让库存监听器忽略非法订单、短信监听器只服务 VIP异步让短信发送在独立 goroutine 中进行发布者无需等待。注意异步监听器需要自行协调等待如sync.WaitGroup否则主流程可能在 goroutine 完成前退出。五、并发安全与实现原理从源码可以梳理出eventbus的并发模型读写锁分层Subscribe、Unsubscribe、ClearListeners、ClearListenersByTopic持有写锁eb.mu.Lock()Publish、GetListenersCount、GetAllListenersCount、GetEvents持有读锁eb.mu.RLock()。因此多个发布者可同时发布但发布与「增删监听器」互斥从结构上避免了遍历切片时被并发修改的问题。存储结构监听器按 topic 组织在sync.Map的 value[]*EventListener[T]中同一 topic 内按 priority 降序排列实现了「同 topic 有序、跨 topic 独立」的派发语义。panic 隔离每个监听器调用都被publishToListener的defer/recover包裹单个监听器 panic 不会影响同一事件的其他监听器也不会让发布者崩溃错误通过SetErrorHandler设置的函数集中上报且上报时携带 topic便于定位是哪个主题的监听器出了问题。异步派发asynctrue的监听器每次Publish都会新起 goroutine。可以推断若高频发布且监听器较多goroutine 创建开销与调度竞争需要纳入容量规划同步监听器则天然保证「先订阅先注册、同优先级按注册序」的执行顺序。这些设计在 eventbus/eventbus_test.go 中都有对应验证TestEventBus_Subscribe_async通过WaitGroup等待异步 goroutine 完成TestEventBus_Subscribe_withPriority断言执行顺序为[2, 1]TestEventBus_ErrorHandler验证 panic 被捕获并转发。六、常见问题与使用建议Q1异步监听器的执行顺序有保证吗没有。asynctrue的监听器通过go启动调度顺序由 Go 运行时决定。若业务要求严格顺序应使用同步模式asyncfalse。Q2Unsubscribe为什么有时「取消不掉」因为实现通过函数指针匹配fmt.Sprintf(%p, listener)见 eventbus/eventbus.go#L69-L89。必须传入订阅时保存的同一个函数变量每次写新的匿名函数去取消是无效的。Q3GetEvents的返回顺序为什么会变因为底层sync.Map.Range遍历无序。需要稳定输出时自行sort.Strings参考 eventbus/eventbus_test.go#L208-L222。Q4监听器 panic 了会不会拖垮整个进程不会。publishToListener内置 recoverpanic 会被捕获并转交errorHandler若已设置未设置errorHandler时异常被静默吞掉。建议生产环境始终通过SetErrorHandler登记日志上报便于排查。Q5SetErrorHandler的参数到底有几个以当前仓库源码为准是两个func(topic string, err error)eventbus/eventbus.go#L129-L131。官方文档该节的示例为旧版单参数形式存在差异请勿照抄文档示例。Q6事件总线适合用在什么场景适合单进程内的模块解耦例如领域事件广播、插件式通知、UI 与业务逻辑解耦、埋点上报聚合等。若需要跨进程、跨机器传递事件并保证持久化应选择 MQ 类中间件eventbus不承担该职责。七、小结lancet 的eventbus包以约 200 行源码实现了「泛型 优先级 过滤器 异步 错误恢复」的完整事件总线能力。本文基于 docs/en/api/packages/eventbus.md 的 API 骨架结合 eventbus/eventbus.go 的实现细节与 eventbus/eventbus_test.go、eventbus/eventbus_example_test.go 的测试用例完整覆盖了 10 个公开 API。读者可以直接复制本文示例代码替换业务类型后即可在自己的 Go 应用中投入使用。赞分享开发工具【免费下载链接】lancetA comprehensive, efficient, and reusable util function library of Go.项目地址https://gitcode.com/gh_mirrors/la/lancet点击查看免费下载相关推荐EggJS Tegg EventBus 事件总线使用指南类型安全的事件发布订阅实战EggJS Tegg EventBus 事件总线使用指南类型安全的事件发布订阅实战 导读 EventBus 是 EggJStegg 模块体系提供的一套基于后端Web框架Go事件总线终极指南使用Lancet EventBus实现高效组件通信Go事件总线终极指南使用Lancet EventBus实现高效组件通信 在现代Go应用开发中组件间通信是构建松耦合系统的关键。Lancet EventBus开发工具VueUse 事件总线实战useEventBus 在 Vue 3 中的订阅、发布与类型安全用法VueUse 事件总线实战useEventBus 在 Vue 3 中的订阅、发布与类型安全用法 导读 useEventBus 是 VueUse 提供的轻量级事前端上一篇Python小红书数据采集实战指南3步掌握高效爬虫工具下一篇3步重塑Windows任务栏用TranslucentTB打造透明美学桌面创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考