kafka内外网配置

素颜马尾好姑娘i 2022-11-16 05:50 321阅读 0赞

# 看此文档，需要首先会安装kafka集群，zookeeper集群；熟悉组件命令； #

# 三台kafka集群（设置加密通道）： #

# 修改kafka server.properties  其中listeners 和 advertised.listeners 中 EXTERNAL的端口必须一致 #

## 1. 修改86 kafka 内外网配置 ##

listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19092
    advertised.listeners=INTERNAL://192.168.1.86:9092,EXTERNAL://外网IP:19092
    inter.broker.listener.name=INTERNAL
    listener.security.protocol.map = SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT

## 2. 修改87 kafka 内外网配置 ##

listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19093
    advertised.listeners=INTERNAL://192.168.1.87:9092,EXTERNAL://外网IP:19093
    inter.broker.listener.name=INTERNAL
    listener.security.protocol.map = SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT

## 3. 修改88 kafka 内外网配置 ##

listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19094
    advertised.listeners=INTERNAL://192.168.1.88:9092,EXTERNAL://外网IP:19094
    inter.broker.listener.name=INTERNAL
    listener.security.protocol.map = SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT

## 4. 注释配置 ##

# security.inter.broker.protocol=SASL_PLAINTEXT
    # listeners=SASL_PLAINTEXT://0.0.0.0:9092
    # advertised.listeners=SASL_PLAINTEXT://192.168.1.86:9092

## 5. 最终配置demo ##

broker.id=0
    authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
    # security.inter.broker.protocol=INTERNAL
    sasl.mechanism.inter.broker.protocol=PLAIN
    sasl.enabled.mechanisms=PLAIN
    super.users=User:admin
    # listeners=SASL_PLAINTEXT://0.0.0.0:9092
    # advertised.listeners=SASL_PLAINTEXT://192.168.1.86:9092
    num.network.threads=3
    num.io.threads=8
    socket.send.buffer.bytes=102400
    socket.receive.buffer.bytes=102400
    socket.request.max.bytes=104857600
    log.dirs=/opt/kafka-logs
    num.partitions=1
    num.recovery.threads.per.data.dir=1
    log.cleanup.policy=delete
    log.retention.hours=72
    log.segment.bytes=1073741824
    zookeeper.connect=192.168.1.86:2181,192.168.1.87:2181,192.168.1.88:2181
    delete.topic.enable=true
    zookeeper.connection.timeout.ms=60000
    
    
    listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:19092
    advertised.listeners=INTERNAL://192.168.1.86:9092,EXTERNAL://外网IP:19092
    inter.broker.listener.name=INTERNAL
    
    # 这个协议底层使用了哪种安全协议 PLAINTEXT  不加密；SASL_PLAINTEXT 普通加密；SSL 高级加密
    listener.security.protocol.map = SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT

# 操作kafka集群命令集合： #

# 如何进入docker：
    docker ps -a 查看所有进程
    docker exec -it 容器id bash / 使用管理员权限进入容器）
    docker exec -it -u root b667f9cfa164 bash
    docker exec -it test bash
    
    #杀掉zk和kafka
    ps -ef|grep -E "kafka|zookeeper"|grep -v grep |awk '{print $2}'|while read pid;do kill -9 $pid;done
    ps -ef|grep -E "kafka" |grep -v grep |awk '{print $2}'|while read pid;do kill -9 $pid;done
    ps -ef|grep -E "zookeeper"|grep -v grep |awk '{print $2}'|while read pid;do kill -9 $pid;done
    
    #启动zk （清除数据：需要删除 zoo.cfg 中 dataDir指向的文件夹下所有的非 myid的文件和文件夹）
    /usr/local/zookeeper/bin/zkServer.sh start /usr/local/zookeeper/conf/zoo.cfg
    
    # 查看zk的状态 （启动状态不正确时，查看zoo.cfg 中 dataDir 下的 zookeeper.out日志）
    /usr/local/zookeeper/bin/zkServer.sh status /usr/local/zookeeper/conf/zoo.cfg
    
    # 临时启动kafka （查看是否有异常，根据日志调试；查看server.properties中log.dirs指向 删除干净）
    /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
    
    # 启动kafka 
    nohup /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties >>/usr/local/kafka/logs/kafka.log 2>1 &
    
    # 查看topic列表
    /usr/local/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
    
    # 删除topic
    /usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 127.0.0.1:2181 --topic douzi-test
    
    # 创建topic
    /usr/local/kafka/bin/kafka-topics.sh --create --topic douzi-test --replication-factor 3 --partitions 3 --zookeeper localhost:2181
    
    # 接收消息：（普通模式）
    ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic douzi-test
    
    # 接收消息：（加密情况下的命令）
    /usr/local/kafka/bin/reader-kafka-console-consumer.sh --bootstrap-server 0.0.0.0:9092 --topic douzi-test --from-beginning --consumer.config /usr/local/kafka/producer.config
    
    # 发送消息：
    ./kafka-console-producer.sh --broker-list localhost:9092 --topic douzi-test
    
    # 发送消息：(加密情况下的命令)
    /usr/local/kafka/bin/writer-kafka-console-producer.sh --broker-list localhost:9092 --topic douzi-test --producer.config /usr/local/kafka/producer.config