0
0

golang消息队列kafka

传统的消费模型
消息队列的消息被消费了，数据则从队列里删除，并且下游的多个 consumer 都要抢这条消息。
发布/订阅模型允许消息被多个 consumer 消费，但是订阅者必须订阅所有分区
kafka为规避上面的缺点，引入了 消费组 模型。

kafka消费组模型
比如订阅了两个主题，每个主题的分区是2，则消费组启动 4 个消费实例，每个消费实例对应一个分区，这样避免了 消费实例的竞争。同时也避免了发布/订阅模型 消费实例必须订阅所有分区的问题。

1、kafka的生产者

sarama提供异步生产者(AsyncProducer)和同步生产者(SyncProducer)

  • 异步生产者用管道接受信息，并且在背后异步地生产消息。异步生产者会尽可能的提高效率，是sarama推荐的选择
  • 同步生产者提供一个阻塞的方法，直到消息被确认成功生产。显然效率上肯定不如异步生产者，且在某些配置中，确认的消息仍会丢失，SyncProducer 是在 AsyncProducer 基础上加sync.WaitGroup条件限制实现的。

相关介绍：

sarama Kafka客户端生产者与消费者梳理 - 平凡键客 - 博客园

1.1、同步方式sarama.NewSyncProducer

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
)

func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认, 等待服务器所有副本都保存成功后的响应
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 随机向partition发送消息
    // 是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
    config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
    config.Producer.Return.Errors = true    // 成功交付的消息将在errors channel返回
    // sasl认证
    config.Net.SASL.Enable = false
    //config.Net.SASL.User = "admin"
    //config.Net.SASL.Password = "admin"
    
    // 连接kafka
    producer, err := sarama.NewSyncProducer([]string{"192.168.103.6:9092"}, config)
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }

    // 关闭生产者
    defer producer.Close()

    // 构造⼀个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = "ffcs"
    for i := 0; i < 20; i++ {
        msg.Value = sarama.StringEncoder(fmt.Sprintf("helloV%d kafka", i))
        // 发送消息
        pid, offset, err := producer.SendMessage(msg)
        if err != nil {
            fmt.Println("send message failed,", err)
            return
        }
        fmt.Printf("pid:%v offset:%v\n", pid, offset)
    }
}

1.2、异步方式sarama.NewASyncProducer

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
    "time"
)

func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认, 等待服务器所有副本都保存成功后的响应
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 随机向partition发送消息
    // 是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
    //config.Producer.RequiredAcks = sarama.NoResponse
    config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
    config.Producer.Return.Errors = true    // 成功交付的消息将在errors channel返回
    // sasl认证
    config.Net.SASL.Enable = false
    //config.Net.SASL.User = "admin"
    //config.Net.SASL.Password = "admin"

    // 连接kafka
    producer, err := sarama.NewAsyncProducer([]string{"192.168.103.6:9092"}, config)
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }

    // 关闭生产者
    defer producer.Close()

    //循环判断哪个通道发送过来数据.
    fmt.Println("start goroutine")
    go func(p sarama.AsyncProducer) {
        for {
            select {
            case <-p.Successes():
                fmt.Println("success")
            case fail := <-p.Errors():
                fmt.Println("err: ", fail.Err)
            }
        }
    }(producer)

    // 构造⼀个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = "ffcs"
    
    for i := 0; i < 20; i++ {
        msg.Value = sarama.StringEncoder(fmt.Sprintf("helloo00000000oV%d kafka", i))
        // 发送消息
        producer.Input() <- msg
    }
    time.Sleep(10 * time.Second)
}

2、kafka使用消费组方式处理数据

package main

import (
    "context"
    "fmt"
    "github.com/Shopify/sarama"
    "log"
    "sync"
)

type Consumer struct {
    Hosts        []string
    Topics       []string
    GroupID      string
    ConsumerName string
}

func main() {
    // 消费者consumer1
    consumer := &Consumer{
        Hosts:        []string{"192.168.103.6:9092"},
        Topics:       []string{"ffcs"},
        GroupID:      "topicname-group",
        ConsumerName: "consumer1",
    }

    wg := &sync.WaitGroup{}
    wg.Add(1)
    go startConsumer(consumer)
    wg.Wait()
}

// 实现  github.com/Shopify/sarama/consumer_group.go/ConsumerGroupHandler  这个接口
// Setup 执行在 获得新 session 后 的第一步, 在 ConsumeClaim() 之前
func (consumer *Consumer) Setup(_ sarama.ConsumerGroupSession) error {
    return nil
}

// Cleanup 执行在 session 结束前, 当所有 ConsumeClaim goroutines 都退出时
func (consumer *Consumer) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

// ConsumeClaim kafka具体的消费逻辑
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for message := range claim.Messages() {
        log.Printf("ConsumerName:%s Message claimed: value = %s, topic = %s,partition=%d\n", consumer.ConsumerName, string(message.Value), message.Topic, message.Partition)
        // 将消息标记为已使用
        session.MarkMessage(message, "")
        //session.Commit()
    }
    return nil
}

func startConsumer(consumer *Consumer) {
    config := sarama.NewConfig()
    config.Version = sarama.V0_11_0_2
    // 是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
    //config.Producer.RequiredAcks = sarama.NoResponse
    config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
    config.Producer.Return.Errors = true    // 成功交付的消息将在errors channel返回
    //config.Consumer.Offsets.AutoCommit.Enable = false // 禁用主动提交，改为手动, 需要执行MarkMessage(msg,"")与Commit()
    config.Consumer.Offsets.AutoCommit.Enable = true // 开启主动提交，须要手动调用MarkMessage(msg,"")
    config.Net.SASL.Enable = false
    //config.Net.SASL.User = "admin"
    //config.Net.SASL.Password = "admin"

    consumerGroup, err := sarama.NewConsumerGroup(consumer.Hosts, consumer.GroupID, config)
    if err != nil {
        fmt.Printf("NewConsumerGroup create consumer error %s\n", err.Error())
        return
    }

    //go func() {
    //    for err := range consumerGroup.Errors() {
    //        fmt.Println("consumerGroup errors : ", err)
    //    }
    //}()

    for {
        // 启动kafka消费组模式，消费的逻辑在上面的 ConsumeClaim 这个方法里
        if err := consumerGroup.Consume(context.Background(), consumer.Topics, consumer); err != nil {
            fmt.Printf("Error from consumer: %v\n", err)
        }
    }
}

3、kafka消费者处理数据

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
    "sync"
)

func main() {
    config := sarama.NewConfig()
    config.Version = sarama.V0_11_0_2
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.AutoCommit.Enable = true
    config.Net.SASL.Enable = false
    //config.Net.SASL.User = "admin"
    //config.Net.SASL.Password = "admin"

    consumer, err := sarama.NewConsumer([]string{"192.168.103.6:9092"}, config)
    if err != nil {
        fmt.Printf("consumer_test create consumer error %s\n", err.Error())
        return
    }

    partitionList, err := consumer.Partitions("ffcs") // 根据topic取到所有的分区
    if err != nil {
        fmt.Printf("fail to get list of partition:err%v\n", err)
        return
    }
    defer consumer.Close()
    // 没有配置groupID的消费者, 当服务重启后会消费历史的数据
    kafkaConsumer := sync.WaitGroup{}
    for partition := range partitionList {
        kafkaConsumer.Add(1)
        go kafkaConsumerMsg(consumer, partition, kafkaConsumer)
    }
    kafkaConsumer.Wait()
}

func kafkaConsumerMsg(consumer sarama.Consumer, partition int, kafkaConsumer sync.WaitGroup) {
    defer kafkaConsumer.Done()
    // sarama.OffsetOldest消费历史数据、sarama.OffsetNewest重新接收新的消费数据
    partitionConsumer, err := consumer.ConsumePartition("ffcs", int32(partition), sarama.OffsetOldest) // 针对每个分区创建一个分区消费者
    if err != nil {
        fmt.Printf("try create partition_consumer error %s\n", err.Error())
        return
    }
    defer partitionConsumer.Close()
    for {
        select {
        case msg := <-partitionConsumer.Messages(): // 阻塞直到有值发送过来，然后再继续等待
            fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
                msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
            //time.Sleep(1 * time.Second)
        case err := <-partitionConsumer.Errors():
            fmt.Printf("err :%s\n", err.Error())
        }
    }
}

4、kafka相关命令

4.1 Topic 相关
创建名为 test 的 Topic。

bin/kafka-topics.sh -zookeeper localhost:2181 --create --partitions 5 --replication-factor 1 --topic test

–partitions：分区数

–replication-factor：副本数

查询 Topic 列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

删除名为 test 的 Topic

bin/kafka-topics.sh --delete --zookeeper localhost:2181  --topic test

配置 delete.topic.enable 为 true，这样才能删除 topic

查询 Topic 的信息

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

如果未指定 topic 则输出所有 topic 的信息

4.2 消息相关
生产者发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

消费者查询消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --group t1

–from-beginning：表示从头开始接收数据

–group：指定消费者组

查询名为 test 的 Topic 的消息。

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -1

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -2

–time-1 表示要获取指定 topic 所有分区当前的最大位移（历史总消息数），–time-2 表示获取当前最早位移（被消费的消息数），两个命令的输出结果相减便可得到所有分区当前的消息总数。

输出示例：test:0:3

第一个数字0表示分区，第二个数字3表示偏移量。

标签： kafka golang
本文转载自: https://blog.csdn.net/zorsea/article/details/128275007
版权归原作者 ahisky 所有， 如有侵权，请联系我们删除。
登录后发布评论

“golang消息队列kafka”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

通过SSH操作Linux虚拟机

C#学习笔记（三）Visual Studio安装与使用

JPA基础教程：实体操作和数据库交互实践

开源低代码-Microi吾码报表引擎

HiveSQL：提取json串内容（含key为中文情况）——get_json_oject和json_tuple

Spring Cloud Stream 3.x+kafka 3.8整合

【大数据·hadoop】项目实践：IDEA实现WordCount词频统计项目

文章导航