event

package
v0.0.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 2, 2026 License: MIT Imports: 9 Imported by: 0

README

event 包 — 事件驱动系统

所属层级: Core Layer
设计理念: 观察者模式，解耦业务逻辑
设计灵感: Spring ApplicationEvent / ApplicationListener

概述

event 包提供应用事件驱动支持，实现观察者模式，解耦业务逻辑。支持同步/异步事件发布、事务事件、死信队列等高级特性。

核心功能
功能 说明
ApplicationEvent 应用事件接口，所有自定义事件需实现此接口
EventBus 事件总线，支持事件的发布和订阅
BaseEvent 基础事件实现，可直接使用或嵌入自定义事件结构体
AsyncPublisher 异步事件发布器，支持上下文超时和错误处理
TransactionalEvent 事务绑定事件，支持 BeforeCommit/AfterCommit/AfterRollback
DeadLetterQueue 死信队列，支持重试机制和退避策略
EventBusWithOrdering 保证事件发布顺序的事件总线

核心接口

ApplicationEvent 接口
type ApplicationEvent interface {
    Type() string
    Timestamp() time.Time
}
EventListener
type EventListener func(event ApplicationEvent)
EventBus
type EventBus struct {
    mu        sync.RWMutex
    listeners map[string][]EventListener
}
BaseEvent
type BaseEvent struct {
    EventType string
    EventTime time.Time
}

func (e *BaseEvent) Type() string
func (e *BaseEvent) Timestamp() time.Time
内置事件类型
常量 值 说明
EventEnvironmentPrepared "EnvironmentPrepared" 环境准备完成
EventContextRefreshed "ContextRefreshed" 上下文刷新完成
EventApplicationStarted "ApplicationStarted" 应用已启动
EventApplicationReady "ApplicationReady" 应用已就绪
EventApplicationStopped "ApplicationStopped" 应用已停止

快速开始

创建事件总线
bus := event.NewEventBus()
订阅事件
bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
    fmt.Println("应用已启动，时间:", e.Timestamp())
})
发布事件
bus.Publish(&event.BaseEvent{EventType: event.EventApplicationStarted})
取消订阅
handler := func(e event.ApplicationEvent) {
    fmt.Println("收到事件:", e.Type())
}

bus.Subscribe(event.EventApplicationReady, handler)
bus.Unsubscribe(event.EventApplicationReady, handler)

API 参考

EventBus 方法
方法 说明 示例
Subscribe(eventType, listener) 订阅指定类型的事件 bus.Subscribe("UserLogin", handler)
Unsubscribe(eventType, listener) 取消订阅 bus.Unsubscribe("UserLogin", handler)
Publish(event) 发布事件 bus.Publish(&event.BaseEvent{...})
BaseEvent 使用
// 创建基础事件
evt := &event.BaseEvent{EventType: "CustomEvent"}

// 嵌入到自定义事件中
type MyCustomEvent struct {
    event.BaseEvent
    UserID   string
    Action   string
}

使用示例

基本事件发布与订阅
package main

import (
    "fmt"
    "github.com/xudefa/enhance/event"
)

func main() {
    bus := event.NewEventBus()

    // 订阅多个事件
    bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
        fmt.Println("应用启动中...")
    })

    bus.Subscribe(event.EventApplicationReady, func(e event.ApplicationEvent) {
        fmt.Println("应用已就绪，可以提供服务")
    })

    bus.Subscribe(event.EventApplicationStopped, func(e event.ApplicationEvent) {
        fmt.Println("应用已停止")
    })

    // 发布事件
    bus.Publish(&event.BaseEvent{EventType: event.EventApplicationStarted})
    bus.Publish(&event.BaseEvent{EventType: event.EventApplicationReady})
    bus.Publish(&event.BaseEvent{EventType: event.EventApplicationStopped})
}
自定义事件类型
type UserRegisteredEvent struct {
    event.BaseEvent
    Username  string
    Email     string
}

func main() {
    bus := event.NewEventBus()

    bus.Subscribe("UserRegistered", func(e event.ApplicationEvent) {
        evt := e.(*UserRegisteredEvent)
        fmt.Printf("用户注册: %s (%s)\n", evt.Username, evt.Email)
    })

    bus.Publish(&UserRegisteredEvent{
        BaseEvent: event.BaseEvent{EventType: "UserRegistered"},
        Username:  "john",
        Email:     "john@example.com",
    })
}
多监听器
bus := event.NewEventBus()

// 多个监听器订阅同一事件
bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
    fmt.Println("监听器 1: 记录启动日志")
})

bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
    fmt.Println("监听器 2: 发送启动通知")
})

bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
    fmt.Println("监听器 3: 初始化监控指标")
})

// 发布后三个监听器按订阅顺序依次调用
bus.Publish(&event.BaseEvent{EventType: event.EventApplicationStarted})
在 Boot 启动中的事件流

Boot.Start() 按顺序发布事件：

// PhaseConfiguring 阶段
eventBus.Publish(&event.BaseEvent{EventType: event.EventEnvironmentPrepared})

// PhaseContextRefreshed 阶段
eventBus.Publish(&event.BaseEvent{EventType: event.EventContextRefreshed})

// PhaseRunning 阶段
eventBus.Publish(&event.BaseEvent{EventType: event.EventApplicationStarted})
eventBus.Publish(&event.BaseEvent{EventType: event.EventApplicationReady})

Boot.Stop() 发布：

eventBus.Publish(&event.BaseEvent{EventType: event.EventApplicationStopped})

完整的事件时间线：

PhaseConfiguring    → EventEnvironmentPrepared
PhaseContextRefreshed → EventContextRefreshed
PhaseRunning        → EventApplicationStarted
PhaseRunning        → EventApplicationReady
PhaseStopped        → EventApplicationStopped

最佳实践

1. 使用自定义事件传递业务数据
// ✅ 推荐：使用自定义事件
type OrderCreatedEvent struct {
    event.BaseEvent
    OrderID string
    Amount  float64
}

// ⚠️ 不推荐：使用 BaseEvent 传递复杂数据
bus.Publish(&event.BaseEvent{EventType: "OrderCreated"})
2. 及时取消订阅

避免内存泄漏，在对象销毁时取消订阅：

type MyComponent struct {
    bus     *event.EventBus
    handler event.EventListener
}

func (c *MyComponent) Start() {
    c.handler = func(e event.ApplicationEvent) {
        // 处理事件
    }
    c.bus.Subscribe("MyEvent", c.handler)
}

func (c *MyComponent) Stop() {
    c.bus.Unsubscribe("MyEvent", c.handler)
}
3. 避免在事件处理中抛出异常

事件处理器应该捕获并处理异常，避免影响其他监听器：

bus.Subscribe("UserLogin", func(e event.ApplicationEvent) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("事件处理异常: %v", r)
        }
    }()
    // 处理逻辑
})
4. 使用异步事件发布处理耗时操作

对于耗时的事件处理，使用异步发布避免阻塞主流程：

// 异步发布事件
asyncPublisher := event.NewAsyncPublisher(bus)
asyncPublisher.Publish(context.Background(), &event.BaseEvent{
    EventType: "EmailNotification",
})

bus.Publish(&UserRegisteredEvent{
    BaseEvent: event.BaseEvent{EventType: "UserRegistered"},
    Username:  "alice",
    Email:     "alice@example.com",
})

}


### 使用事件进行模块解耦

```go
// order 模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("订单 %s 支付完成，更新订单状态\n", evt.OrderID)
})

// notification 模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("发送支付成功通知给用户 %s\n", evt.UserID)
})

// analytics 模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("统计: 订单金额 %.2f\n", evt.Amount)
})

与 context 包的关系

DefaultApplicationContext 内部持有 EventBus：

ctx := context.NewApplicationContext(container, env)

ctx.EventBus().Subscribe(event.EventApplicationReady, func(e event.ApplicationEvent) {
    fmt.Println("上下文已就绪")
})

ctx.Start() // 触发 EventApplicationStarted 和 EventApplicationReady
ctx.Stop()  // 触发 EventApplicationStopped

使用场景

场景 1：应用生命周期管理

描述：监听应用启动、就绪、停止等生命周期事件，执行相应的初始化和清理操作。

bus := event.NewEventBus()

bus.Subscribe(event.EventApplicationStarted, func(e event.ApplicationEvent) {
    fmt.Println("应用启动中，初始化资源...")
})

bus.Subscribe(event.EventApplicationReady, func(e event.ApplicationEvent) {
    fmt.Println("应用已就绪，开始处理请求...")
})

bus.Subscribe(event.EventApplicationStopped, func(e event.ApplicationEvent) {
    fmt.Println("应用停止中，清理资源...")
})

最佳实践：

  • 在应用启动时订阅生命周期事件
  • 使用事件进行资源初始化和清理
  • 避免在监听器中执行耗时操作
场景 2：模块间解耦

描述：使用事件实现模块间松耦合通信，避免直接依赖。

// 订单模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("订单 %s 支付完成，更新订单状态\n", evt.OrderID)
})

// 通知模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("发送支付成功通知给用户 %s\n", evt.UserID)
})

// 分析模块
bus.Subscribe("PaymentCompleted", func(e event.ApplicationEvent) {
    evt := e.(*PaymentCompletedEvent)
    fmt.Printf("统计: 订单金额 %.2f\n", evt.Amount)
})

最佳实践：

  • 使用事件实现发布-订阅模式
  • 每个模块独立订阅感兴趣的事件
  • 避免在事件处理中引入循环依赖
场景 3：审计日志

描述：记录关键业务操作的事件日志，用于审计和追溯。

bus.Subscribe("UserCreated", func(e event.ApplicationEvent) {
    evt := e.(*UserCreatedEvent)
    log.Printf("用户创建: ID=%s, Username=%s, Time=%s",
        evt.UserID, evt.Username, evt.Timestamp())
})

bus.Subscribe("OrderPlaced", func(e event.ApplicationEvent) {
    evt := e.(*OrderPlacedEvent)
    log.Printf("订单创建: OrderID=%s, UserID=%s, Amount=%.2f",
        evt.OrderID, evt.UserID, evt.Amount)
})

最佳实践：

  • 使用事件记录关键业务操作
  • 包含足够的上下文信息
  • 异步处理审计日志，避免影响业务性能
场景 4：缓存失效

描述：当数据变更时，通过事件通知相关模块清理缓存。

bus.Subscribe("DataUpdated", func(e event.ApplicationEvent) {
    evt := e.(*DataUpdatedEvent)
    cacheKey := fmt.Sprintf("%s:%s", evt.DataType, evt.DataID)
    cache.Delete(cacheKey)
    fmt.Printf("缓存已清理: %s\n", cacheKey)
})

最佳实践：

  • 使用事件触发缓存清理
  • 确保缓存清理的幂等性
  • 考虑使用事件版本号避免重复处理

Documentation

Overview

Package event 提供应用事件驱动支持，用于 enhance 框架。

该模块提供完整的事件发布/订阅机制，参考 Spring 的 ApplicationEvent/ApplicationListener 模式。 支持同步和异步事件处理、事务性事件、死信队列等高级功能。

架构设计

  • ApplicationEvent: 应用事件接口
  • EventBus: 事件总线，支持发布/订阅
  • BaseEvent: 基础事件实现
  • EventListener: 事件监听器函数类型
  • ListenerConfig: 监听器配置
  • EventBusWithOrdering: 支持优先级和过滤条件的事件总线

核心功能

  • 事件发布: 支持同步和异步事件发布
  • 事件订阅: 支持按事件类型订阅
  • 异步处理: 支持异步事件处理，提升性能
  • 事务性事件: 支持事务提交后触发事件
  • 死信队列: 处理失败事件，支持重试

使用方式

定义事件： 

type UserCreatedEvent struct {
    *event.BaseEvent
    UserID int64
}

发布事件： 

bus := event.NewEventBus()
bus.Publish(&UserCreatedEvent{
    BaseEvent: &event.BaseEvent{EventType: "user.created"},
    UserID:    123,
})

订阅事件： 

bus.Subscribe(func(e event.ApplicationEvent) {
    if evt, ok := e.(*UserCreatedEvent); ok {
        fmt.Println("User created:", evt.UserID)
    }
})

异步事件

使用异步监听器处理事件： 

bus.SubscribeAsync(func(e event.ApplicationEvent) {
    // 异步处理
})

事务性事件

使用事务性监听器，在事务提交后触发： 

bus.SubscribeTransactional(func(e event.ApplicationEvent) {
    // 事务提交后触发
})

Index

Constants

View Source 
const (
	// EventEnvironmentPrepared 环境配置准备完成事件。
	// 在环境配置加载完成后触发。
	EventEnvironmentPrepared = "EnvironmentPrepared"

	// EventContextRefreshed 应用上下文刷新完成事件。
	// 在应用上下文刷新完成后触发。
	EventContextRefreshed = "ContextRefreshed"

	// EventApplicationStarted 应用启动事件。
	// 在应用开始启动时触发。
	EventApplicationStarted = "ApplicationStarted"

	// EventApplicationReady 应用就绪事件。
	// 在应用完全启动并准备好处理请求时触发。
	EventApplicationReady = "ApplicationReady"

	// EventApplicationStopped 应用停止事件。
	// 在应用停止时触发。
	EventApplicationStopped = "ApplicationStopped"
)

内置事件类型常量。

这些是 enhance 框架生命周期中的标准事件类型， 应用可以在这些事件发生时注册监听器执行自定义逻辑。

Variables

This section is empty.

Functions

func ConditionAfter 

func ConditionAfter(t time.Time) func(ApplicationEvent) bool

ConditionAfter 按时间戳过滤，仅处理指定时间之后的事件

func ConditionAlways 

func ConditionAlways() func(ApplicationEvent) bool

ConditionAlways 始终返回 true 的过滤条件

func ConditionBefore 

func ConditionBefore(t time.Time) func(ApplicationEvent) bool

ConditionBefore 按时间戳过滤，仅处理指定时间之前的事件

func ConditionType 

func ConditionType(types ...string) func(ApplicationEvent) bool

ConditionType 按事件类型过滤

Types

type ApplicationEvent 

type ApplicationEvent interface {
	// Type 返回事件类型字符串，用于事件路由和匹配。
	Type() string

	// Timestamp 返回事件发生的时间戳。
	Timestamp() time.Time
}

ApplicationEvent 应用事件接口。

所有应用事件必须实现此接口。事件通过 Type() 返回的类型字符串进行路由和分发。

设计原则

  • 使用字符串类型标识事件，而非反射类型，提高灵活性和可读性
  • 支持任意结构体实现事件接口，无需继承基类
  • 时间戳用于事件排序和审计

type AsyncPublisher 

type AsyncPublisher struct {
	// contains filtered or unexported fields
}

AsyncPublisher 异步事件发布器

提供异步事件发布功能，支持上下文超时控制和错误处理。 使用工作协程池处理事件，避免阻塞发布者。

使用示例： 

bus := event.NewEventBusWithOrdering()
publisher := event.NewAsyncPublisher(bus,
    event.WithWorkerCount(5),
    event.WithWorkerQueueSize(100),
    event.WithErrorHandler(func(err error, e event.ApplicationEvent) {
        log.Printf("event error: %v", err)
    }),
)
defer publisher.Close()

ctx := context.Background()
publisher.Publish(ctx, &event.BaseEvent{EventType: "MyEvent"})

func NewAsyncPublisher 

func NewAsyncPublisher(bus AsyncPublisherBus, opts ...AsyncPublisherOption) *AsyncPublisher

NewAsyncPublisher 创建异步事件发布器

参数:

  • bus: 事件发布器接口（支持 EventBus、EventBusWithOrdering 等）
  • opts: 可选配置项

返回:

  • *AsyncPublisher: 异步发布器实例

func (*AsyncPublisher) Close 

func (p *AsyncPublisher) Close()

Close 关闭异步发布器

先通知工作协程停止接收新任务，等待所有正在执行的任务完成， 再关闭工作通道让排空循环退出，最后等待所有工作协程退出。

func (*AsyncPublisher) Publish 

func (p *AsyncPublisher) Publish(ctx context.Context, event ApplicationEvent)

Publish 异步发布事件

将事件发布到工作队列，由工作协程异步处理。 支持上下文超时控制，超时后调用错误处理器。

参数:

  • ctx: 上下文，用于超时控制
  • event: 要发布的事件

type AsyncPublisherBuilder 

type AsyncPublisherBuilder struct {
	// contains filtered or unexported fields
}

AsyncPublisherBuilder 异步事件发布器构建器

func NewAsyncPublisherBuilder 

func NewAsyncPublisherBuilder() *AsyncPublisherBuilder

NewAsyncPublisherBuilder 创建异步事件发布器构建器

func (*AsyncPublisherBuilder) Build 

func (b *AsyncPublisherBuilder) Build() (*AsyncPublisher, error)

Build 构建异步事件发布器

func (*AsyncPublisherBuilder) Bus 

func (b *AsyncPublisherBuilder) Bus(bus *EventBus) *AsyncPublisherBuilder

Bus 设置事件总线

func (*AsyncPublisherBuilder) ErrorHandler 

func (b *AsyncPublisherBuilder) ErrorHandler(handler func(error, ApplicationEvent)) *AsyncPublisherBuilder

ErrorHandler 设置错误处理器

func (*AsyncPublisherBuilder) MustBuild 

func (b *AsyncPublisherBuilder) MustBuild() *AsyncPublisher

MustBuild 构建异步事件发布器，失败则panic

func (*AsyncPublisherBuilder) WorkerCount 

func (b *AsyncPublisherBuilder) WorkerCount(count int) *AsyncPublisherBuilder

WorkerCount 设置工作协程池大小

type AsyncPublisherBus 

type AsyncPublisherBus interface {
	// Publish 发布事件。
	Publish(event ApplicationEvent)
}

AsyncPublisherBus 事件发布器接口。

定义事件发布的最小接口，供 AsyncPublisher 使用。

type AsyncPublisherOption 

type AsyncPublisherOption func(*AsyncPublisher)

AsyncPublisherOption 异步发布器选项函数

func WithErrorHandler 

func WithErrorHandler(handler func(error, ApplicationEvent)) AsyncPublisherOption

WithErrorHandler 设置错误处理器

参数:

  • handler: 错误处理函数，接收错误和事件作为参数

返回:

  • AsyncPublisherOption: 选项函数

func WithWorkerCount 

func WithWorkerCount(n int) AsyncPublisherOption

WithWorkerCount 设置工作协程池大小

参数:

  • n: 工作协程数量

返回:

  • AsyncPublisherOption: 选项函数

func WithWorkerQueueSize 

func WithWorkerQueueSize(n int) AsyncPublisherOption

WithWorkerQueueSize 设置工作队列缓冲大小

独立于 workerCount 配置，允许设置更大的队列缓冲以应对瞬时高峰。

参数:

  • n: 队列缓冲大小

返回:

  • AsyncPublisherOption: 选项函数

type BackoffStrategy 

type BackoffStrategy string

BackoffStrategy 退避策略类型 

const (
	BackoffNone        BackoffStrategy = "none"        // 无退避，立即重试
	BackoffFixed       BackoffStrategy = "fixed"       // 固定间隔退避
	BackoffExponential BackoffStrategy = "exponential" // 指数退避
	BackoffLinear      BackoffStrategy = "linear"      // 线性退避
)

type BaseEvent 

type BaseEvent struct {
	EventType string    // 事件类型
	EventTime time.Time // 事件发生时间（可选，为空时自动使用当前时间）
}

BaseEvent 基础事件实现。

可直接使用，也支持嵌入到自定义事件结构体中。 如果 EventTime 未设置，Timestamp() 会自动返回当前时间。

使用示例 

// 直接使用
evt := &event.BaseEvent{EventType: "user.created"}

// 嵌入到自定义事件
type UserCreatedEvent struct {
    event.BaseEvent
    UserID int
}

func (*BaseEvent) Timestamp 

func (e *BaseEvent) Timestamp() time.Time

Timestamp 返回事件发生的时间戳。

如果 EventTime 未设置（零值），自动返回当前时间。

func (*BaseEvent) Type 

func (e *BaseEvent) Type() string

Type 返回事件类型字符串。

type BaseEventBuilder 

type BaseEventBuilder struct {
	// contains filtered or unexported fields
}

BaseEventBuilder 基础事件构建器

func ApplicationReadyEvent 

func ApplicationReadyEvent() *BaseEventBuilder

ApplicationReadyEvent 创建应用就绪事件

func ApplicationStartedEvent 

func ApplicationStartedEvent() *BaseEventBuilder

ApplicationStartedEvent 创建应用启动事件

func ApplicationStoppedEvent 

func ApplicationStoppedEvent() *BaseEventBuilder

ApplicationStoppedEvent 创建应用停止事件

func ContextRefreshedEvent 

func ContextRefreshedEvent() *BaseEventBuilder

ContextRefreshedEvent 创建上下文刷新事件

func EnvironmentPreparedEvent 

func EnvironmentPreparedEvent() *BaseEventBuilder

EnvironmentPreparedEvent 创建环境准备事件

func NewBaseEventBuilder 

func NewBaseEventBuilder() *BaseEventBuilder

NewBaseEventBuilder 创建基础事件构建器

func (*BaseEventBuilder) Build 

func (b *BaseEventBuilder) Build() *BaseEvent

Build 构建基础事件

func (*BaseEventBuilder) Now 

func (b *BaseEventBuilder) Now() *BaseEventBuilder

Now 设置事件时间戳为当前时间

func (*BaseEventBuilder) Publish 

func (b *BaseEventBuilder) Publish(bus *EventBus)

Publish 发布事件到事件总线

func (*BaseEventBuilder) Timestamp 

func (b *BaseEventBuilder) Timestamp(timestamp time.Time) *BaseEventBuilder

Timestamp 设置事件时间戳

func (*BaseEventBuilder) Type 

func (b *BaseEventBuilder) Type(eventType string) *BaseEventBuilder

Type 设置事件类型

type DeadLetterOption 

type DeadLetterOption func(*EventBusWithDeadLetter)

DeadLetterOption 死信队列配置选项

func WithBackoff 

func WithBackoff(strategy BackoffStrategy, initialDelay time.Duration) DeadLetterOption

WithBackoff 设置退避策略

func WithDeadLetterHandler 

func WithDeadLetterHandler(handler func(FailedEvent)) DeadLetterOption

WithDeadLetterHandler 设置永久失败处理器

func WithMaxRetries 

func WithMaxRetries(n int) DeadLetterOption

WithMaxRetries 设置最大重试次数

func WithRetryPolicy 

func WithRetryPolicy(policy RetryPolicy) DeadLetterOption

WithRetryPolicy 设置完整重试策略

type DeadLetterQueue 

type DeadLetterQueue struct {
	// contains filtered or unexported fields
}

DeadLetterQueue 死信队列

存储处理失败的事件，支持重试和永久失败处理。 线程安全，使用 sync.Map 优化并发访问，atomic.Int64 跟踪大小。

func NewDeadLetterQueue 

func NewDeadLetterQueue() *DeadLetterQueue

NewDeadLetterQueue 创建死信队列

func (*DeadLetterQueue) Add 

func (dlq *DeadLetterQueue) Add(fe FailedEvent)

Add 添加失败事件到死信队列

func (*DeadLetterQueue) Clear 

func (dlq *DeadLetterQueue) Clear()

Clear 清空死信队列

func (*DeadLetterQueue) Events 

func (dlq *DeadLetterQueue) Events() []FailedEvent

Events 返回所有死信事件（快照）

func (*DeadLetterQueue) GetByType 

func (dlq *DeadLetterQueue) GetByType(eventType string) []FailedEvent

GetByType 获取指定类型的所有失败事件（快照）

func (*DeadLetterQueue) Peek 

func (dlq *DeadLetterQueue) Peek() (FailedEvent, bool)

Peek 获取下一个可重试的事件（不移除）

func (*DeadLetterQueue) Remove 

func (dlq *DeadLetterQueue) Remove(event ApplicationEvent)

Remove 移除指定事件（重试成功后调用）

匹配规则：按事件类型和发生时间匹配，而非接口指针地址。

func (*DeadLetterQueue) RemoveByType 

func (dlq *DeadLetterQueue) RemoveByType(eventType string) int

RemoveByType 移除指定类型的所有事件

func (*DeadLetterQueue) SetPermanentFailureHandler 

func (dlq *DeadLetterQueue) SetPermanentFailureHandler(handler func(FailedEvent))

SetPermanentFailureHandler 设置永久失败处理器

func (*DeadLetterQueue) Size 

func (dlq *DeadLetterQueue) Size() int

Size 返回死信队列中的事件数量

func (*DeadLetterQueue) Stats 

func (dlq *DeadLetterQueue) Stats() DeadLetterStats

Stats 返回死信队列统计信息

type DeadLetterStats 

type DeadLetterStats struct {
	Total          int            // 总事件数
	Retryable      int            // 可重试事件数
	Exhausted      int            // 已耗尽事件数
	EventTypeCount map[string]int // 按事件类型统计
}

DeadLetterStats 死信队列统计信息

type EventBus 

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus 事件总线。

负责事件的发布与订阅管理，支持多监听器注册。 线程安全，支持并发发布和订阅。 使用 sync.Map 优化读多写少场景的性能。

性能优化

  • 使用 sync.Map 存储监听器，无锁读取
  • 使用 CAS 操作实现无锁订阅更新
  • 避免 range 分配迭代器，使用索引遍历

并发安全

EventBus 的所有方法都是并发安全的。 订阅和取消订阅使用 CAS 操作实现无锁更新， 发布事件使用无锁读取，性能优异。

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus 创建新的事件总线实例。

func (*EventBus) Publish 

func (b *EventBus) Publish(event ApplicationEvent)

Publish 发布事件，通知所有订阅了该事件类型的监听器。

发布流程：

  1. 从 sync.Map 获取事件类型的监听器列表（无锁）
  2. 如果无监听器则直接返回
  3. 遍历监听器列表并逐个调用

参数:

  • event: 要发布的事件实例

注意:

  • 监听器按注册顺序同步调用
  • 如果监听器抛出 panic，会影响后续监听器的执行
  • 对于耗时操作，建议使用异步事件总线

性能提示:

  • 发布操作是无锁的，性能优异
  • 监听器数量较多时，考虑使用 AsyncEventBus

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(eventType string, listener EventListener)

Subscribe 订阅指定类型的事件。

参数:

  • eventType: 事件类型字符串，与 ApplicationEvent.Type() 返回值对应
  • listener: 事件监听器函数

并发安全

使用 CAS 操作实现无锁订阅，支持高并发场景。 多个 goroutine 可以同时订阅同一事件类型，不会丢失任何订阅。

使用示例 

bus.Subscribe("user.created", func(e event.ApplicationEvent) {
    log.Println("New user created")
})

性能提示

  • 首次订阅使用 LoadOrStore 快速路径，无锁
  • 后续订阅使用 CAS 重试，保证并发安全
  • 避免在事件处理函数中调用 Subscribe，可能导致死锁

func (*EventBus) Unsubscribe 

func (b *EventBus) Unsubscribe(eventType string, target EventListener)

Unsubscribe 取消订阅指定类型的事件。

参数:

  • eventType: 事件类型字符串
  • target: 要移除的监听器函数

注意:

  • 使用 reflect 比较函数指针来定位要移除的监听器
  • 如果监听器不存在，此操作是空操作
  • 移除最后一个监听器时会自动删除该事件类型的记录

性能提示:

  • 取消订阅需要遍历监听器列表，O(n) 复杂度
  • 频繁取消订阅的场景，考虑使用一次性监听器

type EventBusBuilder 

type EventBusBuilder struct {
	// contains filtered or unexported fields
}

EventBusBuilder 事件总线构建器，支持链式配置

func NewEventBusBuilder 

func NewEventBusBuilder() *EventBusBuilder

NewEventBusBuilder 创建事件总线构建器

func (*EventBusBuilder) Build 

func (b *EventBusBuilder) Build() *EventBus

Build 构建事件总线

func (*EventBusBuilder) OnApplicationReady 

func (b *EventBusBuilder) OnApplicationReady(listener EventListener) *EventBusBuilder

OnApplicationReady 注册应用就绪事件监听器

func (*EventBusBuilder) OnApplicationStarted 

func (b *EventBusBuilder) OnApplicationStarted(listener EventListener) *EventBusBuilder

OnApplicationStarted 注册应用启动事件监听器

func (*EventBusBuilder) OnApplicationStopped 

func (b *EventBusBuilder) OnApplicationStopped(listener EventListener) *EventBusBuilder

OnApplicationStopped 注册应用停止事件监听器

func (*EventBusBuilder) Subscribe 

func (b *EventBusBuilder) Subscribe(eventType string, listener EventListener) *EventBusBuilder

Subscribe 订阅事件

type EventBusWithDeadLetter 

type EventBusWithDeadLetter struct {
	*EventBusWithOrdering
	// contains filtered or unexported fields
}

EventBusWithDeadLetter 支持死信队列的事件总线

在 EventBusWithOrdering 基础上增加：

  • 监听器 panic 捕获
  • 自动重试机制
  • 死信队列

使用示例： 

bus := event.NewEventBusWithDeadLetter(
    context.Background(),
    event.WithMaxRetries(3),
    event.WithBackoff(event.BackoffExponential, time.Second),
)

bus.Subscribe("MyEvent", func(e event.ApplicationEvent) {
    // 可能失败的处理逻辑
})

bus.Publish(&event.BaseEvent{EventType: "MyEvent"})

func NewEventBusWithDeadLetter 

func NewEventBusWithDeadLetter(ctx context.Context, opts ...DeadLetterOption) *EventBusWithDeadLetter

NewEventBusWithDeadLetter 创建支持死信队列的事件总线

参数:

  • ctx: 父级 context，用于控制异步重试生命周期
  • opts: 配置选项

func (*EventBusWithDeadLetter) Close 

func (b *EventBusWithDeadLetter) Close()

Close 关闭事件总线，取消所有待处理的异步重试并等待 goroutine 退出。

调用后，所有正在进行的异步重试将收到 ctx.Done() 信号并终止。 此方法会阻塞直到所有异步 goroutine 完成（包括重试 goroutine 和异步事件处理器）。 死信队列中的数据不会被清除，可通过 RetryDeadLetter 或 RetryAllDeadLetters 手动处理。

func (*EventBusWithDeadLetter) DeadLetterQueue 

func (b *EventBusWithDeadLetter) DeadLetterQueue() *DeadLetterQueue

DeadLetterQueue 返回死信队列实例

func (*EventBusWithDeadLetter) Publish 

func (b *EventBusWithDeadLetter) Publish(event ApplicationEvent)

Publish 覆盖原有 Publish 方法，使用带恢复的发布

func (*EventBusWithDeadLetter) PublishWithRecovery 

func (b *EventBusWithDeadLetter) PublishWithRecovery(event ApplicationEvent)

PublishWithRecovery 发布事件并捕获错误，失败时进入死信队列

func (*EventBusWithDeadLetter) RetryAllDeadLetters 

func (b *EventBusWithDeadLetter) RetryAllDeadLetters() int

RetryAllDeadLetters 重试所有可重试的死信事件

func (*EventBusWithDeadLetter) RetryDeadLetter 

func (b *EventBusWithDeadLetter) RetryDeadLetter() bool

RetryDeadLetter 手动重试死信队列中的下一个事件

func (*EventBusWithDeadLetter) RetryPolicy 

func (b *EventBusWithDeadLetter) RetryPolicy() RetryPolicy

RetryPolicy 返回重试策略

type EventBusWithOrdering 

type EventBusWithOrdering struct {
	// contains filtered or unexported fields
}

EventBusWithOrdering 支持优先级和过滤条件的事件总线。

在原有 EventBus 基础上扩展，支持：

  • 监听器优先级排序
  • 条件过滤
  • 异步执行
  • 向后兼容原有的 Subscribe/Publish API

使用 sync.Map 优化读多写少场景的并发性能。

func NewEventBusWithOrdering 

func NewEventBusWithOrdering() *EventBusWithOrdering

NewEventBusWithOrdering 创建支持优先级和过滤条件的事件总线。

func (*EventBusWithOrdering) Clear 

func (b *EventBusWithOrdering) Clear(eventType string)

Clear 清除指定事件类型的所有监听器

func (*EventBusWithOrdering) ClearAll 

func (b *EventBusWithOrdering) ClearAll()

ClearAll 清除所有监听器

func (*EventBusWithOrdering) Close added in v0.0.4 

func (b *EventBusWithOrdering) Close()

Close 等待所有异步事件处理器完成并关闭事件总线。

func (*EventBusWithOrdering) Listeners 

func (b *EventBusWithOrdering) Listeners(eventType string) int

Listeners 返回指定事件类型的监听器数量

func (*EventBusWithOrdering) Publish 

func (b *EventBusWithOrdering) Publish(event ApplicationEvent)

Publish 发布事件，按优先级排序并应用过滤条件

性能优化：

  • 使用预分配切片避免动态扩容
  • 快照后释放锁，减少锁持有时间

func (*EventBusWithOrdering) Subscribe 

func (b *EventBusWithOrdering) Subscribe(eventType string, listener EventListener)

Subscribe 订阅事件（向后兼容，等价于 Order=0 无条件的监听器）

func (*EventBusWithOrdering) SubscribeOnce 

func (b *EventBusWithOrdering) SubscribeOnce(eventType string, listener EventListener)

SubscribeOnce 订阅事件，仅消费一次后自动取消订阅

func (*EventBusWithOrdering) SubscribeWithConfig 

func (b *EventBusWithOrdering) SubscribeWithConfig(eventType string, config ListenerConfig)

SubscribeWithConfig 带配置的订阅

func (*EventBusWithOrdering) Unsubscribe 

func (b *EventBusWithOrdering) Unsubscribe(eventType string, target EventListener)

Unsubscribe 取消订阅

func (*EventBusWithOrdering) WaitAsync added in v0.0.4 

func (b *EventBusWithOrdering) WaitAsync()

WaitAsync 等待所有异步事件处理器完成。

在应用关闭时调用此方法，确保所有异步事件处理完成后再退出。

type EventListener 

type EventListener func(event ApplicationEvent)

EventListener 事件监听器函数类型。

接收 ApplicationEvent 参数，处理事件通知。 监听器在事件发布时同步调用。

type FailedEvent 

type FailedEvent struct {
	Event         ApplicationEvent // 原始事件
	Err           error            // 最后一次错误
	RetryCount    int              // 已重试次数
	MaxRetries    int              // 最大重试次数
	FirstFailedAt time.Time        // 首次失败时间
	LastFailedAt  time.Time        // 最后失败时间
	NextRetryAt   time.Time        // 下次重试时间
	// contains filtered or unexported fields
}

FailedEvent 失败事件记录

func (FailedEvent) IsExhausted 

func (fe FailedEvent) IsExhausted() bool

IsExhausted 返回是否已达到最大重试次数

func (FailedEvent) ShouldRetry 

func (fe FailedEvent) ShouldRetry() bool

ShouldRetry 返回是否应该重试

type LegacyEventBusAdapter 

type LegacyEventBusAdapter struct {
	// contains filtered or unexported fields
}

LegacyEventBusAdapter 将 EventBusWithOrdering 适配为 EventBus 接口。 用于需要 *EventBus 类型的场景。

func NewLegacyEventBusAdapter 

func NewLegacyEventBusAdapter(bus *EventBusWithOrdering) *LegacyEventBusAdapter

NewLegacyEventBusAdapter 创建适配器。

func (*LegacyEventBusAdapter) Publish 

func (a *LegacyEventBusAdapter) Publish(event ApplicationEvent)

Publish 转发到 EventBusWithOrdering

func (*LegacyEventBusAdapter) Subscribe 

func (a *LegacyEventBusAdapter) Subscribe(eventType string, listener EventListener)

Subscribe 转发到 EventBusWithOrdering

func (*LegacyEventBusAdapter) Unsubscribe 

func (a *LegacyEventBusAdapter) Unsubscribe(eventType string, target EventListener)

Unsubscribe 转发到 EventBusWithOrdering

type ListenerConfig 

type ListenerConfig struct {
	Handler   EventListener               // 事件处理函数（必填）
	Order     int                         // 执行优先级，数值越小越先执行
	Condition func(ApplicationEvent) bool // 过滤条件，返回 false 时跳过
	Async     bool                        // 是否异步执行
}

ListenerConfig 事件监听器配置。

提供比简单函数签名更丰富的监听器控制能力。

字段说明

  • Handler: 事件处理函数（必填）
  • Order: 执行优先级，数值越小越先执行，默认 0
  • Condition: 过滤条件函数，返回 false 时跳过该监听器
  • Async: 是否异步执行，默认 false

使用示例 

bus.SubscribeWithConfig("MyEvent", event.ListenerConfig{
    Handler: func(e event.ApplicationEvent) {
        fmt.Println("处理事件:", e.Type())
    },
    Order: 10,
    Condition: func(e event.ApplicationEvent) bool {
        return e.Type() == "MyEvent"
    },
})

func NewListenerConfig 

func NewListenerConfig(handler EventListener) ListenerConfig

NewListenerConfig 创建监听器配置

func (ListenerConfig) WithAsync 

func (c ListenerConfig) WithAsync(async bool) ListenerConfig

WithAsync 设置异步执行

func (ListenerConfig) WithCondition 

func (c ListenerConfig) WithCondition(cond func(ApplicationEvent) bool) ListenerConfig

WithCondition 设置过滤条件

func (ListenerConfig) WithOrder 

func (c ListenerConfig) WithOrder(order int) ListenerConfig

WithOrder 设置优先级

type RetryPolicy 

type RetryPolicy struct {
	MaxRetries   int             // 最大重试次数，0 表示不重试
	Strategy     BackoffStrategy // 退避策略
	InitialDelay time.Duration   // 初始延迟
	MaxDelay     time.Duration   // 最大延迟（指数退避上限）
	Multiplier   float64         // 退避乘数（指数退避用）
}

RetryPolicy 重试策略配置

func DefaultRetryPolicy 

func DefaultRetryPolicy() RetryPolicy

DefaultRetryPolicy 默认重试策略

func NoRetryPolicy 

func NoRetryPolicy() RetryPolicy

NoRetryPolicy 不重试策略

func (RetryPolicy) CalculateDelay 

func (p RetryPolicy) CalculateDelay(attempt int) time.Duration

CalculateDelay 计算当前重试次数对应的延迟

type TransactionContext 

type TransactionContext struct {
	// contains filtered or unexported fields
}

TransactionContext 事务上下文

跟踪事务中注册的事件，在 Commit/Rollback 时按阶段发布。 线程安全，支持并发注册事件。 使用分段锁设计，减少高并发场景下的锁竞争。

func NewTransactionContext 

func NewTransactionContext() *TransactionContext

NewTransactionContext 创建事务上下文

func (*TransactionContext) Commit 

func (tc *TransactionContext) Commit(bus *EventBus)

Commit 提交事务，发布 BeforeCommit 和 AfterCommit 事件

线程安全，多次调用只会执行一次。

func (*TransactionContext) IsCommitted 

func (tc *TransactionContext) IsCommitted() bool

IsCommitted 返回事务是否已提交

func (*TransactionContext) IsRolledBack 

func (tc *TransactionContext) IsRolledBack() bool

IsRolledBack 返回事务是否已回滚

func (*TransactionContext) PublishAfterCommit 

func (tc *TransactionContext) PublishAfterCommit(event ApplicationEvent)

PublishAfterCommit 注册 AfterCommit 阶段事件

func (*TransactionContext) PublishAfterRollback 

func (tc *TransactionContext) PublishAfterRollback(event ApplicationEvent)

PublishAfterRollback 注册 AfterRollback 阶段事件

func (*TransactionContext) PublishBeforeCommit 

func (tc *TransactionContext) PublishBeforeCommit(event ApplicationEvent)

PublishBeforeCommit 注册 BeforeCommit 阶段事件

func (*TransactionContext) RegisterEvent 

func (tc *TransactionContext) RegisterEvent(event ApplicationEvent)

RegisterEvent 注册事务事件

线程安全，支持并发调用。

func (*TransactionContext) Rollback 

func (tc *TransactionContext) Rollback(bus *EventBus)

Rollback 回滚事务，发布 AfterRollback 事件

线程安全，多次调用只会执行一次。

type TransactionPhase 

type TransactionPhase string

TransactionPhase 事务阶段枚举 

const (
	PhaseBeforeCommit  TransactionPhase = "before_commit"  // 事务提交前
	PhaseAfterCommit   TransactionPhase = "after_commit"   // 事务提交后
	PhaseAfterRollback TransactionPhase = "after_rollback" // 事务回滚后
)

type TransactionalEvent 

type TransactionalEvent struct {
	// contains filtered or unexported fields
}

TransactionalEvent 事务事件包装器

包装 ApplicationEvent 并绑定事务阶段信息。 在事务提交/回滚时，事件会根据绑定的阶段延迟发布。

使用示例： 

te := event.NewTransactionalEvent(
    &MyEvent{Data: "hello"},
    event.PhaseAfterCommit,
)

func NewTransactionalEvent 

func NewTransactionalEvent(event ApplicationEvent, phase TransactionPhase) *TransactionalEvent

NewTransactionalEvent 创建事务事件

func (*TransactionalEvent) Event 

func (e *TransactionalEvent) Event() ApplicationEvent

Event 返回内部事件

func (*TransactionalEvent) Phase 

func (e *TransactionalEvent) Phase() TransactionPhase

Phase 返回事务阶段

func (*TransactionalEvent) Timestamp 

func (e *TransactionalEvent) Timestamp() time.Time

Timestamp 返回事件时间戳（委托给内部事件）

func (*TransactionalEvent) Type 

func (e *TransactionalEvent) Type() string

Type 返回事件类型（委托给内部事件）

type TransactionalEventPublisher 

type TransactionalEventPublisher struct {
	// contains filtered or unexported fields
}

TransactionalEventPublisher 事务事件发布器

提供便捷的事务事件发布 API。

使用示例： 

publisher := event.NewTransactionalEventPublisher(bus)
tx := publisher.BeginTransaction()
tx.PublishAfterCommit(&MyEvent{})
tx.Commit()

func NewTransactionalEventPublisher 

func NewTransactionalEventPublisher(bus *EventBus) *TransactionalEventPublisher

NewTransactionalEventPublisher 创建事务事件发布器

func (*TransactionalEventPublisher) BeginTransaction 

func (p *TransactionalEventPublisher) BeginTransaction() *TransactionContext

BeginTransaction 开始新事务

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.