终止Go例程中的无限循环

一分耕耘，一分收获！既然打开了这篇文章《终止Go例程中的无限循环》，就坚持看下去吧！文中内容包含等等知识点...希望你能在阅读本文后，能真真实实学到知识或者帮你解决心中的疑惑，也欢迎大佬或者新人朋友们多留言评论，多给建议！谢谢！

我有一个 go 例程，它基本上充当 kafkaconsumer，它从主题读取消息，然后为收到的每条消息生成另一个 go 例程 。现在，当 main goroutine 应用程序关闭时，这个 consumer goroutine 应该关闭。但我在正确关闭它方面遇到了困难。 下面是kafka consumer定义

package svc    

import (
    "event-service/pkg/pb"
    "fmt"
    "github.com/gogo/protobuf/proto"
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
    "log"
    "os"
    "sync"
)    

type eventconsumer func(event eventservice.event)    

type kafkaconsumer struct {
    done            chan bool
    eventchannels   []string
    consumer        *kafka.consumer
    consumermapping map[string]eventconsumer
    wg              *sync.waitgroup
}    

func getkafkaconsumerconfigmap(config map[string]interface{}) *kafka.configmap {
    configmap := &kafka.configmap{}
    for key, value := range config {
        err := configmap.setkey(key, value)
        if err != nil {
            log.println(fmt.sprintf("an error %v occurred while setting %v: %v", err, key, value))
        }
    }
    return configmap
}    

func newkafkaconsumer(channels []string, config map[string]interface{}, consumermapping map[string]eventconsumer) *kafkaconsumer {
    var wg sync.waitgroup
    consumer, err := kafka.newconsumer(getkafkaconsumerconfigmap(config))
    done := make(chan bool, 1)
    if err != nil {
        log.fatalf("an error %v occurred while starting kafka consumer.", err)
    }
    err = consumer.subscribetopics(channels, nil)
    if err != nil {
        log.fatalf("an error %v occurred while subscribing to kafka topics %v.", err, channels)
    }
    return &kafkaconsumer{eventchannels: channels, done: done, wg: &wg, consumer: consumer, consumermapping: consumermapping}
}    

func (kc *kafkaconsumer) getevent(eventdata []byte) *eventservice.event {
    event := eventservice.event{}
    err := proto.unmarshal(eventdata, &event)
    if err != nil {
        log.println(fmt.sprintf("an error %v occurred while un marshalling data from kafka.", err))
    }
    return &event
}    

func (kc *kafkaconsumer) consume() {
    go func() {
        run := true
        for run == true {
            select {
            case sig := <-kc.done:
                log.println(fmt.sprintf("caught signal %v: terminating \n", sig))
                run = false
                return
            default:
            }
            e := <-kc.consumer.events()
            switch event := e.(type) {
            case kafka.assignedpartitions:
                _, _ = fmt.fprintf(os.stderr, "%% %v\n", event)
                err := kc.consumer.assign(event.partitions)
                if err != nil {
                    log.println(fmt.sprintf("an error %v occurred while assigning partitions.", err))
                }
            case kafka.revokedpartitions:
                _, _ = fmt.fprintf(os.stderr, "%% %v\n", event)
                err := kc.consumer.unassign()
                if err != nil {
                    log.println(fmt.sprintf("an error %v occurred while unassigning partitions.", err))
                }
            case *kafka.message:
                domainevent := kc.getevent(event.value)
                kc.wg.add(1)
                go func(event *eventservice.event) {
                    defer kc.wg.done()
                    if eventconsumer := kc.consumermapping[domainevent.entitytype]; eventconsumer != nil {
                        eventconsumer(*domainevent)
                    } else {
                        log.println(fmt.sprintf("event consumer not found for %v event type", domainevent.entitytype))
                    }
                }(domainevent)
            case kafka.partitioneof:
                fmt.printf("%% reached %v\n", e)
            case kafka.error:
                _, _ = fmt.fprintf(os.stderr, "%% error: %v\n", e)
            }
        }
    }()
}    

func (kc *kafkaconsumer) close() {
    log.println("waiting")
    kc.wg.wait()
    kc.done <- true
    log.println("done waiting")
    err := kc.consumer.close()
    if err != nil {
        log.println(fmt.sprintf("an error %v occurred while closing kafka consumer.", err))
    }
}

下面是主线程代​​码

package main    

import (
    "event-service/pkg/pb"
    "event-service/pkg/svc"
    "fmt"
    "log"
)    

func main() {
    eventConsumerMapping := map[string]svc.EventConsumer{"doctor-created": func(event eventService.Event) {
        log.Println(fmt.Sprintf("Got event %v from kafka", event))
    }}
    consumerConfig := map[string]interface{}{
        "bootstrap.servers":               "localhost:9092",
        "group.id":                        "catalog",
        "go.events.channel.enable":        true,
        "go.application.rebalance.enable": true,
        "enable.partition.eof":            true,
        "auto.offset.reset":               "earliest",
    }
    kafkaConsumer := svc.NewKafkaConsumer([]string{"doctor-created"}, consumerConfig, eventConsumerMapping)
    kafkaConsumer.Consume()
    kafkaConsumer.Close()
}

这里的问题是应用程序有时根本没有结束，并且在某些运行中不执行 consume 函数，我在这里缺少什么？

解决方案

好的，解决方案如下， 1. 由于只要主 go 例程还活着，消费者 go 例程就应该存在，并且主 go 例程也是一个无尽的 go 例程，因此在 go 例程运行时关闭消费者 go 例程并不是正确的方法。

因此以下解决方案有效

package main    

import (
    "event-service/pkg/pb"
    "event-service/pkg/svc"
    "fmt"
    "log"
    "sync"
)    

func main() {
    eventconsumermapping := map[string]svc.eventconsumer{"doctor-created": func(event eventservice.event) {
        log.println(fmt.sprintf("got event %v from kafka", event))
    }}
    consumerconfig := map[string]interface{}{
        "bootstrap.servers":               "localhost:9092",
        "group.id":                        "catalog-2",
        "session.timeout.ms":              6000,
        "go.events.channel.enable":        true,
        "go.application.rebalance.enable": true,
        "enable.partition.eof":            true,
        "auto.offset.reset":               "earliest",
    }
    var wg sync.waitgroup
    kafkaconsumer := svc.newkafkaconsumer([]string{"doctor-created"}, consumerconfig, eventconsumermapping)
    kafkaconsumer.consume(&wg)
    wg.wait()
    kafkaconsumer.close()
}

服务定义

package svc    

import (
    "event-service/pkg/pb"
    "fmt"
    "github.com/gogo/protobuf/proto"
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
)    

type EventConsumer func(event eventService.Event)    

type KafkaConsumer struct {
    done            chan bool
    consumer        *kafka.Consumer
    consumerMapping map[string]EventConsumer
    sigChan         chan os.Signal
    channels        []string
}    

func getKafkaConsumerConfigMap(config map[string]interface{}) *kafka.ConfigMap {
    configMap := &kafka.ConfigMap{}
    for key, value := range config {
        err := configMap.SetKey(key, value)
        if err != nil {
            log.Println(fmt.Sprintf("An error %v occurred while setting %v: %v", err, key, value))
        }
    }
    return configMap
}    

func NewKafkaConsumer(channels []string, config map[string]interface{}, consumerMapping map[string]EventConsumer) *KafkaConsumer {
    sigChan := make(chan os.Signal, 1)
    consumer, err := kafka.NewConsumer(getKafkaConsumerConfigMap(config))
    done := make(chan bool, 1)
    if err != nil {
        log.Fatalf("An error %v occurred while starting kafka consumer.", err)
    }
    err = consumer.SubscribeTopics(channels, nil)
    if err != nil {
        log.Fatalf("An error %v occurred while subscribing to kafka topics %v.", err, channels)
    }
    return &KafkaConsumer{channels: channels, sigChan: sigChan, done: done, consumer: consumer, consumerMapping: consumerMapping}
}    

func (kc *KafkaConsumer) getEvent(eventData []byte) *eventService.Event {
    event := eventService.Event{}
    err := proto.Unmarshal(eventData, &event)
    if err != nil {
        log.Println(fmt.Sprintf("An error %v occurred while un marshalling data from kafka.", err))
    }
    return &event
}    

func (kc *KafkaConsumer) Consume(wg *sync.WaitGroup) {
    signal.Notify(kc.sigChan, syscall.SIGINT, syscall.SIGTERM)
    wg.Add(1)
    go func() {
        run := true
        defer wg.Done()
        for run == true {
            select {
            case sig := <-kc.sigChan:
                fmt.Printf("Caught signal %v: terminating\n", sig)
                run = false
            case ev := <-kc.consumer.Events():
                switch e := ev.(type) {
                case kafka.AssignedPartitions:
                    _, _ = fmt.Fprintf(os.Stderr, "%% %v\n", e)
                    _ = kc.consumer.Assign(e.Partitions)
                case kafka.RevokedPartitions:
                    _, _ = fmt.Fprintf(os.Stderr, "%% %v\n", e)
                    _ = kc.consumer.Unassign()
                case *kafka.Message:
                    domainEvent := kc.getEvent(e.Value)
                    wg.Add(1)
                    go func(event *eventService.Event) {
                        defer wg.Done()
                        if eventConsumer := kc.consumerMapping[domainEvent.EntityType]; eventConsumer != nil {
                            eventConsumer(*domainEvent)
                        } else {
                            log.Println(fmt.Sprintf("Event consumer not found for %v event type", domainEvent.EntityType))
                        }
                    }(domainEvent)
                case kafka.PartitionEOF:
                    fmt.Printf("%% Reached %v\n", e)
                case kafka.Error:
                    // Errors should generally be considered as informational, the client will try to automatically recover
                    _, _ = fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
                }
            }
        }
    }()
}    

func (kc *KafkaConsumer) Close() {
    err := kc.consumer.Close()
    if err != nil {
        log.Println(fmt.Sprintf("An error %v occurred while closing kafka consumer.", err))
    }
}

以上就是《终止Go例程中的无限循环》的详细内容，更多关于的资料请关注golang学习网公众号！