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

Posted 2023-02-16

【中文标题】将频道与 google pubsub poll 订阅者一起使用

【英文标题】：using channels with google pubsub poll subscriber

【发布时间】：2019-01-14 00:05:00

【问题描述】：

我正在尝试在 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 拉取代码。

【问题讨论】：

回复：“我认为一旦通道满了，它应该阻塞直到通道被清空”......几乎。一旦通道已满，发送操作将阻塞，直到通道未满——不必为空。那么你看到了什么，rcv: 100 它卡住了？程序是否退出？您是否将打印语句作为writeInflux 函数的第一行并确认它永远不会被调用？顺便说一句，err 在pullMsgs 调用之后被检查之前似乎没有更新。

@jrefior，naw，程序没有退出...随着消息数不断增加，它会继续运行rcv: n...再看一下谷歌文档（@987654321 @)，看起来订阅者本身是非阻塞的，但我不确定如何真正处理它？

【参考方案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()
    

【讨论】：

所以如果我只想编程只是在消息被放入 pubsub 时永远运行我该怎么办..我试图删除 cancel() 但后来得到了这个：panic: send on closed channel

我也尝试像这样重新创建频道：` for var messages = make(chan SensorData, 100) // 通过订阅拉取消息。 pullMsgs(client, sub, messages) if err != nil log.Fatal(err) writeInflux(messages, bp) ` 但这似乎也不起作用......我如何保持这个运行？跨度>