Pubsub.pull 请求无法正常工作 - go

亲爱的编程学习爱好者，如果你点开了这篇文章，说明你对《Pubsub.pull 请求无法正常工作 - go》很感兴趣。本篇文章就来给大家详细解析一下，主要介绍一下，希望所有认真读完的童鞋们，都有实质性的提高。

我正在尝试使用 go 客户端库一次从 pub-sub 订阅中提取 1 条消息。但即使消息存在于订阅中，消息也不会拉取拉取请求。订阅者正在等待所有消息被处理。

我正在尝试 go 中的基本代码，其中一次提取一条消息。我使用了两个实例，并在两个实例的后台运行了脚本（以创建订阅者）4 次。我设置了 ack_deadline 10 秒。 我期待这样的结果，就像每个订阅者在确认一条消息后应该从订阅中获取下一条消息。但直到最后一条消息处理完成后，消息才会拉到实例上。 为什么一条消息处理完成后没有拉取消息？据我所知，不应该对实例或订阅者有任何依赖。 让 mi 知道任何其他更改或需要设置的参数。 提前致谢。

这是一个实例的日志：

2019/10/21 05:22:07 got message: message 0 at 2019-10-21 05:22:07.022772532 
2019/10/21 05:22:11 got message: message 1 at 2019-10-21 05:22:11.330566981 
2019/10/21 05:22:14 got message: message 2 at 2019-10-21 05:22:14.803031569 
2019/10/21 05:22:18 got message: message 3 at 2019-10-21 05:22:18.452912271 
2019/10/21 05:38:39 acking message: message 3 at 2019-10-21 05:38:39.471739478 
2019/10/21 05:39:10 acking message: message 0 at 2019-10-21 05:39:10.039336794 
2019/10/21 05:41:22 acking message: message 1 at 2019-10-21 05:41:22.351124342 
2019/10/21 05:50:31 acking message: message 2 at 2019-10-21 05:50:31.829087762 
2019/10/21 05:50:39 got message: message 13 at 2019-10-21 05:50:39.005916608
2019/10/21 05:50:39 got message: message 11 at 2019-10-21 05:50:39.00623238 
2019/10/21 05:50:39 got message: message 15 at 2019-10-21 05:50:39.007216256
2019/10/21 05:50:39 got message: message 12 at 2019-10-21 05:50:39.008066257

第二个实例的日志：

2019/10/21 05:22:29 got message: message 4 at 2019-10-21 05:22:29.331569077 
2019/10/21 05:22:33 got message: message 5 at 2019-10-21 05:22:33.018801275 
2019/10/21 05:22:36 got message: message 6 at 2019-10-21 05:22:36.803434547 
2019/10/21 05:22:40 got message: message 7 at 2019-10-21 05:22:40.409314927 
2019/10/21 05:39:38 acking message: message 4 at 2019-10-21 05:39:38.349619635 
2019/10/21 05:42:42 acking message: message 6 at 2019-10-21 05:42:42.819874065 
2019/10/21 05:47:40 acking message: message 5 at 2019-10-21 05:47:40.049128075 
2019/10/21 05:50:38 acking message: message 7 at 2019-10-21 05:50:38.42874031 
2019/10/21 05:50:39 got message: message 8 at 2019-10-21 05:50:39.005090906 
2019/10/21 05:50:39 got message: message 9 at 2019-10-21 05:50:39.005334146 
2019/10/21 05:50:39 got message: message 16 at 2019-10-21 05:50:39.006427796 
2019/10/21 05:50:39 got message: message 14 at 2019-10-21 05:50:39.007231713

package main
// [START pubsub_publish_with_error_handling_that_scales]
import (
    "context"
    "fmt"
    "os"
    "log"
    "time"
    "math/rand"
    pubsub "cloud.google.com/go/pubsub/apiv1"
    pubsubpb "google.golang.org/genproto/googleapis/pubsub/v1"
)


func main(){
    f, _:= os.OpenFile("testlogfile", os.O_RDWR | os.O_CREATE | os.O_APPEND, 0666)
    defer f.Close()
    log.SetOutput(f)
    rand.Seed(time.Now().UTC().UnixNano())
    pullMsgs("sureline-dev-1264", "sub7")
}

func random(min, max int) int {
    return rand.Intn(max - min) + min
}

func pullMsgs(projectID, subscriptionID string) error {
    ctx := context.Background()
    client, err := pubsub.NewSubscriberClient(ctx)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()
    sub := fmt.Sprintf("projects/%s/subscriptions/%s", projectID, subscriptionID)
// Be sure to tune the MaxMessages parameter per your project's needs, and accordingly
// adjust the ack behavior below to batch acknowledgements.
    req := pubsubpb.PullRequest{
        Subscription: sub,
        MaxMessages:  1,
    }

    fmt.Println("Listening..")

    for {
        res, err := client.Pull(ctx, &req)
        if err != nil {
            log.Fatal(err)
        }

    // client.Pull returns an empty list if there are no messages available in the
    // backlog. We should skip processing steps when that happens.
        if len(res.ReceivedMessages) == 0 {
            continue
        }

        var recvdAckIDs []string
        for _, m := range res.ReceivedMessages {
            recvdAckIDs = append(recvdAckIDs, m.AckId)
        }

        var done = make(chan struct{})
        var delay = 0 * time.Second // Tick immediately upon reception
        var ackDeadline = 10 * time.Second

    // Continuously notify the server that processing is still happening on this batch.
        go func() {
            for {
                select {
                case <-ctx.Done():
                    return
                case <-done:
                    return
                case <-time.After(delay):
                    err := client.ModifyAckDeadline(ctx, &pubsubpb.ModifyAckDeadlineRequest{
                        Subscription:       sub,
                        AckIds:             recvdAckIDs,
                        AckDeadlineSeconds: int32(ackDeadline.Seconds()),
                    })
                    if err != nil {
                        log.Fatal(err)
                    }
                    delay = ackDeadline - 5*time.Second // 5 seconds grace period.
                }
            }
        }()

        for _, m := range res.ReceivedMessages {
            // Process the message here, possibly in a goroutine.
            log.Printf("Got message: %s at %v", string(m.Message.Data), time.Now())
            fmt.Printf("Got message: %s at %v", string(m.Message.Data), time.Now())
            myrand := random(240, 420)
            log.Printf("Sleeping %d seconds...\n", myrand)
            time.Sleep(time.Duration(myrand)*time.Second)
            err := client.Acknowledge(ctx, &pubsubpb.AcknowledgeRequest{
                Subscription: sub,
                AckIds:       []string{m.AckId},
            })
            log.Printf("Acking message: %s at %v", string(m.Message.Data), time.Now())
            fmt.Printf("Acking message: %s at %v", string(m.Message.Data), time.Now())
            if err != nil {
                log.Fatal(err)
            }
        }

        close(done)
    }
}

我期望输出就像在第一条消息处理完成后从订阅中获取下一条消息。它不应该依赖于任何其他实例。

解决方案

尝试以这种方式一次提取一条消息是 Cloud Pub/Sub 的反模式。就您而言，您的订阅者最终可能会与不同的服务器进行通信，这些服务器已分配消息以发送给连接到它们的订阅者。 Cloud Pub/Sub 期望同时从使用此方法接收消息的客户端接收多个拉取请求。因此，您应该同时有多个未完成的拉取请求，或者您应该使用 asynchronous pull via the Receive method。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Pubsub.pull 请求无法正常工作 - go》文章吧，也可关注golang学习网公众号了解相关技术文章。