面试题：Go并发基础在复杂实时数据分析业务场景的适配策略

数据接收

使用net/http或MQTT库：

对于物联网设备，可以使用MQTT协议来接收数据。Go有优秀的MQTT客户端库，如eclipse/paho.mqtt.golang。示例代码：

package main

import (
    "fmt"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

func main() {
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://broker.example.com:1883")
    opts.SetClientID("go_mqtt_client")
    opts.SetDefaultPublishHandler(f)
    c := mqtt.NewClient(opts)
    if token := c.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }
    if token := c.Subscribe("data_topic", 0, nil); token.Wait() && token.Error() != nil {
        fmt.Println(token.Error())
        c.Disconnect(250)
    }
    select {}
}

对于日志文件，可以使用net/http包创建HTTP服务器接收日志数据。示例：

package main

import (
    "fmt"
    "net/http"
)

func logHandler(w http.ResponseWriter, r *http.Request) {
    // 读取日志数据
    // 这里简单示例，实际中需要处理大文件上传等复杂情况
    fmt.Fprintf(w, "Received log data")
}

func main() {
    http.HandleFunc("/log", logHandler)
    fmt.Println("Server listening on :8080")
    http.ListenAndServe(":8080", nil)
}

使用channel传递数据：接收到的数据通过channel传递到后续处理环节。例如：

dataCh := make(chan []byte, 1000)
// 在MQTT或HTTP处理函数中发送数据到channel
dataCh <- receivedData

数据预处理

创建预处理协程：

启动多个协程来处理数据预处理，如数据清洗、格式转换等。示例：

for i := 0; i < runtime.NumCPU(); i++ {
    go func() {
        for data := range dataCh {
            // 数据清洗，例如去除空值
            cleanedData := cleanData(data)
            // 格式转换，例如JSON到结构体
            preprocessedData := convertToStruct(cleanedData)
            preprocessCh <- preprocessedData
        }
    }()
}

select语句处理多个channel：

如果有多个数据源对应的channel，可以使用select语句在它们之间进行多路复用。例如：

var data1Ch, data2Ch chan []byte
go func() {
    for {
        select {
        case data := <-data1Ch:
            // 处理data1Ch的数据
        case data := <-data2Ch:
            // 处理data2Ch的数据
        }
    }
}()

分析计算

使用sync.WaitGroup管理协程：

启动多个协程进行分析计算，使用sync.WaitGroup确保所有计算完成。示例：

var wg sync.WaitGroup
for i := 0; i < runtime.NumCPU(); i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        for data := range preprocessCh {
            result := analyzeData(data)
            resultCh <- result
        }
    }()
}

高效内存管理：

使用对象池（sync.Pool）来复用对象，减少内存分配。例如：

var dataPool = sync.Pool{
    New: func() interface{} {
        return &MyDataStruct{}
    },
}
func analyzeData(data []byte) *MyDataStruct {
    obj := dataPool.Get().(*MyDataStruct)
    // 填充数据到obj
    // 分析完成后
    dataPool.Put(obj)
    return obj
}

结果输出

结果持久化或展示：

将分析结果输出到数据库、文件或通过HTTP接口展示。例如，将结果写入数据库：

for result := range resultCh {
    // 使用数据库连接库，如`gorm`写入数据库
    err := db.Create(result).Error
    if err != nil {
        // 错误处理
        fmt.Println("Error writing result to database:", err)
    }
}

容错性：

在每个环节添加错误处理，对于不可恢复的错误，如数据库连接失败，进行重试或记录日志。例如：

maxRetries := 3
for i := 0; i < maxRetries; i++ {
    err := db.Create(result).Error
    if err == nil {
        break
    }
    time.Sleep(time.Second)
}

可扩展性：
- 水平扩展：通过增加更多的服务器节点，使用负载均衡器（如Nginx）将数据接收请求分发到不同节点。
- 垂直扩展：升级硬件资源，如增加内存、CPU等。同时，优化代码，减少不必要的计算和内存占用，以提高单机处理能力。例如，在数据预处理和分析计算环节，使用更高效的算法和数据结构。

星途面试题库

面试题：Go并发基础在复杂实时数据分析业务场景的适配策略

知识考点

面试题答案

数据接收

数据预处理

分析计算

结果输出