Kafka（一）学习笔记

Kafka , , 评论

Kafka（一）学习笔记

1.1 Go语言操作Kafka

Kafka是一种高吞吐量的分布式发布订阅消息MQ系统，本文介绍了如何使用kafka-go这个库实现Go语言操作kafka
segmentio/kafka-go 是纯Go实现，提供了与kafka交互的低级别和高级别两套API，同时也支持Context并发控制

1.2 Kafka开发环境搭建

这里使用docker-compose快速搭建一套单节点zookeeper和单节点kafka本地发环境

deployment/docker-compose.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
version: '3'

services:
  zookeeper:
    container_name: zookeeper
    hostname: zookeeper
    image: bitnami/zookeeper:latest
    ports:
      - 2181:2181
    environment:
      ALLOW_ANONYMOUS_LOGIN: yes
  kafka:
    container_name: kafka
    image: bitnami/kafka:3.6.1
    # 失败重启容器3次
    restart: on-failure:3
    links:
      - zookeeper
    ports:
      - 9092:9092
      - 9093:9093
    environment:
      KAFKA_CFG_BROKER_ID: 1
      KAFKA_CFG_DELETE_TOPIC_ENABLE: 'true'
      KAFKA_CFG_ADVERTISED_HOST_NAME: 'localhost'
      KAFKA_CFG_ADVERTISED_PORT: '9092'
      KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_CFG_MESSAGE_MAX_BYTES: '200000000'
      KAFKA_CFG_LISTENERS: 'PLAINTEXT://:9092,SASL_PLAINTEXT://:9093'
      KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093'
      KAFKA_CFG_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
      KAFKA_CFG_AUTHORIZER_CLASS_NAME: 'kafka.security.authorizer.AclAuthorizer'
      KAFKA_CFG_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
      KAFKA_OPTS: "-Djava.security.auth.login.config=/opt/bitnami/kafka/config/kafka_jaas.conf"
      ALLOW_PLAINTEXT_LISTENER: yes
    entrypoint:
      - "/bin/bash"
      - "-c"
      - echo -e 'KafkaServer {\norg.apache.kafka.common.security.scram.ScramLoginModule required\n username="adminscram"\n password="admin-secret";\n org.apache.kafka.common.security.plain.PlainLoginModule required\n username="adminplain"\n password="admin-secret"\n user_adminplain="admin-secret";\n  };' > /opt/bitnami/kafka/config/kafka_jaas.conf; /opt/bitnami/kafka/bin/kafka-configs.sh --zookeeper zookeeper:2181 --alter --add-config "SCRAM-SHA-256=[password=admin-secret-256],SCRAM-SHA-512=[password=admin-secret-512]" --entity-type users --entity-name adminscram; exec /entrypoint.sh /run.sh
1.3 启动Kafka开发环境容器
1
docker-compose up -d
1
2
3
4
docker ps

c2d4264c208b   bitnami/kafka:3.6.1        "/bin/bash -c 'echo …"   About an hour ago   Up 2 minutes   0.0.0.0:9092-9093->9092-9093/tcp                       kafka
f29188f9b4ac   bitnami/zookeeper:latest   "/opt/bitnami/script…"   About an hour ago   Up 2 minutes   2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp, 8080/tcp   zookeeper
1.4 安装kafka-go
1
go get -u -v github.com/segmentio/kafka-go
2.1 kafka生产端
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
// 生产者
func writeKafka(ctx context.Context) {
    writer := &kafka.Writer{
        Addr:                   kafka.TCP("localhost:9092"),
        Topic:                  topic,
        Balancer:               &kafka.Hash{},
        WriteTimeout:           time.Second * 1,
        RequiredAcks:           kafka.RequireNone,
        AllowAutoTopicCreation: true, // 自动创建topic
    }
    defer writer.Close()
    
    // 5次写入重试机会
    for i := 1; i <= 5; i++ {
        if err := writer.WriteMessages(ctx,
            kafka.Message{Key: []byte("Key-A"), Value: []byte("this")},
            kafka.Message{Key: []byte("Key-B"), Value: []byte("is")},
            kafka.Message{Key: []byte("Key-C"), Value: []byte("a")},
            kafka.Message{Key: []byte("Key-D"), Value: []byte("test")},
        ); err != nil {
            if err == kafka.LeaderNotAvailable {
                time.Sleep(time.Millisecond * 500)
                continue
            } else {
                fmt.Printf("failed to write message:%v\n", err)
            }
        } else {
            fmt.Printf("kafka write message success:%d\n", i)
            break // 写入成功退出循环
        }
    }
}
2.2 kafka消费端
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 消费者
func readKafka(ctx context.Context) {
    reader = kafka.NewReader(kafka.ReaderConfig{
        Brokers:        []string{"localhost:9092"},
        Topic:          topic,
        CommitInterval: time.Second * 1,
        GroupID:        "consumer-group-id",
        StartOffset:    kafka.FirstOffset,
    })
    defer reader.Close()
    
    for {
        if message, err := reader.ReadMessage(ctx); err != nil {
            fmt.Printf("failed to read message:%v\n", err)
            break
        } else {
            fmt.Printf("message at topic:%s,partition:%d,offset:%d,key:%s,value:%s\n",
                message.Topic, message.Partition, message.Offset, string(message.Key), string(message.Value))
        }
    }
}
2.3 退出信号监听，用于消费端正确退出
1
2
3
4
5
6
7
8
9
10
func listenSignal() {
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    sig := <-quit // 收到退出信号
    fmt.Printf("recv:%s\n", sig.String())
    if reader != nil {
        reader.Close()
    }
    os.Exit(0)
}
2.4 main函数调用
1
2
3
4
5
6
7
func main() {
    ctx := context.Background()
    writeKafka(ctx)
    
    go listenSignal()
    readKafka(ctx)
}
1
2
3
4
5
6
kafka write message success:1
message at topic:test-topic,partition:0,offset:32,key:Key-A,value:this
message at topic:test-topic,partition:0,offset:33,key:Key-B,value:is
message at topic:test-topic,partition:0,offset:34,key:Key-C,value:a
message at topic:test-topic,partition:0,offset:35,key:Key-D,value:test
^Crecv:interrupt
2.5 最后附上全部代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
// @package    : main
// @file       : kafka.go
// @author     : hqd8080
// @contact    : hanquanding@163.com
// @time       : 2023/6/19
// @description: golang使用kafka

package main

import (
    "context"
    "fmt"
    "github.com/segmentio/kafka-go"
    "os"
    "os/signal"
    "syscall"
    "time"
)

var topic = "test-topic"
var reader *kafka.Reader

// 生产者
func writeKafka(ctx context.Context) {
    writer := &kafka.Writer{
        Addr:                   kafka.TCP("localhost:9092"),
        Topic:                  topic,
        Balancer:               &kafka.Hash{},
        WriteTimeout:           time.Second * 1,
        RequiredAcks:           kafka.RequireNone,
        AllowAutoTopicCreation: true, // 自动创建topic
    }
    defer writer.Close()
    
    // 5次写入重试机会
    for i := 1; i <= 5; i++ {
        if err := writer.WriteMessages(ctx,
            kafka.Message{Key: []byte("Key-A"), Value: []byte("this")},
            kafka.Message{Key: []byte("Key-B"), Value: []byte("is")},
            kafka.Message{Key: []byte("Key-C"), Value: []byte("a")},
            kafka.Message{Key: []byte("Key-D"), Value: []byte("test")},
        ); err != nil {
            if err == kafka.LeaderNotAvailable {
                time.Sleep(time.Millisecond * 500)
                continue
            } else {
                fmt.Printf("failed to write message:%v\n", err)
            }
        } else {
            fmt.Printf("kafka write message success:%d\n", i)
            break // 写入成功退出循环
        }
    }
}

// 消费者
func readKafka(ctx context.Context) {
    reader = kafka.NewReader(kafka.ReaderConfig{
        Brokers:        []string{"localhost:9092"},
        Topic:          topic,
        CommitInterval: time.Second * 1,
        GroupID:        "consumer-group-id",
        StartOffset:    kafka.FirstOffset,
    })
    defer reader.Close()
    
    for {
        if message, err := reader.ReadMessage(ctx); err != nil {
            fmt.Printf("failed to read message:%v\n", err)
            break
        } else {
            fmt.Printf("message at topic:%s,partition:%d,offset:%d,key:%s,value:%s\n",
                message.Topic, message.Partition, message.Offset, string(message.Key), string(message.Value))
        }
    }
}

func listenSignal() {
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    sig := <-quit // 收到退出信号
    fmt.Printf("recv:%s\n", sig.String())
    if reader != nil {
        reader.Close()
    }
    os.Exit(0)
}

func main() {
    ctx := context.Background()
    writeKafka(ctx)
	
    go listenSignal()
    readKafka(ctx)
}

🌴🌴🧩🧩🥥Gopher

心外无物，知行合一