本文共 4116 字，大约阅读时间需要 13 分钟。

配置Zookeeper和Kafka及使用Golang连接Kafka

1. 配置Zookeeper

配置zoo.cfg

在Zookeeper的配置文件zoo.cfg中，添加以下内容：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasljaasLoginRenew=3600000
quorum.auth.enableSasl=true
quorum.auth.learnerRequireSasl=true
quorum.auth.serverRequireSasl=true
quorum.auth.learner.loginContext=QuorumLearner
quorum.auth.server.loginContext=QuorumServer
quorum.cnxn.threads.size=20

创建zk_server_jaas.conf

在zk_server_jaas.conf文件中，添加以下内容：

# Server-to-Server authentication configuration
Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username=admin
    password=admin
    user_kafka=admin
    user_producer=admin
}
# Quorum Server and Learner authentication configuration
QuorumServer {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_zookeeper="zookeeper@password";
}
QuorumLearner {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="zookeeper"
    password="zookeeper@password";
}

启动Zookeeper

在Docker中运行以下命令：

docker run -d -p 2181:2181 -p 2888:2888 -p 3888:3888 \
--name zookeeper --restart always \
-v /path/to/zookeeper/conf:/opt/zookeeper-3.4.13/conf \
-v /data/zookeeper/data:/data \
-v /data/zookeeper/datalog:/datalog \
-e "TZ=Asia/Shanghai" \
-e "SERVER_JVMFLAGS=-Djava.security.auth.login.config=/opt/zookeeper-3.4.13/conf/zk_server_jaas.conf" \
wurstmeister/zookeeper

2. 配置Kafka

创建kafka_server_jaas.conf

在kafka_server_jaas.conf文件中，添加以下内容：

# Kafka Server authentication configuration
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_admin="admin"
    user_alice="alice"
}
# Kafka Client authentication configuration
Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
}

配置server.properties

在server.properties中，添加以下内容：

KAFKA_OPTS: -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf
KAFKA_LISTENERS: SASL_PLAINTEXT://yourHostIP:19092
KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://yourHostIP:19092
KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SASL_PLAINTEXT
KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
KAFKA_SASL_ENABLED_MECHANISMS: PLAIN

启动Kafka

使用Docker运行以下命令：

version: '2'
services:
  kafka1:
    image: wurstmeister/kafka
    container_name: kafka1
    hostname: broker1
    network_mode: host
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zk0:2181,zk1:2181,zk2:2181
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_LISTENERS=SASL_PLAINTEXT://yourHostIP:19092
      - KAFKA_ADVERTISED_LISTENERS=SASL_PLAINTEXT://yourHostIP:19092
      - KAFKA_SECURITY_INTER_BROKER_PROTOCOL=SASL_PLAINTEXT
      - KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
      - KAFKA_SASL_ENABLED_MECHANISMS=PLAIN
      - KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf
      - KAFKA_PORT=19092
    volumes:
      - /path/to/kafka/config:/opt/kafka/config
      - /data/kafka-sasl/logs:/kafka/logs

3. 使用Golang连接Kafka

以下是一个Golang示例，用于连接Kafka：

package main
import (
    "fmt"
    "github.com/Shopify/sarama"
    "time"
)
func connKafka(nodes []string) {
    config := sarama.NewConfig()
    config.Net.SASL.Enable = true
    config.Net.SASL.User = "admin"
    config.Net.SASL.Password = "admin"
    config.Net.DialTimeout = 2 * time.Second
    config.Metadata.Retry.Max = 1
    now := time.Now()
    client, err := sarama.NewClient(nodes, config)
    if err != nil {
        fmt.Println("connect time: ", time.Since(now).Seconds())
        panic(err)
    }
    defer client.Close()
    topics, err := client.Topics()
    if err != nil {
        panic(err)
    }
    for _, topic := range topics {
        fmt.Println("topic: ", topic)
    }
}
func main() {
    nodes := []string{
        "kafka0:19092", "kafka1:19092", "kafka2:19092", "kafka3:19092", "kafka4:19092",
    }
    connKafka(nodes)
}

注意事项

通过以上步骤，您可以成功配置并使用Golang连接到Kafka。