# 添加InfluxData倉庫
wget -q https://repos.influxdata.com/influxdata-archive.key
sudo gpg --dearmor -o /usr/share/keyrings/influxdata-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/influxdata-archive-keyring.gpg] https://repos.influxdata.com/debian stable main" | sudo tee /etc/apt/sources.list.d/influxdata.list
# 安裝InfluxDB 2.x
sudo apt update && sudo apt install influxdb2
# 啟動服務(wù)
sudo systemctl start influxdb
# 初始化配置
influx setup \
  --username myuser \
  --password mypassword \
  --org myorg \
  --bucket mybucket \
  --token mytoken \
  --retention 168h \
  --force

influx ping

go version

mkdir influxdb-demo && cd influxdb-demo
go mod init github.com/yourusername/influxdb-demo
go get github.com/influxdata/influxdb-client-go/v2

package main

import (
	"fmt"
	"log"
	"time"
	
	"github.com/influxdata/influxdb-client-go/v2"
)

func main() {
	// 初始化客戶端
	client := influxdb2.NewClient("http://localhost:8086", "mytoken")
	defer client.Close() // 確保程序退出時關(guān)閉連接
	
	// 檢查服務(wù)健康狀態(tài)
	health, err := client.Health(context.Background())
	if err != nil {
		log.Fatalf("Health check failed: %v", err)
	}
	fmt.Printf("InfluxDB health status: %s, version: %s\n", health.Status, *health.Version)
	
	// 更多操作將在后續(xù)添加...
}

func writeDataSync(client influxdb2.Client) {
	// 獲取同步寫入API
	writeAPI := client.WriteAPIBlocking("myorg", "mybucket")
	
	// 創(chuàng)建數(shù)據(jù)點 - 方式1: 使用完整構(gòu)造函數(shù)
	p1 := influxdb2.NewPoint(
		"temperature",
		map[string]string{"location": "room1", "sensor": "A1"},
		map[string]interface{}{"value": 23.5, "humidity": 45.0},
		time.Now(),
	)
	
	// 創(chuàng)建數(shù)據(jù)點 - 方式2: 使用流式API
	p2 := influxdb2.NewPointWithMeasurement("temperature").
		AddTag("location", "room2").
		AddTag("sensor", "B2").
		AddField("value", 22.1).
		AddField("humidity", 47.3).
		SetTime(time.Now().Add(-time.Minute))
	
	// 寫入數(shù)據(jù)點
	if err := writeAPI.WritePoint(context.Background(), p1); err != nil {
		log.Printf("Write point 1 failed: %v", err)
	}
	
	if err := writeAPI.WritePoint(context.Background(), p2); err != nil {
		log.Printf("Write point 2 failed: %v", err)
	}
	
	// 也可以直接寫入行協(xié)議
	line := `temperature,location=room3,sensor=C3 value=24.8,humidity=43.7`
	if err := writeAPI.WriteRecord(context.Background(), line); err != nil {
		log.Printf("Write line protocol failed: %v", err)
	}
}

func writeDataAsync(client influxdb2.Client) {
	// 獲取異步寫入API
	writeAPI := client.WriteAPI("myorg", "mybucket")
	
	// 設(shè)置錯誤處理通道
	errorsCh := writeAPI.Errors()
	go func() {
		for err := range errorsCh {
			log.Printf("Write error: %v", err)
		}
	}()
	
	// 模擬寫入100個數(shù)據(jù)點
	for i := 0; i < 100; i++ {
		p := influxdb2.NewPoint(
			"cpu_usage",
			map[string]string{"host": fmt.Sprintf("server%d", i%5)},
			map[string]interface{}{
				"user":   rand.Float64() * 30,
				"system": rand.Float64() * 20,
				"idle":   100 - rand.Float64()*50,
			},
			time.Now().Add(-time.Duration(i)*time.Second),
		)
		writeAPI.WritePoint(p)
	}
	
	// 確保所有緩沖數(shù)據(jù)都已寫入
	writeAPI.Flush()
}

func queryBasic(client influxdb2.Client) {
	queryAPI := client.QueryAPI("myorg")
	
	// 執(zhí)行Flux查詢
	query := `from(bucket:"mybucket")
		|> range(start: -1h)
		|> filter(fn: (r) => r._measurement == "temperature")
		|> filter(fn: (r) => r._field == "value")
		|> aggregateWindow(every: 5m, fn: mean)`
	
	result, err := queryAPI.Query(context.Background(), query)
	if err != nil {
		log.Fatalf("Query failed: %v", err)
	}
	
	// 處理查詢結(jié)果
	for result.Next() {
		if result.TableChanged() {
			fmt.Printf("\nTable: %s\n", result.TableMetadata().String())
		}
		fmt.Printf("Time: %v, Value: %v\n", 
			result.Record().Time(), 
			result.Record().Value())
	}
	
	if result.Err() != nil {
		log.Printf("Result processing error: %v", result.Err())
	}
}

func queryWithParams(client influxdb2.Client) {
	queryAPI := client.QueryAPI("myorg")
	
	params := map[string]interface{}{
		"start":    "-30m",
		"measurement": "cpu_usage",
		"min_value":   10.0,
	}
	
	query := `from(bucket:"mybucket")
		|> range(start: duration(v: params.start))
		|> filter(fn: (r) => r._measurement == params.measurement)
		|> filter(fn: (r) => r._value > params.min_value)`
	
	result, err := queryAPI.QueryWithParams(context.Background(), query, params)
	if err != nil {
		log.Fatalf("Query failed: %v", err)
	}
	
	// 處理結(jié)果...
}

func createCustomClient() influxdb2.Client {
	// 創(chuàng)建自定義HTTP客戶端
	httpClient := &http.Client{
		Timeout: 30 * time.Second,
		Transport: &http.Transport{
			MaxIdleConns:        10,
			MaxIdleConnsPerHost: 10,
			IdleConnTimeout:     90 * time.Second,
		},
	}
	
	// 使用自定義選項創(chuàng)建客戶端
	client := influxdb2.NewClientWithOptions(
		"http://localhost:8086",
		"mytoken",
		influxdb2.DefaultOptions().
			SetBatchSize(5000).          // 異步寫入批量大小
			SetFlushInterval(10000).     // 刷新間隔(毫秒)
			SetUseGZip(true).            // 啟用Gzip壓縮
			SetHTTPClient(httpClient).   // 自定義HTTP客戶端
			SetLogLevel(3),              // 日志級別
	)
	
	return client
}

func configureForHighVolume(client influxdb2.Client) {
	// 獲取寫入API時可以直接配置
	writeAPI := client.WriteAPIWithOptions(
		"myorg",
		"mybucket",
		api.WriteOptions{
			BatchSize:      10000,
			FlushInterval:  5000,
			RetryInterval:  2000,
			MaxRetries:     3,
			MaxRetryDelay:  15000,
			MaxRetryTime:   30000,
		},
	)
	
	// 使用writeAPI進(jìn)行寫入...
}

func optimizedQuery(client influxdb2.Client) {
	queryAPI := client.QueryAPI("myorg")
	
	// 創(chuàng)建帶超時的context
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	
	// 優(yōu)化后的查詢
	query := `from(bucket:"mybucket")
		|> range(start: -1h)
		|> filter(fn: (r) => r._measurement == "network" and r._field == "bytes_in")
		|> aggregateWindow(every: 1m, fn: mean)
		|> yield(name: "mean")`
	
	result, err := queryAPI.Query(ctx, query)
	// 處理結(jié)果...
}

package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"runtime"
	"time"
	
	"github.com/shirou/gopsutil/v3/cpu"
	"github.com/shirou/gopsutil/v3/mem"
	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api/write"
)

type SystemMonitor struct {
	client   influxdb2.Client
	writeAPI write.WriteAPI
	hostname string
}

func NewSystemMonitor(url, token, org, bucket string) *SystemMonitor {
	client := influxdb2.NewClient(url, token)
	return &SystemMonitor{
		client:   client,
		writeAPI: client.WriteAPI(org, bucket),
		hostname: getHostname(),
	}
}

func (m *SystemMonitor) StartCollecting(interval time.Duration) {
	// 設(shè)置錯誤處理
	go func() {
		for err := range m.writeAPI.Errors() {
			log.Printf("Write error: %v", err)
		}
	}()
	
	// 定時收集指標(biāo)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	
	for range ticker.C {
		m.collectMetrics()
	}
}

func (m *SystemMonitor) collectMetrics() {
	// 收集CPU使用率
	if cpuPercents, err := cpu.Percent(time.Second, false); err == nil {
		p := influxdb2.NewPointWithMeasurement("cpu").
			AddTag("host", m.hostname).
			AddField("usage_percent", cpuPercents[0]).
			SetTime(time.Now())
		m.writeAPI.WritePoint(p)
	}
	
	// 收集內(nèi)存信息
	if memInfo, err := mem.VirtualMemory(); err == nil {
		p := influxdb2.NewPointWithMeasurement("memory").
			AddTag("host", m.hostname).
			AddField("total", memInfo.Total).
			AddField("available", memInfo.Available).
			AddField("used_percent", memInfo.UsedPercent).
			SetTime(time.Now())
		m.writeAPI.WritePoint(p)
	}
	
	// 收集Goroutine數(shù)量
	p := influxdb2.NewPointWithMeasurement("go_runtime").
		AddTag("host", m.hostname).
		AddField("goroutines", runtime.NumGoroutine()).
		AddField("cgo_calls", runtime.NumCgoCall()).
		SetTime(time.Now())
	m.writeAPI.WritePoint(p)
	
	// 模擬應(yīng)用指標(biāo)
	p = influxdb2.NewPointWithMeasurement("app_metrics").
		AddTag("host", m.hostname).
		AddTag("service", "user_api").
		AddField("request_count", rand.Intn(1000)).
		AddField("error_count", rand.Intn(20)).
		AddField("response_time_ms", rand.Float64()*200).
		SetTime(time.Now())
	m.writeAPI.WritePoint(p)
}

func (m *SystemMonitor) Close() {
	m.writeAPI.Flush()
	m.client.Close()
}

func getHostname() string {
	// 實際實現(xiàn)中應(yīng)該獲取真實主機名
	return "server01"
}

func main() {
	monitor := NewSystemMonitor(
		"http://localhost:8086",
		"mytoken",
		"myorg",
		"mybucket",
	)
	defer monitor.Close()
	
	go monitor.StartCollecting(10 * time.Second)
	
	// 保持程序運行
	select {}
}

type SensorDataProcessor struct {
	client    influxdb2.Client
	writeAPI  write.WriteAPI
	batchSize int
}

func (p *SensorDataProcessor) ProcessData(dataCh <-chan SensorReading) {
	var points []*write.Point
	
	for reading := range dataCh {
		point := influxdb2.NewPointWithMeasurement("sensor_reading").
			AddTag("sensor_id", reading.SensorID).
			AddTag("location", reading.Location).
			AddTag("type", reading.Type).
			AddField("value", reading.Value).
			AddField("battery", reading.Battery).
			SetTime(reading.Timestamp)
		
		points = append(points, point)
		
		// 批量寫入
		if len(points) >= p.batchSize {
			p.writeAPI.WritePoints(points...)
			points = points[:0] // 清空切片但保留底層數(shù)組
		}
	}
	
	// 寫入剩余數(shù)據(jù)
	if len(points) > 0 {
		p.writeAPI.WritePoints(points...)
	}
	p.writeAPI.Flush()
}

type SensorReading struct {
	SensorID  string
	Location  string
	Type      string
	Value     float64
	Battery   float64
	Timestamp time.Time
}