why is kafka

Apache Kafka 是一种实时数据流技术，每天能够处理数万亿个事件。Kafka 最初被设想为消息队列，基于分布式提交日志的抽象。自 2011 年创建并开源以来，Kafka 已成为处理动态数据的行业标准

sarama

这里我们使用在go当中常用的第三方插件
Go语言中连接kafka使用第三方库 https://github.com/Shopify/sarama

1、下载及安装`

go get github.com/Shopify/sarama

注意

sarama v1.20之后的版本加入了zstd压缩算法，需要用到cgo，在Windows平台编译时会提示类似如下错误：

# github.com/DataDog/zstd
exec: "gcc":executable file not found in %PATH%

所以在Windows平台请使用v1.19版本的sarama。

2、连接kafka发送消息

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
)

// 基于sarama第三方库开发的kafka client

func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
    config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

    // 构造一个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = "web_log"
    msg.Value = sarama.StringEncoder("this is a test log")
    // 连接kafka
    client, err := sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    defer client.Close()
    // 发送消息
    pid, offset, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send msg failed, err:", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

3、连接kafka消费消息

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
)

// kafka consumer
func main() {
    consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
    if err != nil {
        fmt.Printf("fail to start consumer, err:%v\n", err)
        return
    }
    partitionList, err := consumer.Partitions("web_log") // 根据topic取到所有的分区
    if err != nil {
        fmt.Printf("fail to get list of partition:err%v\n", err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList { // 遍历所有的分区
        // 针对每个分区创建一个对应的分区消费者
        pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
            return
        }
        defer pc.AsyncClose()
        // 异步从每个分区消费信息
        go func(sarama.PartitionConsumer) {
                      wg.Add(1)//等待数量+1
            for msg := range pc.Messages() {
                fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
            }
                        wg.Done()//等待数量-1
        }(pc)
    }
    time.Sleep(time.Hour)
}

4、至此一个简单的kafka生产者和消费者就完成了，具体的业务可再次基础之上封装抽象

随着go语音开发大军的增长sarama也出现了一些不一样声音

为什么不推荐使用Sarama Go客户端收发消息？

问题现象

所有Sarama Go版本客户端存在以下已知问题：

• 当Topic新增分区时，Sarama Go客户端无法感知并消费新增分区，需要客户端重启后，才能消费到新增分区。
• 当Sarama Go客户端同时订阅两个以上的Topic时，有可能会导致部分分区无法正常消费消息。
• 当Sarama Go客户端的消费位点重置策略设置为Oldest(earliest)时，如果客户端宕机或服务端版本升级，由于Sarama Go客户端自行实现OutOfRange机制，有可能会导致客户端从最小位点开始重新消费所有消息。

解决方案

Confluent Go请访问https://github.com/confluentinc/confluent-kafka-go

注意 如果无法在短期内替换客户端，请注意以下事项：

• 针对生产环境，请将位点重置策略设置为Newest(latest)；针对测试环境，或者其他明确可以接收大量重复消息的场景，设置为Oldest(earliest)。
• 如果发生了位点重置，产生大量堆积，您可以使用消息队列Kafka版控制台提供的重置消费位点功能，手动重置消费位点到某一时间点，无需改代码或换Consumer Group。具体操作，请参见重置消费位点https://help.aliyun.com/document_detail/68329.htm#task-68329-zh重置消费位点是指改变订阅者当前的消费位置。您可通过重置消费位点，按需清除堆积的或不想消费的这部分消息，或直接跳转到某个时间点，从指定时间点的位点开始消费消息。")。

小结

Kafka的消息组织方式实际上是三级结构:主题-分区-消息。主题下的每条消息只会保存在某一一个分区中，而不会在多个分区中被保存多份。
分区是实现负载均衡以及高吞吐量的关键。
所谓分区策略，也就是决定生产者将消息发送到哪个分区的算法。Kafka为我们提供了默认分区策略，同时，它也支持你自定义分区策略。
比较常见的分区策略包括轮询策略、随机策略和按消息键保序策略。还有一种是基于地理位置的分区策略，但这种策略一-般只针对那些 大规模的Kafka集群，特别是跨城市、跨国家甚至是跨大洲的集群。