kafka基础知识

kafka.png kafka.png image.png image.png

为什么要用kafka:

1.缓冲和削峰
2.解耦和扩展性
3.异步通信
4.健壮性
5.冗余

什么是ISR和AR

ISR:In-Sync Replicas 副本同步队列
AR:Assigned Replicas 所有副本

下面的所有终端命令基于都是windows10环境下
1.启动zookeeper:在kafka目录下执行：
首先需要修改config目录下的zookeeper.properties的配置，指定数据存放的目录
dataDir=D:/mysoftware/ELK/zookeeperdata

启动命令
bin\windows\zookeeper-server-start.bat config\zookeeper.properties

2.启动kafka，同理先修改配置文件，config下的server.properties
log.dirs=D:/mysoftware/ELK/kafkalog // 日志目录
zookeeper.connect=localhost:2181 //zookeeper地址

启动命令
bin\windows\kafka-server-start.bat config\server.properties

zookeeper：kafka节点启动的时候注册节点，查询节点（轮询，hash，权重）

操作kafka：github.com/Shopify/sarama 客户端包

生产者

package main

import (
    "fmt"
    "log"
    "os"
    "time"

    "github.com/Shopify/sarama"
)

var Address = []string{"127.0.0.1:9092"}

func main() {
    fmt.Println("hello")
    syncProducer(Address)
}

//同步消息模式
func syncProducer(address []string) {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    config.Producer.Timeout = 5 * time.Second
    p, err := sarama.NewSyncProducer(address, config)
    if err != nil {
        log.Printf("sarama.NewSyncProducer err, message=%s \n", err)
        return
    }
    defer p.Close()
    topic := "test"
    srcValue := "sync: this is a message. index=%d"
    for i := 0; i < 100; i++ {
        value := fmt.Sprintf(srcValue, i)
        msg := &sarama.ProducerMessage{
            Topic: topic,
            Value: sarama.ByteEncoder(value),
        }
        part, offset, err := p.SendMessage(msg)
        if err != nil {
            log.Printf("send message(%s) err=%s \n", value, err)
        } else {
            fmt.Fprintf(os.Stdout, value+"发送成功，partition=%d, offset=%d \n", part, offset)
        }
        time.Sleep(2 * time.Second)
    }
}

func SaramaProducer() {

    config := sarama.NewConfig()
    //等待服务器所有副本都保存成功后的响应
    config.Producer.RequiredAcks = sarama.WaitForAll
    //随机向partition发送消息
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    //是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true
    //设置使用的kafka版本,如果低于V0_10_0_0版本,消息中的timestrap没有作用.需要消费和生产同时配置
    //注意，版本设置不对的话，kafka会返回很奇怪的错误，并且无法成功发送消息
    config.Version = sarama.V0_10_0_1

    fmt.Println("start make producer")
    //使用配置,新建一个异步生产者
    producer, e := sarama.NewAsyncProducer(Address, config)
    if e != nil {
        fmt.Println(e)
        return
    }
    defer producer.AsyncClose()

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

    var value string
    for i := 0; ; i++ {
        time.Sleep(500 * time.Millisecond)
        time11 := time.Now()
        value = "this is a message 0606 " + time11.Format("15:04:05")

        // 发送的消息,主题。
        // 注意：这里的msg必须得是新构建的变量，不然你会发现发送过去的消息内容都是一样的，因为批次发送消息的关系。
        msg := &sarama.ProducerMessage{
            Topic: "0606_test",
        }

        //将字符串转化为字节数组
        msg.Value = sarama.ByteEncoder(value)
        //fmt.Println(value)

        //使用通道发送
        producer.Input() <- msg
    }
}

消费者

package main

import (
    "fmt"
    // "log"
    // "os"
    // "os/signal"
    // "sync"

    //cluster "github.com/bsm/sarama-cluster"

    "github.com/Shopify/sarama"
)

func main() {
    consumer_test()
}


func consumer_test() {
    fmt.Printf("consumer_test")

    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
    config.Version = sarama.V0_11_0_2

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

    defer consumer.Close()

    partition_consumer, err := consumer.ConsumePartition("test", 0, sarama.OffsetNewest)
    if err != nil {
        fmt.Printf("try create partition_consumer error %s\n", err.Error())
        return
    }
    defer partition_consumer.Close()

    for {
        select {
        case msg := <-partition_consumer.Messages():
            fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
                msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
        case err := <-partition_consumer.Errors():
            fmt.Printf("err :%s\n", err.Error())
        }
    }
}

终端启动消费者：进入kafka按照目录
bin\windows\kafka-console-consumer.bat --bootstrap-server=127.0.0.1:9092 --topic=test --from-beginning

重复消费和消息丢失的问题：
1.offset没有及时更新，导致消息重复发送消费
方法一：引入第三方中间件， 将偏移量存入redis, 每次消费前判断redis的偏移量(双重保证，看业务接收程度 )

2.offset有及时更新，但是业务处理失败，导致消息消费不成功，造成消息丢失
方法一：把数据处理和提交offset放在一个事务里
方法二：把数据处理操作编程幂等的（消费一个和多次效果是一样的）