- 前言
- 目录结构
- docker-compose.yml 如下
- golang测试代码
- 基于docker环境和docker-compose
- 依赖zk环境docker zk搭建
- 当前kafka 2.8.1
zk1,2,3 组成集群 kafka1,kafka2,kafka3组成集群 zk0 为单机 kafka0 为单机
sheepkiller/kafka-manager kafka可视化管理工具
单机
集群
# cd kafka/ docker compose up .
version: '3.8'
networks:
default:
external:
name: zk-net
services:
kafka1:
image: wurstmeister/kafka
restart: unless-stopped
container_name: kafka1
hostname: kafka1
ports:
- "9092:9092"
# - "1099:1099"
external_links:
- zk1
- zk2
- zk3
environment:
KAFKA_BROKER_ID: 1
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.31.29:9092 # 宿主机ip
KAFKA_ADVERTISED_HOST_NAME: kafka1
KAFKA_ADVERTISED_PORT: 9092
KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
# CUSTOM_JMX_PORT: "1099"
# JVM_XMS: "256M"
# JVM_XMX: "512M"
volumes:
- ./kafa1/data:/data
- ./kafa1/datalog:/datalog
kafka2:
image: wurstmeister/kafka
restart: unless-stopped
container_name: kafka2
hostname: kafka2
ports:
- "9093:9092"
# - "1100:1100"
external_links:
- zk1
- zk2
- zk3
environment:
KAFKA_BROKER_ID: 2
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.31.29:9093 # 宿主机ip
KAFKA_ADVERTISED_HOST_NAME: kafka2
KAFKA_ADVERTISED_PORT: 9093
KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
# CUSTOM_JMX_PORT: "1100"
# JVM_XMS: "256M"
# JVM_XMX: "512M"
volumes:
- ./kafa2/data:/data
- ./kafa2/datalog:/datalog
kafka3:
image: wurstmeister/kafka
restart: unless-stopped
container_name: kafka3
hostname: kafka3
ports:
- "9094:9092"
# - "1101:1101"
external_links:
- zk1
- zk2
- zk3
environment:
KAFKA_BROKER_ID: 3
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.31.29:9094 # 宿主机ip
KAFKA_ADVERTISED_HOST_NAME: kafka3
KAFKA_ADVERTISED_PORT: 9094
KAFKA_ZOOKEEPER_CONNECT: "zk1:2181,zk2:2181,zk3:2181"
# CUSTOM_JMX_PORT: "1101"
# JVM_XMS: "256M"
# JVM_XMX: "512M"
volumes:
- ./kafa3/data:/data
- ./kafa3/datalog:/datalog
# 单机kafka
kafka0:
image: wurstmeister/kafka
restart: unless-stopped
container_name: kafka0
hostname: kafka0
ports:
- "9095:9092"
# - "1102:1102"
external_links:
- zk1
- zk2
- zk3
environment:
KAFKA_BROKER_ID: 0
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.31.29:9095 # 宿主机ip
KAFKA_ADVERTISED_HOST_NAME: kafka0
KAFKA_ADVERTISED_PORT: 9095
KAFKA_ZOOKEEPER_CONNECT: "zk0:2181"
# CUSTOM_JMX_PORT: "1102"
# JVM_XMS: "256M"
# JVM_XMX: "512M"
volumes:
- ./kafa0/data:/data
- ./kafa0/datalog:/datalog
# Kafka 图形管理界面
kafka-manager:
image: sheepkiller/kafka-manager
restart: unless-stopped
container_name: kafka-manager
hostname: kafka-manager
ports:
- "9000:9000"
links: # 连接本compose文件创建的container
- kafka1
- kafka2
- kafka3
external_links: # 连接外部compose文件创建的container
- zk1
- zk2
- zk3
environment:
ZK_HOSTS: zk1:2181,zk2:2181,zk3:2181
KAFKA_BROKERS: kafka1:9092,kafka2:9093,kafka3:9094,kafka0:9095
# APPLICATION_SECRET: letmein
# KAFKA_MANAGER_AUTH_ENABLED: "true"
# KAFKA_MANAGER_USERNAME: admin
# KAFKA_MANAGER_PASSWORD: admin
golang测试代码
package main
import (
"context"
"fmt"
"github.com/Shopify/sarama"
"sync"
"time"
)
func InitKafkaProducer(addr []string) sarama.SyncProducer {
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll //ack模式 -1
config.Producer.Partitioner = sarama.NewRandomPartitioner //分片器采用随机分片
config.Producer.Return.Successes = true
client, err := sarama.NewSyncProducer(addr, config)
if err != nil {
panic(err)
}
return client
}
func InitKafkaConsumer(addr []string) sarama.Consumer {
config := sarama.NewConfig()
config.Consumer.Offsets.Initial = sarama.OffsetNewest //重最新地方开始读
client, err := sarama.NewConsumer(addr, config)
if err != nil {
panic(err)
}
return client
}
func SendToKafka(client sarama.SyncProducer, topic, data string) error {
msg := &sarama.ProducerMessage{}
msg.Topic = topic
msg.Key = sarama.StringEncoder(data)
msg.Value = sarama.StringEncoder(data)
pid, offset, err := client.SendMessage(msg)
if err != nil {
fmt.Println("send msg failed, err:", err)
return err
}
fmt.Println("send msg ", topic, data, pid, offset)
return nil
}
func KafkaTo(ctx context.Context, client sarama.Consumer, topic string, f func(msg *sarama.ConsumerMessage)) error {
partitions, err := client.Partitions("test")
if err != nil {
fmt.Println("deal msg failed, err:", err)
return err
}
var wg sync.WaitGroup
for partition := range partitions {
pc, err := client.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
if err != nil {
fmt.Printf("Failed to start consumer for partition %d: %sn", partition, err)
}
wg.Add(1)
go func(sarama.PartitionConsumer) {
defer pc.AsyncClose()
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case msg := <-pc.Messages():
//发送消息
f(msg)
}
}
}(pc)
}
wg.Wait()
return nil
}
func main() {
p := InitKafkaProducer([]string{"127.0.0.1:9092", "127.0.0.1:9093", "127.0.0.1:9094"})
defer p.Close()
c := InitKafkaConsumer([]string{"127.0.0.1:9092", "127.0.0.1:9093", "127.0.0.1:9094"})
defer c.Close()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
go func() {
t := time.NewTicker(time.Second)
for {
select {
case <-ctx.Done():
//5秒超时关闭
return
case <-t.C:
//每秒发送信息
err := SendToKafka(p, "test", fmt.Sprintf("test%d", time.Now().Unix()))
if err != nil {
panic(err)
}
}
}
}()
// 接收消息
err := KafkaTo(ctx, c, "test", func(msg *sarama.ConsumerMessage) {
fmt.Printf("deal msg Partition:%d, Offset:%d, key:%s, value:%sn", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
})
if err != nil {
panic(err)
}
}
