2017-02-01 18:07
浏览 739


An available library is sarama (or its expansion sarama-cluster) however no consumer group example are provided, not in sarama nor in sarama-cluster.

I do not understand the API. May I have an example of creating a consumer group for a topic?

  • 点赞
  • 写回答
  • 关注问题
  • 收藏
  • 邀请回答

2条回答 默认 最新

  • doushoubu5360
    doushoubu5360 2017-02-01 22:12

    The consumer group is specified by the second argument of the cluster consumer "constructor". Here's a very basic sketch:

    import (
    conf := cluster.NewConfig()
    // add config values
    brokers := []string{"kafka-1:9092", "kafka-2:9092"}
    group := "Your-Consumer-Group"
    topics := []{"topicName"}
    consumer := cluster.NewConsumer(broker, group, topics, conf)

    And so you'll have a consumer belonging to the specified consumer group.

    点赞 评论
  • dphe5602
    dphe5602 2019-06-20 05:49

    There is no need to use sarama-cluster library. It is DEPRECATED for apache kafka integration. Sarama original library itself provide a way to connect to kafka cluster using consumer group.

    We need to create client and then we initialize consumer group where we create claims and wait for message channel to receive message.

    Initializing client :-

    kfversion, err := sarama.ParseKafkaVersion(kafkaVersion) // kafkaVersion is the version of kafka server like
    if err != nil {
    config := sarama.NewConfig()
    config.Version = kfversion
    config.Consumer.Return.Errors = true
    // Start with a client
    client, err := sarama.NewClient([]string{brokerAddr}, config)
    if err != nil {
    defer func() { _ = client.Close() }()

    Connection to consumer group :-

    // Start a new consumer group
    group, err := sarama.NewConsumerGroupFromClient(consumer_group, client)
    if err != nil {
    defer func() { _ = group.Close() }()

    Start consuming messages from topic partition :-

    // Iterate over consumer sessions.
    ctx := context.Background()
    for {
        topics := []string{topicName}
        handler := &Message{}
        err := group.Consume(ctx, topics, handler)
        if err != nil {

    The last part is to wait for message channel to consume messages. We need to implement all of the functions (three) to implement ConsumerGroupHandler interface.

    func (exampleConsumerGroupHandler) Setup(_ ConsumerGroupSession) error   { return nil }
    func (exampleConsumerGroupHandler) Cleanup(_ ConsumerGroupSession) error { return nil }
    func (h exampleConsumerGroupHandler) ConsumeClaim(sess ConsumerGroupSession, claim ConsumerGroupClaim) error {
        for msg := range claim.Messages() {
            fmt.Printf("Message topic:%q partition:%d offset:%d
    ", msg.Topic, msg.Partition, msg.Offset)
            sess.MarkMessage(msg, "")
        return nil

    For more information on kafka using golang check sarama library.

    点赞 评论