将频道与 google pubsub poll 订阅者一起使用

Question

我正在尝试在 golang 中创建一个 google pubsub 订阅者，我一次接收 100 条消息，然后将它们写入 influx。我正在尝试使用频道来执行此操作：

package main

import (
    "os"
    "fmt"
    "cloud.google.com/go/pubsub"
    "log"
    "sync"
    "golang.org/x/net/context"
    "encoding/json"
    clnt "github.com/influxdata/influxdb/client/v2"
    "time"
)

type SensorData struct {
    Pressure      float64 `json:"pressure"`
    Temperature   float64 `json:"temperature"`
    Dewpoint      float64 `json:"dewpoint"`
    Timecollected int64   `json:"timecollected"`
    Latitude      float64 `json:"latitude"`
    Longitude     float64 `json:"longitude"`
    Humidity      float64 `json:"humidity"`
    SensorID      string  `json:"sensorId"`
    Zipcode       int     `json:"zipcode"`
    Warehouse     string  `json:"warehouse"`
    Area          string  `json:"area"`
}

type SensorPoints struct {
    SensorData      []SensorData
}

func main () {

    messages := make(chan SensorData, 100)

    // Create a new Influx HTTPClient
    c, err := clnt.NewHTTPClient(clnt.HTTPConfig{
        Addr:     "http://localhost:8086",
        Username: "user",
        Password: "pass",
    })
    if err != nil {
        log.Fatal(err)
    }


    // Create pubsub subscriber
    ctx := context.Background()
    proj := os.Getenv("GOOGLE_CLOUD_PROJECT")
    if proj == "" {
        fmt.Fprintf(os.Stderr, "GOOGLE_CLOUD_PROJECT environment variable must be set.\n")
        os.Exit(1)
    }
    client, err := pubsub.NewClient(ctx, proj)
    if err != nil {
        log.Fatalf("Could not create pubsub Client: %v", err)
    }
    const sub = "influxwriter"


    //create influx a blank batchpoint set
    bp, err := clnt.NewBatchPoints(clnt.BatchPointsConfig{
        Database:  "sensordata",
        Precision: "s",
    })
    if err != nil {
        log.Fatal(err)
    }



    // Pull messages via the subscription.
    go pullMsgs(client, sub, messages)
    if err != nil {
        log.Fatal(err)
    }

    writeInflux(messages, bp)

    c.Close()

}


func pullMsgs(client *pubsub.Client, name string, messages chan<- SensorData) {
    ctx := context.Background()

    // [START pubsub_subscriber_async_pull]
    // [START pubsub_quickstart_subscriber]
    // Consume 10 messages.

    var mu sync.Mutex
    var sensorinfos SensorPoints
    sensorinfo := &SensorData{}

    received := 0
    sub := client.Subscription(name)
    cctx, _ := context.WithCancel(ctx)
    err := sub.Receive(cctx, func(ctx context.Context, msg *pubsub.Message) {
        msg.Ack()

        json.Unmarshal(msg.Data, sensorinfo)

        //fmt.Println(string(msg.Data))
        //fmt.Println(sensorinfo.SensorID)

        sensorinfos.SensorData = append(sensorinfos.SensorData, *sensorinfo)

        mu.Lock()
        defer mu.Unlock()
        received++
        fmt.Println("rcv: ", received)
        messages <- *sensorinfo

    })
    if err != nil {
        fmt.Println(err)
    }
    // [END pubsub_subscriber_async_pull]
    // [END pubsub_quickstart_subscriber]
}

func writeInflux(sensorpoints <- chan SensorData, bp clnt.BatchPoints) {

    for p := range sensorpoints {

        // Create a point and add to batch
        tags := map[string]string{
            "sensorId": p.SensorID,
            "warehouse": p.Warehouse,
            "area": p.Area,
            "zipcode": string(p.Zipcode),
        }

        fields := map[string]interface{}{
            "pressure":   p.Pressure,
            "humidity": p.Humidity,
            "temperature":   p.Temperature,
            "dewpoint":   p.Dewpoint,
            "longitude":   p.Longitude,
            "latitude":   p.Latitude,
        }

        pt, err := clnt.NewPoint("sensordata", tags, fields, time.Unix(p.Timecollected, 0))
        if err != nil {
            log.Fatal(err)
        }
        bp.AddPoint(pt)


    }


}

但它并没有看到每个人都能通过最初的 pullMsgs 函数，只是继续在那里打印输出：

rcv:  1
rcv:  2
rcv:  3
rcv:  4
rcv:  5
rcv:  6
rcv:  7

我认为一旦通道满了，它应该阻塞直到通道被清空

这是我用作参考的 pubsub 拉取代码。

Answer 1

当您在频道上发送了所需数量的消息后，请关闭频道并取消上下文。尝试使用the documentation 中演示的技术，在收到一定数量的消息后取消。由于您的缓冲区为 100，并且您尝试一次使用 100 条消息，所以这就是数字。如果您希望程序退出，请关闭通道，以便 writeInflux 中的 for e := range ch 循环到达停止点并且不会阻塞等待更多元素添加到通道中。

请在the Go pubsub API doc 中注意这一点：

要终止对 Receive 的调用，请取消其上下文。

这不是阻碍你的主 goroutine 的原因，但你的 pullMsgs goroutine 不会在没有取消的情况下自行退出。

另外，检查Unmarshal 上的错误。如果此时您不想在代码中处理解组错误，请考虑更改通道类型并改为发送msg 或msg.Data，并在收到通道时解组。

cctx, cancel := context.WithCancel(ctx)
err := sub.Receive(cctx, func(ctx context.Context, msg *pubsub.Message) {
    msg.Ack()
    err := json.Unmarshal(msg.Data, sensorinfo)
    if err != nil {
         fmt.Printf("Failed to unmarshal: %s\n", err)
    }
    mu.Lock()
    defer mu.Unlock()
    received++
    fmt.Println("rcv: ", received)
    messages <- *sensorinfo
    if received == 100 {
        close(messages)  // no more messages will be sent on channel
        cancel()
    }