Go语言中实现周期性通道清空与高级缓冲策略

package main

import (
    "fmt"
    "math/rand"
    "time"
)

// fillchan 协程：周期性向通道写入随机整数
func fillchan(commch chan int) {
    // 使用 for range time.Tick 避免 time.Tick 造成的资源泄露
    for range time.Tick(300 * time.Millisecond) {
        commch <- rand.Int()
    }
}

// drainchan 协程：非阻塞地清空通道所有当前数据
func drainchan(commch chan int) {
    for {
        select {
        case e := <-commch: // 尝试从通道读取数据
            fmt.Printf("取出的数据: %d\n", e)
        default: // 如果通道为空，则立即返回，避免阻塞
            return
        }
    }
}

func main() {
    commch := make(chan int, 1000) // 创建一个带缓冲的通道
    go fillchan(commch)             // 启动数据填充协程

    // 主协程：周期性触发通道清空
    // 使用 for range time.Tick 避免 time.Tick 造成的资源泄露
    for range time.Tick(1000 * time.Millisecond) {
        fmt.Println("--- 周期性清空开始 ---")
        drainchan(commch) // 调用清空函数
        fmt.Println("--- 周期性清空结束 ---")
    }
}

package main

import (
    "fmt"
    "math/rand"
    "time"
)

// dataProcessor 协程：接收数据，并在接收到flush信号时处理缓冲区内容
func dataProcessor(inputCh <-chan int, flushCh <-chan struct{}, outputCh chan<- []int) {
    buffer := make([]int, 0, 100) // 内部缓冲区
    for {
        select {
        case data := <-inputCh: // 接收输入数据
            buffer = append(buffer, data)
            // 可以在这里设置缓冲区大小限制，达到限制时自动flush
            if len(buffer) >= 50 { // 示例：缓冲区达到50个元素时自动flush
                fmt.Printf("缓冲区满，自动处理 %d 个元素\n", len(buffer))
                outputCh <- buffer
                buffer = make([]int, 0, 100) // 重置缓冲区
            }
        case <-flushCh: // 接收到清空信号
            if len(buffer) > 0 {
                fmt.Printf("收到清空信号，处理 %d 个元素\n", len(buffer))
                outputCh <- buffer
                buffer = make([]int, 0, 100) // 重置缓冲区
            } else {
                fmt.Println("收到清空信号，但缓冲区为空。")
            }
        }
    }
}

func main() {
    inputCh := make(chan int, 100)
    flushCh := make(chan struct{}) // 清空触发通道
    outputCh := make(chan []int)   // 处理后的数据输出通道

    go dataProcessor(inputCh, flushCh, outputCh)

    // 模拟数据生成
    go func() {
        for i := 0; i < 200; i++ {
            inputCh <- rand.Intn(1000)
            time.Sleep(50 * time.Millisecond)
        }
        close(inputCh) // 模拟数据生成结束
    }()

    // 模拟周期性发送清空信号
    go func() {
        for range time.Tick(1 * time.Second) {
            flushCh <- struct{}{} // 发送清空信号
        }
    }()

    // 接收并打印处理后的数据
    for processedData := range outputCh {
        fmt.Printf("已处理数据批次: %v\n", processedData)
    }

    // 注意：这里需要更完善的机制来优雅地关闭所有协程和通道
    // 例如，使用 context.Context 或额外的退出通道
}

package main

import (
    "fmt"
    "time"
)

// overwritingBuffer 协程：实现一个固定大小的覆盖式缓冲区
func overwritingBuffer(inputCh <-chan int, outputCh chan<- int, bufferSize int) {
    buffer := make([]int, 0, bufferSize) // 内部切片作为缓冲区

    for {
        select {
        case data := <-inputCh: // 尝试从输入通道读取
            if len(buffer) < bufferSize {
                buffer = append(buffer, data) // 缓冲区未满，直接添加
            } else {
                // 缓冲区已满，丢弃最旧的数据，添加新数据
                buffer = append(buffer[1:], data)
                fmt.Printf("缓冲区满，丢弃旧数据，添加新数据: %d\n", data)
            }
        case outputCh <- buffer[0]: // 尝试向输出通道写入最旧的数据
            // 成功写入后，移除已发送的数据
            buffer = buffer[1:]
            fmt.Printf("发送数据并移除: %d\n", buffer[0])
        default: // 如果输入和输出都无法进行，则等待
            // 避免CPU空转，可以短暂休眠或等待特定事件
            if len(buffer) == 0 {
                // 如果缓冲区为空，且没有新的输入，则阻塞等待输入
                data := <-inputCh
                buffer = append(buffer, data)
            } else {
                // 如果缓冲区不为空，但输出通道阻塞，且没有新的输入，
                // 此时可以等待输出，或者根据策略决定是否丢弃更多旧数据
                time.Sleep(10 * time.Millisecond) // 简单示例：短暂休眠
            }
        }
    }
}

func main() {
    input := make(chan int)
    output := make(chan int)
    bufferSize := 5

    go overwritingBuffer(input, output, bufferSize)

    // 模拟生产者：快速生产数据
    go func() {
        for i := 0; i < 20; i++ {
            input <- i
            time.Sleep(100 * time.Millisecond)
        }
    }()

    // 模拟消费者：慢速消费数据
    go func() {
        for {
            data := <-output
            fmt.Printf("消费者收到: %d\n", data)
            time.Sleep(500 * time.Millisecond) // 慢速消费
        }
    }()

    time.Sleep(10 * time.Second) // 运行一段时间
}