- zookeeper 镜像选择官方镜像
docker pull zookeeper
docker run --name zoo -p 2181:2181 -d zookeeper
- 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代表接受所有地址。设置了上个变量就要设置此变量。
- 使用 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 。
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)
}
- 启动生产者
- 在宿主机执行以下命令（其中 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
