使用 goroutine 处理数据并发送响应

本篇文章向大家介绍《使用 goroutine 处理数据并发送响应》，主要包括，具有一定的参考价值，需要的朋友可以参考一下。

我想利用 go 中的并发将数据发送到 goroutine，以便使用通道进行处理和计算。数据点在一个函数中一个接一个地出现，该函数可以是 main 函数，也可以是某个 senddata 函数。 如果可能，我希望从主函数中完成数据的发送。

我想将数据从发送函数发送到一个 goroutine，其中数据存储在一个切片中（我们将此 goroutine 称为 getdata）。某些计算是在此切片上完成的。 达到某个条件后（取决于切片），我希望 goroutine 向 senddata 函数发出信号，表明某批数据点的处理已完成。 现在，senddata 函数不断发送数据通过通道指向 getdata goroutine，其中不断构建新切片，并且当达到条件时发送信号 - 处理完成并且整个序列不断重复。

举个例子，假设数字形式的数据从 senddata 发送到 getdata。条件是 getdata 接收到的数字的运行平均值应等于 4。让我们将以下数字序列作为我们的数据 - []int{3, 2, 3, 8, 2, 1, 1, 1, 15}。这里，第一批数字是 {3, 2, 3, 8}，因为这些数字按照这个顺序发送到 getdata 后，发现收到数字 8 后，数字的运行平均值等于 4通过 getdata。然后它向 senddata 发送信号。发送数据的过程再次按数字顺序开始，下一批是 {2, 1, 1, 1, 15}。这里，在getdata接收到数字15后，它发现运行平均值等于4，再次向senddata发送信号。 ( 这是一个非常基本的示例 - 在我的实际用例中，输入数据和条件更加复杂。我有将在 senddata 中实时读取的数据。这里每个数据点都是按顺序读取的，但每个数据点都会到达在前一个之后的几微秒处。因此这里数据的到达速度很快，我不想在这个函数中进行太多的处理和计算。此外，我希望保持并发完整，因为在函数中读取数据的地方，数据到达的速度很快。而且，我不希望因为在完成处理的 goroutine 中处理数据而错过这里的数据读取。） p>

这是我尝试构建代码的方式：

package main

import (
    "fmt"
)

func main() {
    ch := make(chan int)
    go sendData(ch)
    go getData(ch)
}

func sendData(ch chan int) {
    syntheticData := []int{3, 2, 3, 8, 2, 1, 1, 1, 15}
    for _, data := range syntheticData {
            ch <- data
        }
}

func getData(ch chan int) {
        dataArr := []int{}
        dataArr = append( dataArr, <-ch )
        fmt.Println(dataArr)

        if mean(dataArr) == 4{
          close(ch)
        }

}

func sum(array []int) int {
    var result int = 0
    for _, v := range array {
        result += v
    }

    return result
}


func mean(array []int) float64 {
    sumArr := float64(sum(array)) / float64(len(array))
    return sumArr
}

我没有通过上面的代码实现我想要的功能。 如何在 go 中实现所需的功能？

解决方案

你只需要一个外部接收goroutine，例如getdata 然后 main goroutine 将使用名为 ch 的通道发送数据，并且您需要一个缓冲通道来发送信号，例如batchcompleted 和 waitgroup 等待 getdata 同步（当它完成时）。
仅此而已，请尝试 it：

package main

import (
    "fmt"
    "sync"
)

func main() {
    wg := &sync.waitgroup{}
    ch := make(chan int)
    batchcompleted := make(chan struct{}, 1) // non-blocking signaling channel
    wg.add(1)
    go getdata(ch, batchcompleted, wg)

    syntheticdata := []int{3, 2, 3, 8, 2, 1, 1, 1, 15}
    i := 0
    check := func() {
        select {
        case <-batchcompleted:
            i++
            fmt.println(i, " batch completed")
        default:
        }
    }
    for _, data := range syntheticdata {
        ch <- data
        check()
    }
    close(ch)
    wg.wait()
    check()
}

func getdata(ch chan int, batchcompleted chan struct{}, wg *sync.waitgroup) {
    defer wg.done()
    a := []int{}
    sum, n := 0, 0
    for v := range ch {
        sum += v
        n++
        a = append(a, v)
        if sum == 4*n {
            batchcompleted <- struct{}{}
            fmt.println(a)
            sum, n = 0, 0
            a = a[:0]
        }
    }
    if len(a) > 0 {
        fmt.println("remaining data:", a)
    }
}

输出：

[3 2 3 8]
1  batch completed
[2 1 1 1 15]
2  batch completed

如果我正确理解你的问题，这应该可以正常工作。但这不是一个好的解决方案。我已经注释掉了代码，以便可以理解。

package main

import "fmt"

func main() {
    // Data channel
    ch := make(chan int)
    // Batch close/create channel for signalling
    bsig := make(chan struct{})

    go put(ch, bsig)
    get(ch, bsig, 4)
}

func put(ch chan<- int, bsig chan struct{}) {
    // Listen for batch creation
    go func() {
        n := 1
        // Receives signal from get on the bidrectional
        // channel, and ping back when done on the same
        // channel.
        for range bsig {
            fmt.Printf("Batch #%d\n", n)
            bsig <- struct{}{}
            n++
        }
    }()

    // Steam of numbers
    stream := []int{3, 2, 3, 8, 2, 1, 1, 1, 15}
    for _, s := range stream {
        ch <- s
    }
    close(ch)
}

func get(ch <-chan int, bsig chan struct{}, mean int) {
    num := 0
    sum := 0
    batch := make([]int, 0)
    for s := range ch {
        sum += s
        num++
        batch = append(batch, s)
        // Matches the mean, close batch
        // and signal the sender
        if sum/num == mean {
            // Send ping
            bsig <- struct{}{}
            // Wait for pong?
            <-bsig
            // Print the batch
            fmt.Println(batch)
            // Reset the batch
            sum = 0
            num = 0
            batch = batch[:0]
        }
    }
    // Close the channel
    close(bsig)
}

到这里，我们也就讲完了《使用 goroutine 处理数据并发送响应》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！