go中简单使用kafka

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

windows上kafka的安装

1.安装jdk

下载地址:https://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html

下载需要注册oracle

添加环境变量JAVA_HOME=C:\Program Files\Java\jre1.8.0_211

2.安装Zookeeper

下载地址:https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/

①进入zookeeper的相关设置所在的文件目录，例如本文的：D:\zookeeper-3.4.14\conf

②将"zoo_sample.cfg"重命名为"zoo.cfg"

③打开zoo.cfg 找到并编辑：

dataDir=/tmp/zookeeper to D:/zookeeper-3.4.14/data或 D:/zookeeper-3.4.14/data(路径仅为示例，具体可根据需要配置)

④与配置jre类似，在系统环境变量中添加：

a.系统变量中添加ZOOKEEPER_HOME=D:\zookeeper-3.4.14

b.编辑系统变量中的path变量，增加%ZOOKEEPER_HOME%\bin

⑤在zoo.cfg文件中修改默认的Zookeeper端口(默认端口2181)

⑥打开cmd窗口，输入zkserver，运行Zookeeper

3.安装kafka

下载地址:https://www.apache.org/dyn/closer.cgi?path=/kafka/2.2.0/kafka_2.11-2.2.0.tgz

①进入kafka配置文件所在目录，D:\kafka_2.11-2.2.0\config

②编辑文件"server.properties"，找到并编辑：

log.dirs=/tmp/kafka-logs to log.dirs=D:/kafka_2.11-2.2.0/kafka-logs

③在server.properties文件中，zookeeper.connect=localhost:2181代表kafka所连接的zookeeper所在的服务器IP以及端口，可根据需要更改

修改advertised.host.name=服务器ip

④进入kafka安装目录D:\kafka_2.11-2.2.0,打开cmd启动

.\bin\windows\kafka-server-start.bat .\config\server.properties

补充知识点

topic

topic是存储消息的逻辑概念，不同的topic下的数据是分开存储的。不同的 topic 的消息是分开存储的，每个 topic 可以有多个生产者向它发送消息，也可以有多个消费者去消费其中的消息。

partition

一个 topic 可以划分多个分区partition（每个 Topic至少有一个分区partition），同一topic下的不同分区包含的消息是不同的。第i个分区分配在第 i mod n 个broker上。

每个消息在被添加到分区时，都会被分配一个offset（称之为偏移量），它是消息在此分区中的唯一编号，kafka通过offset 保证消息在分区内的顺序，offset的顺序不跨分区，即kafka 只保证在同一个分区内的消息是有序的。

offset

每个消息在被添加到分区时，都会被分配一个 offset（称之为偏移量），它是消息在此分区中的唯一编号，kafka 通过 offset 保证消息在分区内的顺序。offset 的顺序不跨分区，即 kafka 只保证在同一个分区内的消息是有序的；对于应用层的消费来说，每次消费一个消息并且提交以后，会保存当前消费到的最近的一个 offset。

docker安装

1、下载镜像
这里使用了wurstmeister/kafka和wurstmeister/zookeeper这两个版本的镜像

docker pull wurstmeister/zookeeper
docker pull wurstmeister/kafka

2、启动

启动zookeeper容器
docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

启动kafka
docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.0.168:2181,192.168.0.169:2181,192.168.0.170:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.0.170:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

这里面主要设置了4个参数

KAFKA_BROKER_ID=0 
KAFKA_ZOOKEEPER_CONNECT=192.168.0.168:2181,192.168.0.169:2181,192.168.0.170:2181
KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.0.170:9092
KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092

KAFKA_ZOOKEEPER_CONNECT  配置的是zookeeper的地址，可以单节点配置，也可以配置zookeeper集群多节点，用逗号隔开

中间两个参数的192.168.0.170改为宿主机器的IP地址，如果不这么设置，可能会导致在别的机器上访问不到kafka。

3、进入kafka容器

docker exec -it ${CONTAINER ID} /bin/bash

进入kafka默认目录 /opt/kafka_2.11-0.10.1.0

go运行代码

producer

package main

import (
    "fmt"
    "time"
    "github.com/Shopify/sarama"
)

func main() {

    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Partitioner = sarama.NewRandomPartitioner
    config.Producer.Return.Successes = true

    client, err := sarama.NewSyncProducer([]string{"192.168.3.118:9092"}, config)
    if err != nil {
        fmt.Println("producer close, err:", err)
        return
    }

    defer client.Close()
    for {
        msg := &sarama.ProducerMessage{}
        msg.Topic = "nginx_log"
        msg.Value = sarama.StringEncoder("this is a good test, my message is good")

        pid, offset, err := client.SendMessage(msg)
        if err != nil {
            fmt.Println("send message failed,", err)
            return
        }

        fmt.Printf("pid:%v offset:%v\n", pid, offset)
        time.Sleep(10 * time.Second)
    }
}

consumer

package main

import (
    "fmt"
    "strings"
    "sync"
    "github.com/Shopify/sarama"
    "time"
)

var (
    wg sync.WaitGroup
)

func main() {

    consumer, err := sarama.NewConsumer(strings.Split("127.0.0.1:9092", ","), nil)
    if err != nil {
        fmt.Println("Failed to start consumer: %s", err)
        return
    }
    partitionList, err := consumer.Partitions("nginx_log")
    if err != nil {
        fmt.Println("Failed to get the list of partitions: ", err)
        return
    }
    fmt.Println(partitionList)
    for partition := range partitionList {
        pc, err := consumer.ConsumePartition("nginx_log", int32(partition), sarama.OffsetNewest)
        if err != nil {
            fmt.Printf("Failed to start consumer for partition %d: %s\n", partition, err)
            return
        }
        defer pc.AsyncClose()
        go func(pc sarama.PartitionConsumer) {
            wg.Add(1)
            for msg := range pc.Messages() {
                fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
                fmt.Println()
            }
            wg.Done()
        }(pc)
    }
    time.Sleep(10*time.Second)
    wg.Wait()
    consumer.Close()
}