264 lines
7.4 KiB
Go
264 lines
7.4 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"time"
|
||
|
||
"git.pyer.club/kingecg/gotidb/pkg/engine"
|
||
"git.pyer.club/kingecg/gotidb/pkg/engine/file"
|
||
"git.pyer.club/kingecg/gotidb/pkg/engine/memory"
|
||
)
|
||
|
||
func main() {
|
||
// 创建引擎注册表
|
||
registry := engine.NewEngineRegistry()
|
||
|
||
// 注册内存引擎和文件引擎
|
||
memory.Register(registry)
|
||
file.Register(registry)
|
||
|
||
// 创建临时目录用于文件引擎
|
||
tempDir, err := os.MkdirTemp("", "gotidb_example_*")
|
||
if err != nil {
|
||
log.Fatal("Failed to create temp dir:", err)
|
||
}
|
||
defer os.RemoveAll(tempDir)
|
||
|
||
// 创建内存引擎配置
|
||
memConfig := engine.NewEngineConfig().
|
||
WithMaxRetention(24 * time.Hour).
|
||
WithMaxPoints(1000)
|
||
|
||
// 创建文件引擎配置
|
||
fileConfig := engine.NewEngineConfig()
|
||
fileConfig.SetFileConfig(&engine.FileEngineConfig{
|
||
DataDir: tempDir,
|
||
SegmentSize: 1024 * 1024, // 1MB
|
||
CompactWindow: time.Hour,
|
||
MaxSegments: 10,
|
||
UseCompression: true,
|
||
CompressionLevel: 6,
|
||
})
|
||
|
||
// 创建内存引擎和文件引擎实例
|
||
memEng, err := registry.Create("memory", memConfig)
|
||
if err != nil {
|
||
log.Fatal("Failed to create memory engine:", err)
|
||
}
|
||
|
||
fileEng, err := registry.Create("file", fileConfig)
|
||
if err != nil {
|
||
log.Fatal("Failed to create file engine:", err)
|
||
}
|
||
|
||
// 打开引擎
|
||
if err := memEng.Open(); err != nil {
|
||
log.Fatal("Failed to open memory engine:", err)
|
||
}
|
||
defer memEng.Close()
|
||
|
||
if err := fileEng.Open(); err != nil {
|
||
log.Fatal("Failed to open file engine:", err)
|
||
}
|
||
defer fileEng.Close()
|
||
|
||
// 演示不同场景下的引擎使用
|
||
|
||
// 场景1：高频写入，短期存储 - 使用内存引擎
|
||
fmt.Println("\n=== 场景1：高频写入，短期存储（内存引擎）===")
|
||
demoHighFrequencyWrites(memEng)
|
||
|
||
// 场景2：长期存储，历史数据查询 - 使用文件引擎
|
||
fmt.Println("\n=== 场景2：长期存储，历史数据查询（文件引擎）===")
|
||
demoHistoricalData(fileEng)
|
||
|
||
// 场景3：聚合查询性能对比
|
||
fmt.Println("\n=== 场景3：聚合查询性能对比 ===")
|
||
demoAggregationComparison(memEng, fileEng)
|
||
|
||
// 打印引擎统计信息
|
||
printEngineStats("Memory Engine", memEng)
|
||
printEngineStats("File Engine", fileEng)
|
||
}
|
||
|
||
// 演示高频写入场景
|
||
func demoHighFrequencyWrites(eng engine.Engine) {
|
||
start := time.Now()
|
||
count := 1000
|
||
|
||
// 批量写入数据
|
||
var points []engine.DataPoint
|
||
for i := 0; i < count; i++ {
|
||
points = append(points, engine.DataPoint{
|
||
DeviceID: "sensor001",
|
||
MetricCode: "temperature",
|
||
Labels: map[string]string{
|
||
"location": "room1",
|
||
"floor": "1st",
|
||
},
|
||
Value: 25.5 + float64(i%10),
|
||
Timestamp: time.Now().Add(time.Duration(i) * time.Millisecond).UnixNano(),
|
||
})
|
||
}
|
||
|
||
if err := eng.WriteBatch(context.Background(), points); err != nil {
|
||
log.Printf("Failed to write batch: %v", err)
|
||
return
|
||
}
|
||
|
||
duration := time.Since(start)
|
||
fmt.Printf("写入 %d 个数据点耗时: %v (%.2f points/sec)\n",
|
||
count, duration, float64(count)/duration.Seconds())
|
||
|
||
// 查询最新数据
|
||
query := engine.NewQueryBuilder().
|
||
ForMetric("temperature").
|
||
WithTag("location", engine.OpEqual, "room1").
|
||
Build()
|
||
query.Type = engine.QueryTypeLatest
|
||
|
||
result, err := eng.Query(context.Background(), query)
|
||
if err != nil {
|
||
log.Printf("Failed to query latest data: %v", err)
|
||
return
|
||
}
|
||
|
||
if tsResult, ok := result.(*engine.TimeSeriesResult); ok {
|
||
fmt.Printf("最新数据点: %.2f (时间: %v)\n",
|
||
tsResult.Points[0].Value,
|
||
time.Unix(0, tsResult.Points[0].Timestamp).Format(time.RFC3339))
|
||
}
|
||
}
|
||
|
||
// 演示历史数据存储和查询场景
|
||
func demoHistoricalData(eng engine.Engine) {
|
||
// 写入跨越多个时间段的数据
|
||
now := time.Now()
|
||
var points []engine.DataPoint
|
||
for i := 0; i < 24; i++ {
|
||
points = append(points, engine.DataPoint{
|
||
DeviceID: "sensor002",
|
||
MetricCode: "power",
|
||
Labels: map[string]string{
|
||
"device": "solar_panel",
|
||
"unit": "watts",
|
||
},
|
||
Value: 100 + float64(i*50),
|
||
Timestamp: now.Add(time.Duration(-i) * time.Hour).UnixNano(),
|
||
})
|
||
}
|
||
|
||
if err := eng.WriteBatch(context.Background(), points); err != nil {
|
||
log.Printf("Failed to write historical data: %v", err)
|
||
return
|
||
}
|
||
|
||
// 查询24小时内的数据
|
||
query := engine.NewQueryBuilder().
|
||
ForMetric("power").
|
||
WithTimeRange(now.Add(-24*time.Hour).UnixNano(), now.UnixNano()).
|
||
WithTag("device", engine.OpEqual, "solar_panel").
|
||
Build()
|
||
|
||
result, err := eng.Query(context.Background(), query)
|
||
if err != nil {
|
||
log.Printf("Failed to query historical data: %v", err)
|
||
return
|
||
}
|
||
|
||
if tsResult, ok := result.(*engine.TimeSeriesResult); ok {
|
||
fmt.Printf("24小时内的数据点数量: %d\n", len(tsResult.Points))
|
||
if len(tsResult.Points) > 0 {
|
||
fmt.Printf("最早数据点: %.2f (时间: %v)\n",
|
||
tsResult.Points[0].Value,
|
||
time.Unix(0, tsResult.Points[0].Timestamp).Format(time.RFC3339))
|
||
fmt.Printf("最新数据点: %.2f (时间: %v)\n",
|
||
tsResult.Points[len(tsResult.Points)-1].Value,
|
||
time.Unix(0, tsResult.Points[len(tsResult.Points)-1].Timestamp).Format(time.RFC3339))
|
||
}
|
||
}
|
||
}
|
||
|
||
// 演示聚合查询性能对比
|
||
func demoAggregationComparison(memEng, fileEng engine.Engine) {
|
||
// 准备测试数据
|
||
now := time.Now()
|
||
var points []engine.DataPoint
|
||
for i := 0; i < 1000; i++ {
|
||
points = append(points, engine.DataPoint{
|
||
DeviceID: "sensor003",
|
||
MetricCode: "cpu_usage",
|
||
Labels: map[string]string{
|
||
"host": "server1",
|
||
},
|
||
Value: float64(30 + (i % 40)),
|
||
Timestamp: now.Add(time.Duration(-i) * time.Minute).UnixNano(),
|
||
})
|
||
}
|
||
|
||
// 写入两个引擎
|
||
if err := memEng.WriteBatch(context.Background(), points); err != nil {
|
||
log.Printf("Failed to write to memory engine: %v", err)
|
||
return
|
||
}
|
||
|
||
if err := fileEng.WriteBatch(context.Background(), points); err != nil {
|
||
log.Printf("Failed to write to file engine: %v", err)
|
||
return
|
||
}
|
||
|
||
// 创建聚合查询
|
||
query := engine.NewQueryBuilder().
|
||
ForMetric("cpu_usage").
|
||
WithTimeRange(now.Add(-24*time.Hour).UnixNano(), now.UnixNano()).
|
||
WithTag("host", engine.OpEqual, "server1").
|
||
WithAggregation(engine.AggAvg, 1*time.Hour).
|
||
Build()
|
||
|
||
// 测试内存引擎聚合性能
|
||
memStart := time.Now()
|
||
memResult, err := memEng.Query(context.Background(), query)
|
||
if err != nil {
|
||
log.Printf("Memory engine aggregation failed: %v", err)
|
||
return
|
||
}
|
||
memDuration := time.Since(memStart)
|
||
|
||
// 测试文件引擎聚合性能
|
||
fileStart := time.Now()
|
||
fileResult, err := fileEng.Query(context.Background(), query)
|
||
if err != nil {
|
||
log.Printf("File engine aggregation failed: %v", err)
|
||
return
|
||
}
|
||
fileDuration := time.Since(fileStart)
|
||
|
||
// 打印性能对比
|
||
fmt.Printf("内存引擎聚合查询耗时: %v\n", memDuration)
|
||
fmt.Printf("文件引擎聚合查询耗时: %v\n", fileDuration)
|
||
|
||
if memAgg, ok := memResult.(*engine.AggregateResult); ok {
|
||
fmt.Printf("内存引擎聚合组数: %d\n", len(memAgg.Groups))
|
||
}
|
||
if fileAgg, ok := fileResult.(*engine.AggregateResult); ok {
|
||
fmt.Printf("文件引擎聚合组数: %d\n", len(fileAgg.Groups))
|
||
}
|
||
}
|
||
|
||
// 打印引擎统计信息
|
||
func printEngineStats(name string, eng engine.Engine) {
|
||
stats := eng.Stats()
|
||
caps := eng.Capabilities()
|
||
|
||
fmt.Printf("\n=== %s 统计信息 ===\n", name)
|
||
fmt.Printf("数据点总数: %d\n", stats.PointsCount)
|
||
fmt.Printf("最后写入时间: %v\n", stats.LastWriteTime.Format(time.RFC3339))
|
||
fmt.Printf("支持压缩: %v\n", caps.SupportsCompression)
|
||
fmt.Printf("支持持久化: %v\n", caps.SupportsPersistence)
|
||
fmt.Printf("支持复制: %v\n", caps.SupportsReplication)
|
||
fmt.Printf("最大并发写入: %d\n", caps.MaxConcurrentWrites)
|
||
}
|