介绍Go中的for select case和chan 和goroutine

远方雪-

1332人浏览 · 2024-11-11 17:03:32

远方雪- · 2024-11-11 17:03:32 发布

好的，让我们详细分析您提供的代码片段：

for {
    select {
    case msg := <-sensorChan:
        // 处理消息的代码
    }
}

1. 整体结构概述

这段代码使用了 Go 语言的并发控制结构，结合无限循环 (for)、选择语句 (select)、以及 case 分支来处理从 sensorChan 通道接收的数据。让我们逐步拆解每个部分的功能和作用。

2. 逐行解析

2.1 `for { ... }`

作用：
- 这是一个无限循环，意味着它会一直运行，直到程序被显式终止（如通过 break、return、panic，或外部信号）。
用途：
- 在并发程序中，常用来持续监听和处理来自通道的数据或其他事件。

2.2 `select { ... }`

作用：
- select 语句用于等待多个通道操作中的一个完成。
- 它会阻塞当前 goroutine，直到其中一个 case 分支的通信操作可以进行，然后执行该 case 分支中的代码。
- 如果有多个 case 分支同时满足条件，Go 语言会随机选择其中一个执行，以避免偏向某个 case。
用途：
- 在需要同时监听多个通道或处理多种事件时，select 提供了一种优雅的方式来管理这些并发操作。

2.3 `case msg := <-sensorChan:`

作用：
- 这是一个接收操作，从 sensorChan 通道中接收一个消息并赋值给变量 msg。
- 如果 sensorChan 里有数据，select 会选择这个 case 分支并执行相应的代码。
- 如果 sensorChan 为空且没有关闭（即仍然可能接收数据），case 操作会阻塞，直到有数据可接收。
用途：
- sensorChan 通道通常用于在不同的 goroutine 之间传递数据。在这个上下文中，sensorChan 可能用于传递来自传感器的数据消息。

3. 详细流程说明

让我们结合整个代码片段，说明它的运行流程：

启动无限循环：
- 程序进入 for 循环，并持续运行。
进入 select 语句：
- 在每次循环迭代中，程序进入 select 语句，准备监听多个通道操作。
监听 sensorChan 通道：
- select 语句中有一个 case 分支，是监听 sensorChan 通道的接收操作。
- 如果 sensorChan 中有数据，程序会接收该数据，并将其赋值给变量 msg，然后执行对应的处理代码。
- 如果 sensorChan 中没有数据，且通道未关闭，程序会在此 case 分支处阻塞，直到有数据可接收。
处理接收到的消息：
- 一旦有数据从 sensorChan 被接收，msg 就包含了该数据。
- 接着，程序会执行 case 分支中的处理代码，对 msg 进行相应的处理（例如，写入数据库、发送 HTTP 请求等）。
循环继续：
- 处理完当前消息后，循环继续，重新进入 select 语句，等待下一个消息的到来。

4. 并发与单线程的关系

虽然 for 循环和 select 语句本身运行在一个 goroutine 中（可能是主 goroutine 或另一个专门的处理 goroutine），但整个程序可能包含多个 goroutine：

消息生产者：
- 例如，MQTT 客户端在独立的 goroutine 中接收消息，并将其发送到 sensorChan。
消息消费者：
- 这里的 for { select { ... } } 结构通常运行在一个专门的 goroutine 中，负责从 sensorChan 接收并处理消息。

5. 可能的后续代码行为

假设您的 case 分支代码如下：

case msg := <-sensorChan:
    currentTime := time.Now()
    lastWrite, exists := sensorLastWrite[msg.SensorID]

    // 检查是否需要处理这条数据
    if !exists || currentTime.Sub(lastWrite) >= time.Minute {
        // 处理数据
        fiberData := FiberOpticInclinometerData{
            Number:   msg.Data.SensorID,
            Datetime: parseTime(msg.Data.Time),
            Deep:     fmt.Sprintf("%.6f", msg.Data.SensorPosition),
            Lateral:  fmt.Sprintf("%.6f", msg.Data.Value),
            IfDamage: false,
        }

        // 写入数据库
        if enabledMySQL {
            if err := writeToMySQL(fiberData); err == nil {
                sensorLastWriteMutex.Lock()
                sensorLastWrite[msg.SensorID] = currentTime
                sensorLastWriteMutex.Unlock()

                fmt.Printf("Successfully processed sensor %s at %v\n",
                    msg.SensorID, currentTime.Format("2006-01-02 15:04:05"))
            }
        }

        // 发送HTTP请求
        if enabledHTTP {
            singlepointData := FiberOpticInclinometerData{
                Number:   fiberData.Number,
                Datetime: fiberData.Datetime,
                Deep:     fiberData.Deep,
                Lateral:  fiberData.Lateral,
                IfDamage: fiberData.IfDamage,
            }
            SendSensorData(urlSensor+singlepointURI, tokenf, []FiberOpticInclinometerData{singlepointData})
        }
    } else {
        fmt.Printf("Skipping sensor %s, last write was %v ago\n",
            msg.SensorID, currentTime.Sub(lastWrite))
    }

5.1 处理流程

获取当前时间：
- currentTime := time.Now()
检查传感器的上次写入时间：
- 使用 sensorLastWrite 映射，获取传感器 msg.SensorID 的上次写入时间 lastWrite。
- 检查该传感器是否存在于 sensorLastWrite 中，以及当前时间与上次写入时间的差值是否超过一分钟。
条件判断：
- 如果传感器没有写入过数据 (!exists)，或者 距离上次写入超过一分钟 (currentTime.Sub(lastWrite) >= time.Minute)：
  - 处理数据：
    - 创建 FiberOpticInclinometerData 结构体，填充传感器数据。
  - 写入数据库：
    - 调用 writeToMySQL(fiberData) 函数，将数据写入 MySQL 数据库。
    - 如果写入成功，更新 sensorLastWrite 中对应传感器的最后写入时间。
  - 发送 HTTP 请求：
    - 创建 singlepointData 结构体，发送 HTTP 请求到指定的 URL。
- 否则：
  - 打印跳过该传感器的数据处理日志，显示距离上次写入的时间差。

5.2 并发安全

互斥锁 (sensorLastWriteMutex)：
- 由于 sensorLastWrite 映射可能被多个 goroutine 访问（虽然在单线程处理模式下不太常见），使用互斥锁确保读写操作的并发安全。

6. `for`、`select` 和 `case` 的关系

for 循环：
- 在无限循环中持续运行，确保程序不停地监听和处理来自 sensorChan 的消息。
select 语句：
- 随着 for 循环的每次迭代，select 会检查所有 case 分支的条件，并执行第一个满足条件的分支。
- 在此情境下，只有一个 case 分支：接收 sensorChan 中的消息。
case 分支：
- case msg := <-sensorChan: 表示尝试从 sensorChan 中接收一个消息。
- 如果有消息可接收，执行 case 中的代码。
- 如果没有消息，且 sensorChan 未关闭，则阻塞等待。

7. 代码作为一个 goroutine 的运行

单独的 goroutine：
- 通常，这样的消息处理循环会被放置在一个单独的 goroutine 中运行，以便与其他并发操作（如 MQTT 消息接收、HTTP 请求等）并行进行。

示例：

func main() {
    // ... 初始化代码 ...

    // 启动 MQTT 订阅
    subscribeAndProcessBatch(mqttTopic)

    // 启动消息处理循环在一个单独的 goroutine 中
    go processDataBatch()

    // 主 goroutine 可能做其他事情，或者等待信号
    select {}
}