- 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容器部署必須指定以下環(huán)境變量：
- KAFKA_ZOOKEEPER_CONNECT 指定 zookeeper 的地址：端口。
- ALLOW_PLAINTEXT_LISTENER 允許使用PLAINTEXT偵聽器褥民。
- KAFKA_ADVERTISED_LISTENERS 是指向Kafka代理的可用地址列表爸黄。 Kafka將在初次連接時將它們發(fā)送給客戶葱轩。格式為 PLAINTEXT://host:port 屠阻，此處已將容器9092端口映射到宿主機9092端口，所以host指定為localhost策菜，便可在宿主機執(zhí)行測試程序連接 kafka晶疼。
- KAFKA_LISTENERS 是 Kafka 代理將偵聽傳入連接的地址列表。格式為 PLAINTEXT://host:port 又憨， 0.0.0.0代表接受所有地址翠霍。設(shè)置了上個變量就要設(shè)置此變量。
- 使用 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
生產(chǎn)者
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)
}
- 啟動生產(chǎn)者
- 在宿主機執(zhí)行以下命令（其中 kafka1 是上邊 yml 文件中的 container_name ）進入容器 kafka1
docker exec -it kafka1 /bin/bash
cd opt/bitnami/kafka/bin
- 執(zhí)行以下命令創(chuàng)建 topic
./kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 3 --partitions 3 --topic tp33
- 執(zhí)行以下命令查看 topic 列表
./kafka-topics.sh --bootstrap-server localhost:9092 --list
輸出
tp33
- 執(zhí)行以下命令查看指定 topic 相關(guān)信息
./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
第一行給出了所有分區(qū)的摘要寒匙，每個附加行提供有關(guān)一個分區(qū)的信息。由于此主題只有一個分區(qū)躏将，因此只有一行锄弱。
“l(fā)eader”是負責給定分區(qū)的所有讀寫的節(jié)點。每個節(jié)點將成為隨機選擇的分區(qū)部分的領(lǐng)導(dǎo)者祸憋。
“replicas”是復(fù)制此分區(qū)日志的節(jié)點列表会宪，無論它們是否為領(lǐng)導(dǎo)者，或者即使它們當前處于活動狀態(tài)蚯窥。
“isr”是“同步”復(fù)制品的集合掸鹅。這是副本列表的子集，該列表當前處于活躍狀態(tài)并且已經(jīng)被領(lǐng)導(dǎo)者捕獲拦赠。
- 執(zhí)行以下命令像指定 topic 發(fā)送消息
./kafka-console-producer.sh --broker-list localhost:9092 --topic tp33
>this test1
>test2
- 執(zhí)行以下命令消費消息
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic tp33
this test1
test2
- 現(xiàn)在讓我們測試一下容錯性巍沙。broker3 充當 leader 所以讓我們殺了它：
docker container rm kafka3
此時 kafka3 節(jié)點已經(jīng)關(guān)閉，但我們依舊可以發(fā)送/接受消息荷鼠。
再次查看 topic 相關(guān)信息
./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
