Kafka消费者的Goroutine

学习Golang要努力，但是不要急！今天的这篇文章《Kafka消费者的Goroutine》将会介绍到等等知识点，如果你想深入学习Golang，可以关注我！我会持续更新相关文章的，希望对大家都能有所帮助！

我目前有一个程序，它创建一个大小为 1 的工作组，然后调用 startworker：

package main

import (
    "db_write_consumer/db"
    "db_write_consumer/worker"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    sigchan := make(chan os.signal, 1)
    signal.notify(sigchan, syscall.sigint, syscall.sigterm)
    mysqlclient, _ := db.newmysqldbclient("root", "", "localhost", 3306, "testbase")
    workers := worker.creategroup("localhost:9092", "testgroup", 1)
    for _, w := range workers {
        w_ := w
        worker.startworker(w_, []string{"test-topic"}, sigchan, mysqlclient)
    }
}

creategroup 的写入位置：

func creategroup(bootstrapservers string, groupid string, numworkers int) []*kafka.consumer {
    consumers := []*kafka.consumer{}
    for i := 0; i < numworkers; i++ {
        c := newworker(bootstrapservers, groupid)
        consumers = append(consumers, c)
    }
    return consumers
}

startworker 是这样写的：

func StartWorker(c *kafka.Consumer, topics []string, sigchan chan os.Signal, mySQLClient *sql.DB) {
    _ = c.SubscribeTopics(topics, nil)
    fmt.Println(c)
    run := true
    for run {
        select {
        case sig := <-sigchan:
            fmt.Printf("Caught signal %v: terminating\n", sig)
            run = false
        default:
            ev, _ := c.ReadMessage(100)
            if ev == nil {
                continue
            }
            msg := &pb.Person{}
            proto.Unmarshal(ev.Value, msg)
            WriteStuff(mySQLClient, msg.Id, msg.Lastname, msg.Firstname, msg.Address, msg.City)
            if ev.Headers != nil {
                fmt.Printf("%% Headers: %v\n", ev.Headers)
            }
            _, err := c.StoreMessage(ev)
            if err != nil {
                fmt.Fprintf(os.Stderr, "%% Error storing offset after message %s:\n",
                    ev.TopicPartition)
            }
        }
    }

    fmt.Printf("Closing consumer\n")
    c.Close()
}

这对于工作组大小 1 来说效果很好，但是每次尝试使其适用于更大的工作组大小都会失败 - 到目前为止我所学到的是我想要 context.withcancel(context.background())从 main 传递到工作函数中，但我不知道如何设置等待组或 goroutine 来实际完成这项工作

正确答案

我知道您的问题是如何使用上下文（而不是 sigchan ）管理工作人员的生命周期。最简单的方法是使用 signal.notifycontext - 这为您提供了一个上下文，当发送其中一个信号。所以主要会变成

func main() {
    ctx, stop := signal.notifycontext(context.background(), syscall.sigint, syscall.sigterm)
    defer stop()

    mysqlclient, _ := db.newmysqldbclient("root", "", "localhost", 3306, "testbase")
    workers := worker.creategroup("localhost:9092", "testgroup", 1)
    var wg sync.waitgroup
    for _, w := range workers {
        w_ := w
        wg.add(1)
        go func() {
            defer wg.done()
            worker.startworker(ctx, w_, []string{"test-topic"}, mysqlclient)
        }()
    }
    wg.wait()
}

还要注意使用 waitgroup 以避免 main 在所有工作人员完成之前退出。 startworker 会像

func StartWorker(ctx context.Context, c *kafka.Consumer, topics []string, mySQLClient *sql.DB) {
    _ = c.SubscribeTopics(topics, nil)
    fmt.Println(c)
    for {
        select {
        case <-ctx.Done:
            return
        default:
        ...

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Kafka消费者的Goroutine》文章吧，也可关注golang学习网公众号了解相关技术文章。