进阶示例

本篇收录自定义函数、性能模式等进阶用法。基础入门见快速开始。

自定义函数

StreamSQL 支持注册自定义函数（UDF），注册后可在 SQL 里像内置函数一样使用。

package main

import (
    "fmt"
    "math"
    "time"

    "github.com/rulego/streamsql"
    "github.com/rulego/streamsql/functions"
    "github.com/rulego/streamsql/utils/cast"
)

func main() {
    registerCustomFunctions() // 先注册 UDF

    ssql := streamsql.New()
    defer ssql.Stop()

    sql := `SELECT
                device,
                square(value)         AS squared_value,
                f_to_c(temperature)   AS celsius,
                circle_area(radius)   AS area
            FROM stream
            WHERE value > 0`
    if err := ssql.Execute(sql); err != nil {
        panic(err)
    }

    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("自定义函数结果: %v\n", results)
    })

    testData := []map[string]interface{}{
        {"device": "sensor1", "value": 5.0, "temperature": 68.0, "radius": 3.0},
        {"device": "sensor2", "value": 10.0, "temperature": 86.0, "radius": 2.5},
    }
    for _, data := range testData {
        ssql.Emit(data)
        time.Sleep(200 * time.Millisecond)
    }
    time.Sleep(500 * time.Millisecond)
}

// 注册自定义函数
func registerCustomFunctions() {
    // 数学函数：平方
    functions.RegisterCustomFunction(
        "square", functions.TypeMath, "数学函数", "计算平方", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            val := cast.ToFloat64(args[0])
            return val * val, nil
        },
    )
    // 华氏度转摄氏度
    functions.RegisterCustomFunction(
        "f_to_c", functions.TypeConversion, "温度转换", "华氏度转摄氏度", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            fahrenheit := cast.ToFloat64(args[0])
            return (fahrenheit - 32) * 5 / 9, nil
        },
    )
    // 圆面积
    functions.RegisterCustomFunction(
        "circle_area", functions.TypeMath, "几何计算", "计算圆的面积", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            radius := cast.ToFloat64(args[0])
            if radius < 0 {
                return nil, fmt.Errorf("半径必须为正数")
            }
            return math.Pi * radius * radius, nil
        },
    )
}

要点：

• functions.RegisterCustomFunction(name, type, category, desc, minArgs, maxArgs, fn) 注册一个 UDF；返回 (any, error)。
• 函数类型：TypeMath / TypeConversion / TypeString / TypeDateTime / TypeAggregation / TypeAnalytical / TypeCustom。
• 注册后即可在 SELECT/WHERE 中像内置函数一样调用（如 square(value)、f_to_c(temperature)）。
• 参数用 cast.ToFloat64 等做容错转换；注册需在 Execute 之前完成。

性能模式

StreamSQL 提供多种预置性能模式，按场景选择：

package main

import (
    "fmt"

    "github.com/rulego/streamsql"
)

func main() {
    // 高性能模式：大缓冲 + 扩展策略，适合高吞吐
    ssqlHighPerf := streamsql.New(streamsql.WithHighPerformance())
    defer ssqlHighPerf.Stop()

    // 低延迟模式：小缓冲 + 快速响应，适合实时告警
    ssqlLowLatency := streamsql.New(streamsql.WithLowLatency())
    defer ssqlLowLatency.Stop()

    sql := "SELECT deviceId, AVG(temperature) FROM stream GROUP BY deviceId, TumblingWindow('5s')"
    ssqlHighPerf.Execute(sql)
    ssqlLowLatency.Execute(sql)

    fmt.Println("不同性能模式已启动")
}

模式	适合场景	特点
默认 New()	通用	平衡的缓冲与 worker 配置
WithHighPerformance()	高吞吐	大缓冲、扩展（expand）溢出策略
WithLowLatency()	实时响应	小缓冲、block 溢出、低延迟

需要更细粒度控制时，用 streamsql.NewStreamWithCustomPerformance(config) 自定义缓冲大小、worker 数、溢出策略（drop/block/expand）等。

📚 相关文档

• 快速开始
• API 参考 —— 性能配置项、RegisterCustomFunction