Go语言中的复杂事件处理（CEP）引擎探索

package main

import (
    "fmt"
    "sync"
    "time"
)

// Event 定义了一个通用的事件结构
type Event struct {
    Type      string    // 事件类型，如 "transaction", "login", "sensor_data"
    Timestamp time.Time // 事件发生时间
    Value     float64   // 事件携带的数值，例如交易金额
    Metadata  map[string]string // 其他元数据
}

// EventProcessor 模拟一个简单的CEP规则引擎
// 目标：检测在5秒内发生两次或更多“高价值交易”事件
func EventProcessor(eventStream <-chan Event, alertStream chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()

    // 用于存储最近的高价值事件
    var recentHighValueEvents []Event
    // 互斥锁保护 recentHighValueEvents
    var mu sync.Mutex

    fmt.Println("Event Processor Started...")

    for event := range eventStream {
        fmt.Printf("Received: Type=%s, Value=%.2f, Time=%s\n",
            event.Type, event.Value, event.Timestamp.Format("15:04:05"))

        // 检查是否为“高价值交易”事件
        if event.Type == "transaction" && event.Value > 1000.0 {
            mu.Lock()
            recentHighValueEvents = append(recentHighValueEvents, event)

            // 清理超出时间窗口的旧事件
            var cleanedEvents []Event
            for _, e := range recentHighValueEvents {
                if time.Since(e.Timestamp) <= 5*time.Second { // 5秒时间窗口
                    cleanedEvents = append(cleanedEvents, e)
                }
            }
            recentHighValueEvents = cleanedEvents

            // 检查是否满足模式：5秒内两次或更多高价值交易
            if len(recentHighValueEvents) >= 2 {
                alertMsg := fmt.Sprintf("ALERT! Detected %d high-value transactions (>1000) within 5 seconds! (Latest: %.2f)",
                    len(recentHighValueEvents), event.Value)
                alertStream <- alertMsg
                // 模式匹配后可以清空或保留，这里选择清空以检测新的模式
                recentHighValueEvents = nil
            }
            mu.Unlock()
        }
    }
    fmt.Println("Event Processor Finished.")
}

func main() {
    eventStream := make(chan Event)
    alertStream := make(chan string)
    var wg sync.WaitGroup

    // 启动事件处理器
    wg.Add(1)
    go EventProcessor(eventStream, alertStream, &wg)

    // 模拟事件生成器
    go func() {
        defer close(eventStream) // 所有事件发送完毕后关闭事件流
        defer close(alertStream) // 确保在所有处理完成后关闭警报流

        fmt.Println("Simulating events...")

        // 事件 1
        eventStream <- Event{"transaction", time.Now(), 500.0, nil}
        time.Sleep(1 * time.Second)

        // 事件 2 (高价值)
        eventStream <- Event{"transaction", time.Now(), 1200.0, nil}
        time.Sleep(1 * time.Second)

        // 事件 3 (普通)
        eventStream <- Event{"login", time.Now(), 0.0, nil}
        time.Sleep(1 * time.Second)

        // 事件 4 (高价值) - 此时应触发警报 (1200.0 和 1500.0 在5秒内)
        eventStream <- Event{"transaction", time.Now(), 1500.0, nil}
        time.Sleep(1 * time.Second)

        // 等待时间窗口过去
        time.Sleep(6 * time.Second)

        // 事件 5 (高价值) - 不会触发警报，因为之前的事件已过期
        eventStream <- Event{"transaction", time.Now(), 1100.0, nil}
        time.Sleep(1 * time.Second)

        fmt.Println("Event simulation finished.")
    }()

    // 接收并打印警报
    for alert := range alertStream {
        fmt.Println("--- ALERT ---:", alert)
    }

    wg.Wait() // 等待事件处理器完成
    fmt.Println("Program Exited.")
}