docker 部署kafka（测试环境）

1. zookeeper 镜像选择官方镜像

docker pull zookeeper
docker run --name zoo -p 2181:2181 -d zookeeper

2. kafka 镜像选择 bitnami/kafka

docker pull bitnami/kafka
docker run --name kafka -p 9092:9092  -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.3:2181 -e ALLOW_PLAINTEXT_LISTENER=yes -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -d  bitnami/kafka

docker容器部署必须指定以下环境变量：

• KAFKA_ZOOKEEPER_CONNECT 指定 zookeeper 的地址：端口。
• ALLOW_PLAINTEXT_LISTENER 允许使用PLAINTEXT侦听器。
• KAFKA_ADVERTISED_LISTENERS 是指向Kafka代理的可用地址列表。 Kafka将在初次连接时将它们发送给客户。格式为 PLAINTEXT://host:port ，此处已将容器9092端口映射到宿主机9092端口，所以host指定为localhost，便可在宿主机执行测试程序连接 kafka。
• KAFKA_LISTENERS 是 Kafka 代理将侦听传入连接的地址列表。格式为 PLAINTEXT://host:port ， 0.0.0.0代表接受所有地址。设置了上个变量就要设置此变量。

3. 使用 docker-compose 集群部署
   docker-compose.yml

version: '2'

services:
  zoo1:
    image: zookeeper
    container_name: zoo
    ports:
      - 2181:2181

  kafka1:
    image: 'bitnami/kafka:latest'
    ports:
      - '9092:9092'
    container_name: kafka1
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zoo1:2181
      - KAFKA_BROKER_ID=1
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.152.48:9092
    depends_on:
      - zoo1

  kafka2:
    image: 'bitnami/kafka:latest'
    ports:
      - '9093:9092'
    container_name: kafka2
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zoo1:2181
      - KAFKA_BROKER_ID=2
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.152.48:9093
    depends_on:
      - zoo1

  kafka3:
    image: 'bitnami/kafka:latest'
    ports:
      - '9094:9092'
    container_name: kafka3
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zoo1:2181
      - KAFKA_BROKER_ID=3
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.152.48:9094
    depends_on:
      - zoo1

将 KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.152.48:9094 中的 192.168.152.48 替换为docker宿主机的 ip 。

4. golang 连接 kafka
   此处使用 Shopify/sarama

go get github.com/Shopify/sarama

生产者

package main

import (
    "fmt"
    "time"

    "github.com/Shopify/sarama"
)

func main() {
    addrs := []string{"192.168.152.48:9092", "192.168.152.48:9093", "192.168.152.48:9094"}
    //addrs := []string{"192.168.152.48:9092"}
    config := sarama.NewConfig()
    config.Version = sarama.V2_1_0_0
    admin, err := sarama.NewClusterAdmin(addrs, config)
    if err != nil {
        fmt.Println(err)
    }
    err = admin.CreateTopic("tp33", &sarama.TopicDetail{NumPartitions: 1, ReplicationFactor: 3}, false)
    if err != nil {
        fmt.Println(err)
    }

    err = admin.Close()
    if err != nil {
        fmt.Println(err)
    }

    producer, err := sarama.NewSyncProducer(addrs, nil)
    if err != nil {
        fmt.Println(err)
    }
    defer func() {
        if err := producer.Close(); err != nil {
            fmt.Println(err)
        }
    }()

    msg := &sarama.ProducerMessage{Topic: "tp33", Value: sarama.StringEncoder("testing 123")}
    for {
        partition, offset, err := producer.SendMessage(msg)
        if err != nil {
            fmt.Println("failed to send message: ", err)
        } else {
            fmt.Printf("message sent to partition %d at offset %d\n", partition, offset)
        }
        time.Sleep(1500 * time.Millisecond)
    }
}

消费者

package main

import (
    "fmt"
    "os"
    "os/signal"

    "github.com/Shopify/sarama"
)

func main() {
    addrs := []string{"192.168.152.48:9092", "192.168.152.48:9093", "192.168.152.48:9094"}
    //addrs := []string{"192.168.152.48:9092"}
    consumer, err := sarama.NewConsumer(addrs, nil)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := consumer.Close(); err != nil {
            fmt.Println(err)
        }
    }()

    partitionConsumer, err := consumer.ConsumePartition("tp33", 0, sarama.OffsetNewest)
    if err != nil {
        panic(err)
    }
    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            fmt.Println(err)
        }
    }()

    signals := make(chan os.Signal)
    signal.Notify(signals, os.Interrupt)

    consumed := 0
ConsumerLoop:
    for {
        select {
        case msg := <-partitionConsumer.Messages():
            fmt.Println("Consumed message offset", msg.Offset)
            fmt.Println(string(msg.Value))
            consumed++
        case <-signals:
            break ConsumerLoop
        }
    }

    fmt.Println("Consumed:", consumed)
}

5. 启动生产者

• 在宿主机执行以下命令（其中 kafka1 是上边 yml 文件中的 container_name ）进入容器 kafka1

docker exec -it kafka1 /bin/bash
cd opt/bitnami/kafka/bin

• 执行以下命令创建 topic

./kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 3 --partitions 3 --topic tp33

• 执行以下命令查看 topic 列表

./kafka-topics.sh --bootstrap-server localhost:9092 --list

输出
tp33

• 执行以下命令查看指定 topic 相关信息

./kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic tp33

输出
Topic:tp33 PartitionCount:1 ReplicationFactor:3 Configs:
Topic: tp33 Partition: 0 Leader: 3 Replicas: 3,2,1 Isr: 3,2,1
第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。由于此主题只有一个分区，因此只有一行。
“leader”是负责给定分区的所有读写的节点。每个节点将成为随机选择的分区部分的领导者。
“replicas”是复制此分区日志的节点列表，无论它们是否为领导者，或者即使它们当前处于活动状态。
“isr”是“同步”复制品的集合。这是副本列表的子集，该列表当前处于活跃状态并且已经被领导者捕获。

• 执行以下命令像指定 topic 发送消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic tp33
>this test1
>test2

• 执行以下命令消费消息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic tp33
this test1
test2

• 现在让我们测试一下容错性。broker3 充当 leader 所以让我们杀了它：

docker container rm kafka3

此时 kafka3 节点已经关闭，但我们依旧可以发送/接受消息。
再次查看 topic 相关信息

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic tp33

输出
Topic:tp33 PartitionCount:1 ReplicationFactor:3 Configs:
Topic: tp33 Partition: 0 Leader: 2 Replicas: 3,2,1 Isr: 2,1