API参考
# API参考
本章提供了StreamSQL的完整API参考文档，包括核心接口、配置选项、函数库等详细信息。
# 核心API
# Streamsql 主类
# 构造函数
func New(options ...Option) *Streamsql
1
创建新的StreamSQL实例。
参数：
options- 可选配置项
返回值：
*Streamsql- StreamSQL实例
示例：
// 默认配置
ssql := streamsql.New()
// 高性能配置
ssql := streamsql.New(streamsql.WithHighPerformance())
// 自定义配置
ssql := streamsql.New(
streamsql.WithLogLevel(logger.DEBUG),
streamsql.WithDiscardLog(),
)
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# Execute
func (s *Streamsql) Execute(sql string) error
1
执行SQL查询，启动流处理。
参数：
sql- SQL查询语句
返回值：
error- 执行错误，成功时为nil
示例：
sql := "SELECT deviceId, AVG(temperature) FROM stream GROUP BY deviceId, TumblingWindow('5m')"
err := ssql.Execute(sql)
if err != nil {
log.Fatal(err)
}
1
2
3
4
5
2
3
4
5
# Emit
func (s *Streamsql) Emit(data map[string]interface{})
1
向数据流异步添加数据。
参数：
data- 数据记录，必须为map[string]interface{}类型
示例：
data := map[string]interface{}{
"deviceId": "sensor001",
"temperature": 25.5,
"timestamp": time.Now(),
}
ssql.Emit(data)
1
2
3
4
5
6
2
3
4
5
6
# EmitSync
func (s *Streamsql) EmitSync(data map[string]interface{}) (map[string]interface{}, error)
1
同步处理数据并立即返回结果，仅支持非聚合查询。
参数：
data- 数据记录，必须为map[string]interface{}类型
返回值：
map[string]interface{}- 处理结果，如果不匹配过滤条件返回nilerror- 处理错误
示例：
data := map[string]interface{}{
"deviceId": "sensor001",
"temperature": 25.5,
"timestamp": time.Now(),
}
result, err := ssql.EmitSync(data)
if err != nil {
log.Printf("处理错误: %v", err)
} else if result != nil {
fmt.Printf("处理结果: %v", result)
}
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# IsAggregationQuery
func (s *Streamsql) IsAggregationQuery() bool
1
检查当前查询是否为聚合查询。
返回值：
bool- 是否为聚合查询
示例：
if ssql.IsAggregationQuery() {
fmt.Println("当前查询包含聚合操作")
} else {
fmt.Println("当前查询为简单查询")
}
1
2
3
4
5
2
3
4
5
# Stream
func (s *Streamsql) Stream() *stream.Stream
1
获取底层流处理实例。
返回值：
*stream.Stream- 流处理实例
示例：
// 通常推荐使用 Streamsql 上的便捷方法
ssql.AddSink(func(results []map[string]interface{}) {
fmt.Printf("结果: %v\n", results)
})
1
2
3
4
2
3
4
# GetStats
func (s *Streamsql) GetStats() map[string]int64
1
获取流处理统计信息。
返回值：
map[string]int64- 统计信息映射
示例：
stats := ssql.GetStats()
fmt.Printf("处理数据量: %d\n", stats["processed_count"])
1
2
2
# Stop
func (s *Streamsql) Stop()
1
停止流处理并清理资源。
示例：
defer ssql.Stop()
1
# AddSink
func (s *Streamsql) AddSink(sink func([]map[string]interface{}))
1
添加结果处理回调函数。
参数：
sink- 结果处理回调函数，接收[]map[string]interface{}类型的结果数据
示例：
ssql.AddSink(func(results []map[string]interface{}) {
fmt.Printf("处理结果: %v\n", results)
})
1
2
3
2
3
# PrintTable
func (s *Streamsql) PrintTable()
1
便捷方法，自动添加一个将结果以表格形式打印到控制台的 sink 函数（先打印列名，再打印数据行，类似数据库输出）。
示例：
ssql.PrintTable()
// 输出格式：
// +--------+----------+
// | device | max_temp |
// +--------+----------+
// | aa | 30.0 |
// | bb | 22.0 |
// +--------+----------+
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# AddSyncSink
func (s *Streamsql) AddSyncSink(sink func([]map[string]interface{}))
1
添加同步结果处理回调函数。与 AddSink 不同，同步 sink 在结果处理协程中按顺序执行，适合对执行顺序敏感的场景（多次调用的多个 sink 串行调用）。
参数：
sink- 结果处理回调函数，接收[]map[string]interface{}类型的结果数据
示例：
ssql.AddSyncSink(func(results []map[string]interface{}) {
// 顺序敏感的处理：先写库，再发通知
saveToDatabase(results)
})
1
2
3
4
2
3
4
# TriggerWindow
func (s *Streamsql) TriggerWindow()
1
手动触发当前窗口立即输出结果，不再等待自然触发（时间到/计数满）。
主要用于：
- 测试：需要窗口确定性地立即输出，避免
time.Sleep等待窗口自然触发 - 显式刷新钩子：业务需要在某个时刻强制 flush 当前窗口（如关机前、外部事件触发）
重要说明：
- 仅对基于时间的窗口（TumblingWindow / SlidingWindow / SessionWindow）有效
CountingWindow按 count 触发，Trigger()为空实现（设计如此）- 非聚合（直通）查询没有窗口，调用为安全空操作，不会 panic
- 数据需先进入窗口后再调用（首次 Emit 后窗口才会初始化）
示例：
ssql := streamsql.New()
defer ssql.Stop()
ssql.Execute("SELECT deviceId, COUNT(*) AS cnt FROM stream GROUP BY deviceId, TumblingWindow('5s')")
ssql.AddSink(func(rows []map[string]interface{}) {
fmt.Printf("结果: %v\n", rows)
})
ssql.Emit(map[string]interface{}{"deviceId": "d1"})
time.Sleep(200 * time.Millisecond) // 让数据进入窗口
ssql.TriggerWindow() // 立即触发，不等 5s
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# GetStats / GetDetailedStats
func (s *Streamsql) GetStats() map[string]int64
func (s *Streamsql) GetDetailedStats() map[string]interface{}
1
2
2
获取流处理统计信息 / 详细的性能统计信息（包含缓冲区使用、丢弃数等）。
示例：
stats := ssql.GetStats()
fmt.Printf("处理数据量: %d\n", stats["processed_count"])
detailed := ssql.GetDetailedStats()
fmt.Printf("详细统计: %v\n", detailed)
1
2
3
4
5
2
3
4
5
# ToChannel
func (s *Streamsql) ToChannel() <-chan []map[string]interface{}
1
返回结果通道，用于异步获取处理结果。
返回值：
<-chan []map[string]interface{}- 只读的结果通道，如果未执行SQL则返回nil
示例：
// 获取结果通道
resultChan := ssql.ToChannel()
if resultChan != nil {
go func() {
for results := range resultChan {
fmt.Printf("异步结果: %v\n", results)
}
}()
}
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 配置选项
# 性能配置
# WithHighPerformance
func WithHighPerformance() Option
1
使用高性能配置，适用于需要最大吞吐量的场景。
示例：
ssql := streamsql.New(streamsql.WithHighPerformance())
1
# WithLowLatency
func WithLowLatency() Option
1
使用低延迟配置，适用于实时交互应用。
示例：
ssql := streamsql.New(streamsql.WithLowLatency())
1
# WithCustomPerformance
func WithCustomPerformance(config types.PerformanceConfig) Option
1
使用自定义性能配置。
参数：
config- 自定义性能配置
示例：
config := types.DefaultPerformanceConfig()
config.BufferConfig.DataChannelSize = 2000
ssql := streamsql.New(streamsql.WithCustomPerformance(config))
1
2
3
2
3
# 日志配置
# WithLogLevel
func WithLogLevel(level logger.Level) Option
1
设置日志级别。
参数：
level- 日志级别（DEBUG, INFO, WARN, ERROR, OFF）
示例：
ssql := streamsql.New(streamsql.WithLogLevel(logger.DEBUG))
1
# WithDiscardLog
func WithDiscardLog() Option
1
禁用日志输出（生产环境推荐）。
示例：
ssql := streamsql.New(streamsql.WithDiscardLog())
1
# 缓冲区配置
# WithBufferSizes
func WithBufferSizes(dataChannelSize, resultChannelSize, windowOutputSize int) Option
1
设置自定义缓冲区大小（会切换到 custom 性能模式，其余参数取默认值）。
参数：
dataChannelSize- 数据输入通道大小resultChannelSize- 结果输出通道大小windowOutputSize- 窗口输出缓冲大小
示例：
ssql := streamsql.New(streamsql.WithBufferSizes(2000, 1000, 500))
1
# 溢出策略配置
# WithOverflowStrategy
func WithOverflowStrategy(strategy string, blockTimeout time.Duration) Option
1
设置缓冲区溢出策略。
参数：
strategy- 溢出策略："drop"- 丢弃策略（默认）。缓冲区满时丢弃最旧的数据腾出空间，保留最新数据"block"- 阻塞策略。缓冲区满时阻塞写入方，直到有空间或超时"expand"- 扩展策略（仅高性能预设启用）。缓冲区满时按增长因子扩容
blockTimeout- 阻塞超时时间（仅block策略有效）
示例：
// 丢弃策略（默认），适用于容忍少量数据丢失的高吞吐场景
ssql := streamsql.New(streamsql.WithOverflowStrategy("drop", 5*time.Second))
// 阻塞策略，适用于不能丢数据但可容忍背压的场景
ssql := streamsql.New(streamsql.WithOverflowStrategy("block", 5*time.Second))
1
2
3
4
5
2
3
4
5
# 工作池配置
# WithWorkerConfig
func WithWorkerConfig(sinkPoolSize, sinkWorkerCount, maxRetryRoutines int) Option
1
设置工作池配置。
参数：
sinkPoolSize- 结果处理池大小sinkWorkerCount- 工作线程数maxRetryRoutines- 最大重试协程数
示例：
ssql := streamsql.New(streamsql.WithWorkerConfig(100, 10, 5))
1
# 监控配置
# WithMonitoring
func WithMonitoring(updateInterval time.Duration, enableDetailedStats bool) Option
1
启用详细监控。
参数：
updateInterval- 统计更新间隔enableDetailedStats- 是否启用详细统计
示例：
ssql := streamsql.New(streamsql.WithMonitoring(10*time.Second, true))
1
# Stream-Table JOIN API
StreamSQL 支持 stream-table JOIN（流表连接），用于用静态或缓存的元数据丰富流数据。SQL 中通过 JOIN 子句声明，Go 侧通过以下方法注册表数据源。必须在 Execute 之后调用。
限制
当前版本（v0.5）：JOIN 仅用于流数据丰富（enrichment），运行在非窗口路径上。JOIN 与聚合/窗口组合会被拒绝（Execute 返回 "JOIN with aggregation/window is not supported"）。如需窗口聚合，请去掉 JOIN。
# RegisterTable
func (s *Streamsql) RegisterTable(name string, rows []map[string]interface{}, keyFields ...string) (*stream.MemoryTableSource, error)
1
注册内存元数据表，用于 stream-table JOIN。
name— 表名，需与 SQL 中JOIN ... ON引用的表名一致rows— 初始数据行keyFields— 索引键字段。省略时自动从 JOIN ON 子句的表侧字段推导（单键或复合键均可），调用方无需重复声明；显式传入可覆盖
示例：
// SQL: SELECT ... FROM stream JOIN meta ON deviceId = m.deviceId
ssql.Execute(`SELECT deviceId, m.location, temperature
FROM stream JOIN meta ON deviceId = m.deviceId`)
// 自动从 ON 推导 key（推荐）
meta, _ := ssql.RegisterTable("meta", rows)
// 或显式指定复合键
meta, _ := ssql.RegisterTable("meta", rows, "deviceId", "tenant")
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# RegisterTableSource
func (s *Streamsql) RegisterTableSource(src stream.TableSource) error
1
注册自定义表源（文件 / 数据库 / Redis / HTTP 等）。实现方自行负责数据加载、刷新和清理；Lookup 必须并发安全。
# UpsertTable
func (s *Streamsql) UpsertTable(name string, row map[string]interface{}) error
1
向已注册的内存表新增或替换一行。注意：表是快照式查询，仅影响此调用之后发出的行。
# Stream()
func (s *Streamsql) Stream() *stream.Stream
1
返回底层流处理实例，用于访问更底层的能力（窗口、结果通道、生命周期）。通常推荐使用 Streamsql 上的便捷方法（AddSink/AddSyncSink/ToChannel 等）而非直接操作底层实例。
# 窗口API
窗口通过 SQL 的 GROUP BY 子句中的窗口函数声明，由 StreamSQL 根据 WITH 子句自动创建，通常不需要手动实例化。下面列出底层接口供高级用法和扩展参考。
# 窗口类型常量
const (
TypeTumbling = "tumbling"
TypeSliding = "sliding"
TypeCounting = "counting"
TypeSession = "session"
)
1
2
3
4
5
6
2
3
4
5
6
# 窗口构造工厂
func CreateWindow(config types.WindowConfig) (Window, error)
1
根据 WindowConfig 创建对应的窗口实例。WindowConfig 携带窗口类型、参数、时间戳字段（TsProp）、GroupByKeys、CountStateTTL、PerformanceConfig 等配置。
# Window 接口
type Window interface {
Add(item interface{}) // 向窗口添加一条数据
Reset() // 重置窗口状态
Start() // 启动窗口处理协程
Stop() // 停止窗口并清理资源
OutputChan() <-chan []types.Row // 获取窗口输出通道
SetCallback(callback func([]types.Row))// 设置同步回调（结果同时写入 OutputChan）
Trigger() // 手动触发当前窗口（TriggerWindow 的底层调用）
GetStats() map[string]int64 // 获取窗口统计（sentCount/droppedCount/bufferSize 等）
}
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
方法说明：
Add(item)— 向窗口添加数据，内部提取时间戳并路由到对应分组缓冲Start()/Stop()— 启停窗口后台协程；Stop后Add被忽略OutputChan()— 窗口聚合结果的输出通道，流处理主循环从此读取Trigger()— 手动触发当前窗口立即输出。注意： 时间窗口（Tumbling/Sliding/Session）支持；CountingWindow按 count 触发，此方法为空实现GetStats()— 返回sentCount、droppedCount、bufferSize、bufferUsed等统计
关于 Trigger
Streamsql.TriggerWindow() 是对 Window.Trigger() 的安全封装：未执行 SQL、非窗口查询、窗口为 nil 时都是安全空操作，不会 panic。
# 函数系统API
# 函数注册
# RegisterCustomFunction
func RegisterCustomFunction(
name string,
funcType FunctionType,
category string,
description string,
minArgs int,
maxArgs int,
handler FunctionHandler,
) error
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
注册自定义函数。
参数：
name- 函数名funcType- 函数类型category- 函数分类description- 函数描述minArgs- 最小参数数量maxArgs- 最大参数数量handler- 函数处理器
返回值：
error- 注册错误
示例：
err := functions.RegisterCustomFunction(
"my_function",
functions.TypeMath,
"数学计算",
"自定义数学函数",
2, 2,
func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
// 函数实现
return result, nil
},
)
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# Register
func Register(function Function) error
1
注册函数实例。
参数：
function- 函数实例
返回值：
error- 注册错误
# Unregister
func Unregister(name string)
1
注销函数。
参数：
name- 函数名
示例：
functions.Unregister("my_function")
1
# Get
func Get(name string) (Function, bool)
1
获取函数实例。
参数：
name- 函数名
返回值：
Function- 函数实例bool- 是否存在
# GetByType
func GetByType(funcType FunctionType) []Function
1
根据类型获取函数列表。
参数：
funcType- 函数类型
返回值：
[]Function- 函数实例列表
# ListAll
func ListAll() map[string]Function
1
列出所有已注册的函数。
返回值：
map[string]Function- 函数名到函数实例的映射
# Execute
func Execute(name string, args []interface{}) (interface{}, error)
1
执行指定名称的函数。
参数：
name- 函数名args- 函数参数
返回值：
interface{}- 执行结果error- 执行错误
# 函数类型
type FunctionType string
const (
TypeMath FunctionType = "math"
TypeString FunctionType = "string"
TypeConversion FunctionType = "conversion"
TypeDateTime FunctionType = "datetime"
TypeAggregation FunctionType = "aggregation"
TypeAnalytical FunctionType = "analytical"
TypeWindow FunctionType = "window"
TypeCustom FunctionType = "custom"
)
1
2
3
4
5
6
7
8
9
10
11
12
2
3
4
5
6
7
8
9
10
11
12
# 函数处理器
type FunctionHandler func(ctx *FunctionContext, args []interface{}) (interface{}, error)
1
# FunctionContext
type FunctionContext struct {
Data map[string]interface{} // 当前数据行
WindowInfo *WindowInfo // 窗口信息（聚合/窗口函数适用）
Extra map[string]interface{} // 附加上下文
}
1
2
3
4
5
2
3
4
5
字段说明：
Data- 当前处理的数据行（旧文档误作CurrentRow，已修正）WindowInfo- 窗口信息（仅窗口/聚合函数有效）Extra- 额外上下文信息
# WindowInfo
type WindowInfo struct {
WindowStart int64 // 窗口起始时间戳
WindowEnd int64 // 窗口结束时间戳
RowCount int // 窗口内行数
}
1
2
3
4
5
2
3
4
5
# 类型转换工具（utils/cast）
自定义函数中常用的类型转换位于 github.com/rulego/streamsql/utils/cast 包：
| 函数 | 说明 |
|---|---|
cast.ToFloat64E(v) | 转 float64，带错误返回（ToFloat64 不带） |
cast.ToIntE(v) / cast.ToInt(v) | 转 int |
cast.ToInt64E(v) / cast.ToInt64(v) | 转 int64 |
cast.ToStringE(v) / cast.ToString(v) | 转 string |
cast.ToBoolE(v) / cast.ToBool(v) | 转 bool |
cast.ToDurationE(v) | 解析时长字符串（"5s" 等） |
cast.ConvertIntToTime(ts, unit) | 整数时间戳按单位转 time.Time |
# 聚合器API
# 聚合类型
type AggregateType string
const (
Sum AggregateType = "sum"
Count AggregateType = "count"
Avg AggregateType = "avg"
Max AggregateType = "max"
Min AggregateType = "min"
Median AggregateType = "median"
Percentile AggregateType = "percentile"
StdDev AggregateType = "stddev"
StdDevS AggregateType = "stddevs"
Var AggregateType = "var"
VarS AggregateType = "vars"
Collect AggregateType = "collect"
FirstValue AggregateType = "first_value"
LastValue AggregateType = "last_value"
MergeAgg AggregateType = "merge_agg"
Deduplicate AggregateType = "deduplicate"
WindowStart AggregateType = "window_start"
WindowEnd AggregateType = "window_end"
Latest AggregateType = "latest"
Lag AggregateType = "lag"
ChangedCol AggregateType = "changed_col"
HadChanged AggregateType = "had_changed"
Expression AggregateType = "expression"
PostAggregation AggregateType = "post_aggregation"
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
AggregateType 是函数包（github.com/rulego/streamsql/functions）定义的别名类型，聚合器包通过 aggregator.AggregateType 引用。用户通常不直接接触这些常量——SQL 中的 SUM/AVG/COUNT 等函数名会被解析器自动映射。
# Aggregator 接口
type Aggregator interface {
Add(data interface{}) error
Put(key string, val interface{}) error
GetResults() ([]map[string]interface{}, error)
Reset()
RegisterExpression(field, expression string, fields []string, evaluator func(data interface{}) (interface{}, error))
}
1
2
3
4
5
6
7
2
3
4
5
6
7
聚合器由 StreamSQL 内部按 Config.SelectFields 创建，每个分组一个实例。Add 累积数据，GetResults 返回该分组的聚合结果行。
# 表达式API
表达式能力由 github.com/rulego/streamsql/expr 包提供，SQL 解析器在内部使用，用户通常无需直接调用。
# Expression
Expression 是一个结构体（非接口），由 NewExpression 创建：
func NewExpression(exprStr string) (*Expression, error)
1
参数：
exprStr- 表达式字符串（如temperature * 1.8 + 32）
返回值：
*Expression- 表达式实例（内部优先用自定义解析器，失败时回退到 expr-lang）error- 语法错误时返回
# 日志API
# Logger 接口
type Logger interface {
Debug(format string, args ...interface{})
Info(format string, args ...interface{})
Warn(format string, args ...interface{})
Error(format string, args ...interface{})
SetLevel(level Level)
}
1
2
3
4
5
6
7
2
3
4
5
6
7
# 日志级别
type Level int
const (
DEBUG Level = iota // 详细调试信息
INFO // 一般信息（默认）
WARN // 警告
ERROR // 错误
OFF // 关闭日志
)
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 创建日志器
# NewLogger
func NewLogger(level Level, output io.Writer) Logger
1
创建新的日志器（printf 风格）。注意参数顺序：先 level，后 output。
# NewLoggerWithFormat
func NewLoggerWithFormat(level Level, output io.Writer, format Format) Logger
1
创建指定输出格式的日志器。format 取 TextFormat（logfmt 风格，默认）或 JSONFormat（每行一个 JSON 对象）。
# NewDiscardLogger
func NewDiscardLogger() Logger
1
创建丢弃所有输出的日志器（用于禁用日志的生产环境，或 WithDiscardLog()）。
# 全局默认日志器
func SetDefault(logger Logger) // 设置全局默认日志器
func GetDefault() Logger // 获取全局默认日志器
1
2
2
# 类型定义
# Config
type Config struct {
WindowConfig WindowConfig // 窗口配置
GroupFields []string // GROUP BY 字段
SelectFields map[string]aggregator.AggregateType // 聚合字段（别名→类型）
FieldAlias map[string]string // 输出字段→输入字段映射
SimpleFields []string // 非聚合 SELECT 字段
FieldExpressions map[string]FieldExpression // 字段表达式
PostAggExpressions []PostAggregationExpression // 聚合后表达式
FieldOrder []string // SELECT 原始顺序（用于表格输出）
Where string // WHERE 条件
Having string // HAVING 条件
NeedWindow bool // 是否需要窗口
Distinct bool // DISTINCT 去重
Limit int // 结果数上限
Projections []Projection // SELECT 投影
OrderBy []OrderByField // ORDER BY 排序键（按批次）
JoinConfigs []JoinConfig // stream-table JOIN 配置
SourceAlias string // FROM 别名（如 FROM stream AS s）
PerformanceConfig PerformanceConfig // 性能配置
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
Config 由 SQL 解析器（rsql.Parse）生成，作为内部执行计划的载体，通常不需要用户手动构造——通过 Execute(sql) 间接消费。
# WindowConfig
type WindowConfig struct {
Type string // "tumbling" / "sliding" / "counting" / "session"
Params []interface{} // 窗口函数参数（如 ['5m'] 或 [100]）
TsProp string // 事件时间字段名（空则使用处理时间）
TimeUnit time.Duration // 整数时间戳的解析单位（默认 ms）
TimeCharacteristic TimeCharacteristic // ProcessingTime（默认）或 EventTime
MaxOutOfOrderness time.Duration // 最大乱序时间（事件时间，默认 0）
WatermarkInterval time.Duration // Watermark 推进间隔（默认 200ms）
AllowedLateness time.Duration // 窗口触发后允许延迟（默认 0，立即关闭）
IdleTimeout time.Duration // 空闲源超时（默认 0，禁用）
CountStateTTL time.Duration // CountingWindow 状态 TTL（默认 0，禁用）
GroupByKeys []string // 分组字段列表（支持多字段）
PerformanceConfig PerformanceConfig // 性能配置
Callback func([]Row) // 直接回调（不经过输出通道）
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
2
3
4
5
6
7
8
9
10
11
12
13
14
15
对应 SQL WITH 子句的字段映射：TsProp ← TIMESTAMP，TimeUnit ← TIMEUNIT，MaxOutOfOrderness ← MAXOUTOFORDERNESS，AllowedLateness ← ALLOWEDLATENESS，IdleTimeout ← IDLETIMEOUT，CountStateTTL ← STATETTL。
# FieldExpression
type FieldExpression struct {
Field string
Expression string
Fields []string
}
1
2
3
4
5
2
3
4
5
# Projection
type Projection struct {
OutputName string
SourceType ProjectionSourceType
InputName string
}
type ProjectionSourceType int
const (
SourceGroupKey ProjectionSourceType = iota
SourceAggregateResult
SourceWindowProperty
)
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
# PerformanceConfig
type PerformanceConfig struct {
BufferConfig BufferConfig
OverflowConfig OverflowConfig
WorkerConfig WorkerConfig
MonitoringConfig MonitoringConfig
}
1
2
3
4
5
6
2
3
4
5
6
# BufferConfig
type BufferConfig struct {
DataChannelSize int
ResultChannelSize int
WindowOutputSize int
EnableDynamicResize bool
MaxBufferSize int
UsageThreshold float64
}
1
2
3
4
5
6
7
8
2
3
4
5
6
7
8
# OverflowConfig
type OverflowConfig struct {
Strategy string // "drop"（默认）、"block"、"expand"
BlockTimeout time.Duration // "block" 策略的阻塞超时
AllowDataLoss bool // 是否允许丢数据（drop 时为 true）
ExpansionConfig ExpansionConfig // "expand" 策略的扩容参数
}
1
2
3
4
5
6
2
3
4
5
6
Strategy 取值："drop"（丢弃最旧）、"block"（阻塞写入方）、"expand"（扩容缓冲）。
# ExpansionConfig
type ExpansionConfig struct {
GrowthFactor float64 // 扩容增长因子（默认 1.5）
MinIncrement int // 最小扩容量（默认 1000）
TriggerThreshold float64 // 触发扩容的使用率阈值（默认 0.9）
ExpansionTimeout time.Duration // 扩容超时（默认 5s）
}
1
2
3
4
5
6
2
3
4
5
6
# WorkerConfig
type WorkerConfig struct {
SinkPoolSize int
SinkWorkerCount int
MaxRetryRoutines int
}
1
2
3
4
5
2
3
4
5
# MonitoringConfig
type MonitoringConfig struct {
EnableMonitoring bool
StatsUpdateInterval time.Duration
EnableDetailedStats bool
WarningThresholds WarningThresholds
}
1
2
3
4
5
6
2
3
4
5
6
# WarningThresholds
type WarningThresholds struct {
DropRateWarning float64
DropRateCritical float64
BufferUsageWarning float64
BufferUsageCritical float64
}
1
2
3
4
5
6
2
3
4
5
6
# 配置预设函数
# DefaultPerformanceConfig
func DefaultPerformanceConfig() PerformanceConfig
1
返回默认性能配置，平衡性能和资源使用。关键默认值：数据通道 1000、结果通道 100、窗口输出 50、溢出策略 drop、sink 工作协程 2。
# HighPerformanceConfig
func HighPerformanceConfig() PerformanceConfig
1
返回高性能配置预设，优化吞吐量：大数据通道（5000）、expand 溢出策略、sink 工作协程 4，并开启监控。对应 Option WithHighPerformance()。
# LowLatencyConfig
func LowLatencyConfig() PerformanceConfig
1
返回低延迟配置预设，优化响应延迟：小数据通道（100）、block 溢出策略、1 秒统计间隔。对应 Option WithLowLatency()。
# 错误处理
StreamSQL 的错误通过 error 返回（fmt.Errorf 包装上下文），不导出哨兵错误变量。常见错误来源：
| 调用 | 错误场景 |
|---|---|
Execute | SQL 解析失败；同一实例第二次调用 Execute（返回 Execute() has already been called）；JOIN 与聚合/窗口组合（返回 JOIN with aggregation/window is not supported）；窗口参数非法（如 CountingWindow 阈值 ≤ 0、CountingWindow 配合 EventTime） |
EmitSync | 聚合查询下调用（返回 synchronous mode only supports non-aggregation queries）；stream 未初始化 |
RegisterTable / RegisterTableSource / UpsertTable | 在 Execute 之前调用（返回 Execute must be called before ...）；表名未在 JOIN 中声明 |
示例：
err := ssql.Execute(sql)
if err != nil {
log.Printf("执行失败: %v", err)
return
}
1
2
3
4
5
2
3
4
5
# 完整示例
package main
import (
"fmt"
"log"
"time"
"github.com/rulego/streamsql"
"github.com/rulego/streamsql/functions"
"github.com/rulego/streamsql/logger"
"github.com/rulego/streamsql/utils/cast"
)
func main() {
// 1. 创建StreamSQL实例
ssql := streamsql.New(
streamsql.WithLogLevel(logger.INFO),
)
defer ssql.Stop()
// 2. 注册自定义函数
err := functions.RegisterCustomFunction(
"celsius_to_fahrenheit",
functions.TypeConversion,
"温度转换",
"摄氏度转华氏度",
1, 1,
func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
celsius, err := cast.ToFloat64E(args[0])
if err != nil {
return nil, err
}
fahrenheit := celsius*9/5 + 32
return fahrenheit, nil
},
)
if err != nil {
log.Fatal(err)
}
// 3. 执行SQL查询
sql := `SELECT deviceId,
AVG(temperature) as avg_celsius,
AVG(celsius_to_fahrenheit(temperature)) as avg_fahrenheit,
COUNT(*) as sample_count,
window_start() as window_start
FROM stream
WHERE temperature > 0
GROUP BY deviceId, TumblingWindow('1m')`
err = ssql.Execute(sql)
if err != nil {
log.Fatal(err)
}
// 4. 添加结果处理（接收 []map[string]interface{}）
ssql.AddSink(func(results []map[string]interface{}) {
fmt.Printf("聚合结果: %v\n", results)
})
// 5. 发送数据
devices := []string{"sensor001", "sensor002", "sensor003"}
go func() {
for i := 0; i < 100; i++ {
for _, device := range devices {
data := map[string]interface{}{
"deviceId": device,
"temperature": 20.0 + rand.Float64()*15,
"timestamp": time.Now(),
}
ssql.Emit(data)
}
time.Sleep(5 * time.Second)
}
}()
// 6. 等待结果
time.Sleep(5 * time.Minute)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# 最佳实践
# 错误处理
// 检查SQL执行错误
err := ssql.Execute(sql)
if err != nil {
log.Printf("SQL执行失败: %v", err)
return
}
// 检查函数注册错误
err = functions.RegisterCustomFunction(...)
if err != nil {
log.Printf("函数注册失败: %v", err)
return
}
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
# 资源管理
// 确保资源释放
ssql := streamsql.New()
defer ssql.Stop()
// 函数注册和注销
err := functions.RegisterCustomFunction(...)
if err == nil {
defer functions.Unregister("function_name")
}
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 并发安全
// StreamSQL实例是并发安全的
var ssql = streamsql.New()
go func() {
for {
ssql.Emit(generateData())
time.Sleep(100 * time.Millisecond)
}
}()
go func() {
for {
ssql.Emit(generateData())
time.Sleep(200 * time.Millisecond)
}
}()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
在 GitHub 上编辑此页 (opens new window)
上次更新: 2026/07/11, 12:44:39