docker 安装 kafka+zookeeper，golang操作kafka

环境:

• centos8的服务器
• 安装好docker和docker-compose

首先需要开放对外的ip端口,如阿里云的

docker-compose.yml

mkdir -p /usr/local/kafka
vim docker-compose.yml

写入下面的代码

version: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    restart: always
    volumes:
      - ./data:/data
    ports:
      - 2181:2181

  kafka:
    image: wurstmeister/kafka
    restart: always
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 0
      #kafka会在zookeeper中使用这个参数进行注册，如果不设置，zk就无法向外提供broker的信息,这里填写kafka的ip地址
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.16.0.110:9092 
      # KAFKA_ADVERTISED_HOST_NAME: 182.92.234.23 # 注册到zk里的kafka HOST_NAME dkafka 0.9.x以后弃用
      KAFKA_CREATE_TOPICS: "test:2:0"   #kafka启动后初始化一个有2个partition(分区)0个副本名叫test的topic 
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_MESSAGE_MAX_BYTES: 10000000
      # kafka的tcp侦听ip地址 例如“127.0.0.1”，那么只有与该ip正确连接的客户端能成功连接到kafka；
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092 
      KAFKA_HEAP_OPTS: "-Xmx256M -Xms128M"
    volumes:
      - ./kafka-logs:/kafka
    depends_on:
      - zookeeper

kafka listeners 和 advertised.listeners 的区别及应用

https://blog.csdn.net/weixin_38251332/article/details/105638535

启动服务

docker-compose up -d

A窗口 查看/发送消息到web_log里

# 查看kafka版本
# echo $KAFKA_HOME
/opt/kafka
# cd /opt/
/opt # ls
kafka             kafka_2.13-2.8.1  overrides

# 进入kafka_kafka_1中
docker exec -it kafka_kafka_1 bash

#查看已经建好的topic （docker-compose.yml）
$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper kafka_zookeeper_1:2181 

#发送消息
$KAFKA_HOME/bin/kafka-console-producer.sh --topic=web_log --broker-list kafka_kafka_1:9092
>1
>2
>3
>

B窗口接收消息web_log里信息

# 另开一个窗口,进入kafka_kafka_1中
docker exec -it kafka_kafka_1 bash
#接收消息
$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server  kafka_kafka_1:9092 --from-beginning --topic web_log

1
2
3

停止服务

docker-compose down

kafka其它命令

二进制安装方式 zookeeper启动

/usr/local/kafka/bin/zookeeper-server-start.sh -daemon /usr/local/kafka/config/zookeeper.properties
netstat -tplnu | grep 2181

二进制安装方式 kafka启动

nohup /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
netstat -tplnu | grep 9092

查看kafka有哪些主题

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
./opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

crawler
wjes7
data_analysis
checkurl
grab

查看kafka 指定主题有没有消费

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group wjes7 --describe
./opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group grab --describe

TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
es_insert                      0          109             109             0          wjes7-e9e26bd0-4e4d-4d7a-9dd2-c2c45df38083        /172.34.1.162                  wjes7

查看kafka 指定主题消费情况

/opt/bitnami/kafka/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic=es_insert_full --from-beginning

go操作kafka

producer 消费者

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
)

var Topic = "web_log" //主题名称

// 基于sarama第三方库开发的kafka client
func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
    config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回


    // 连接kafka
    client, err := sarama.NewSyncProducer([]string{"182.92.234.24:9092"}, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    defer client.Close()
    //例子一发单个消息
    // 构造一个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = Topic
    content := "this is a test log"
    send01(client,msg,content)

    //例子二发多个消息
    for _, word := range []string{"Welcome11", "to", "the", "Confluent", "Kafka", "Golang", "client"} {
        send01(client,msg,word)
    }
}

//发消息
func send01(client sarama.SyncProducer,msg *sarama.ProducerMessage,content string)  {
    msg.Value = sarama.StringEncoder(content)

    // 发送消息
    pid, offset, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send msg failed, err:", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)

}

consumer 消费者

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
)

// kafka consumer

var Topic = "web_log" //主题名称

func main() {
    consumer, err := sarama.NewConsumer([]string{"182.92.234.24:9092"}, nil)
    if err != nil {
        fmt.Printf("fail to start consumer, err:%v\n", err)
        return
    }
    partitionList, err := consumer.Partitions(Topic) // 根据topic取到所有的分区
    if err != nil {
        fmt.Printf("fail to get list of partition:err%v\n", err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList { // 遍历所有的分区
        // 针对每个分区创建一个对应的分区消费者
        pc, err := consumer.ConsumePartition(Topic, int32(partition), sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
            return
        }
        defer pc.AsyncClose()
        // 异步从每个分区消费信息
        go func(sarama.PartitionConsumer) {
            for msg := range pc.Messages() {
                fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
            }
        }(pc)
    }
    select{} //阻塞进程
}

结合gin框架操作kafka

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
    "github.com/gin-gonic/gin"
    "sync"
    "time"
)

var Topic = "web_log" //主题名称
var kafkaIp = "182.92.234.24:9092"

func Test(ctx *gin.Context) {
    //读取
    ctx.JSON(200, gin.H{
        "data": "product",
    })
}

func main() {

    //启动消息者
    go InitConsumer()

    r := gin.Default()
    r.GET("/ping", func(c *gin.Context) {
        c.JSON(200, gin.H{
            "message": "pong",
        })
    })
    r.GET("/send", SendMessage)  //http://localhost:8082/send

    r.Run("0.0.0.0:8082") // 监听并在 0.0.0.0:8080 上启动服务



}

//发消息到kakfa
func SendMessage(ctx *gin.Context) {
    fmt.Println("SendMessage")
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
    config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
    config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回


    // 连接kafka
    client, err := sarama.NewSyncProducer([]string{kafkaIp}, config)
    if err != nil {
        fmt.Println("producer closed, err:", err)
        return
    }
    defer client.Close()
    //例子一发单个消息
    // 构造一个消息
    msg := &sarama.ProducerMessage{}
    msg.Topic = Topic
    content := "this is a test log"
    sendTokafka(client,msg,content)

    //例子二发多个消息
    for _, word := range []string{"Welcome11", "to", "the", "Confluent", "Kafka", "Golang", "client"} {
        sendTokafka(client,msg,word)
    }
}

//发消息函数
func sendTokafka(client sarama.SyncProducer,msg *sarama.ProducerMessage,content string)  {
    msg.Value = sarama.StringEncoder(content)

    // 发送消息
    pid, offset, err := client.SendMessage(msg)
    if err != nil {
        fmt.Println("send msg failed, err:", err)
        return
    }
    fmt.Printf("pid:%v offset:%v\n", pid, offset)

}


func InitConsumer() {
    time.Sleep(time.Second * 3)
    fmt.Println("init Counsumer success")

    var wg sync.WaitGroup
    consumer, err := sarama.NewConsumer([]string{kafkaIp}, nil)
    if err != nil {
        fmt.Printf("fail to start consumer, err:%v\n", err)
        return
    }
    partitionList, err := consumer.Partitions(Topic) // 根据topic取到所有的分区
    if err != nil {
        fmt.Printf("fail to get list of partition:err%v\n", err)
        return
    }
    fmt.Println(partitionList)


    for partition := range partitionList { // 遍历所有的分区
        wg.Add(1)
        // 针对每个分区创建一个对应的分区消费者
        pc, err := consumer.ConsumePartition(Topic, int32(partition), sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
            return
        }
        // 异步从每个分区消费信息
        go func(sarama.PartitionConsumer) {
            for msg := range pc.Messages() {
                fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
            }
            defer pc.AsyncClose()
            wg.Done()
        }(pc)
    }
    //select{} //阻塞进程
    wg.Wait()
    consumer.Close()
}

访问:
http://localhost:8082/send

结果:

SendMessage
pid:0 offset:250
Partition:0 Offset:250 Key:[] Value:this is a test log
Partition:0 Offset:251 Key:[] Value:Welcome11
pid:0 offset:251
Partition:0 Offset:252 Key:[] Value:to
pid:0 offset:252
pid:0 offset:253
Partition:0 Offset:253 Key:[] Value:the
Partition:0 Offset:254 Key:[] Value:Confluent
pid:0 offset:254
Partition:0 Offset:255 Key:[] Value:Kafka
pid:0 offset:255
pid:0 offset:256
Partition:0 Offset:256 Key:[] Value:Golang
Partition:0 Offset:257 Key:[] Value:client
pid:0 offset:257

go-queue操作kafka

参考地址：
github.com/zeromicro/go-queue/kq

参考文档：
https://www.jianshu.com/p/97fa257622ee
https://segmentfault.com/a/1190000021746086