kafka伪分布式安装

柔光的暖阳◎ 2022-06-02 00:05 182阅读 0赞

必须先安装zookeeper，参考：http://blog.csdn.net/u012572955/article/details/78910393
    
    解压文件至 /usr/local/tanlei/kafkacluster/
    tar -zxvf kafka_2.11-0.9.0.1.tgz
    
    
    这里只创建了2个节点，复制两个节点:
    cp -r ./kafka_2.11-0.9.0.1/ ./kafka1
    cp -r ./kafka_2.11-0.9.0.1/ ./kafka2
    
    
    创建两个节点的消息文件夹，对应log.dirs=/usr/local/tanlei/kafkacluster/kafka2/kafkalogs
    mkdir /usr/local/tanlei/kafkacluster/kafka1/kafkalogs
    mkdir /usr/local/tanlei/kafkacluster/kafka2/kafkalogs
    
    
    修改各自config下的server.properties文件:
    kafaka1的config下该文件:
    broker.id=0
    port=9091
    host.name=10.1.5.71
    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=/usr/local/tanlei/kafkacluster/kafka1/kafkalogs
    num.partitions=1
    num.recovery.threads.per.data.dir=5
    log.flush.interval.message=10000
    log.flush.interval.ms=1000
    log.retention.hours=168
    log.retention.bytes=1073741824
    log.segment.bytes=1073741824
    log.retention.check.interval.ms=300000
    zookeeper.connect=10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183
    zookeeper.connect.timeout.ms=6000
    
    
    kafaka1的config下该文件:
    broker.id=1
    port=9092
    host.name=10.1.5.71
    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=/usr/local/tanlei/kafkacluster/kafka2/kafkalogs
    num.partitions=1
    num.recovery.threads.per.data.dir=5
    log.flush.interval.message=10000
    log.flush.interval.ms=1000
    log.retention.hours=168
    log.retention.bytes=1073741824
    log.segment.bytes=1073741824
    log.retention.check.interval.ms=300000
    zookeeper.connect=10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183
    zookeeper.connect.timeout.ms=6000
    
    
    字段解释:
    broker.id=0         #当前机器在kafka机器里唯一标识，与zookeeper的myid一个意思，由于我使用独立zookeeper这里可以注释掉
    port=9092          #socket监听的端口
    host.name=192.168.152.192    #broker绑定的IP，我这里三台机器都绑定了，后面远程复制的时候记得修改
    num.network.threads=3        #broker处理网络的线程数目
    num.io.threads=8                  #broker处理IO的线程数目
    socket.send.buffer.bytes=102400     #kafka发送数据缓冲区的大小，到达阈值才发送
    socket.receive.buffer.bytes=102400     #kafka接收数据缓冲区的大小，达到阈值才落地
    socket.request.max.bytes=104857600    #kafka所能接受的数据最大值，100MB，这个值应该小于机器配置的Java heap，一般Java heap为内存的1/4
    log.dirs=/home/kafka/kafkalogs        #存储消息的目录位置
    num.partitions=1       #每个主题的默认分区数目，更多的分区可以允许更多的消费，可以提高性能
    num.recovery.threads.per.data.dir        #在kafka启动和关闭flush时，还原日志时，对于每个数据目录所能使用的线程数目
    log.flush.interval.message=10000        #当消息达到10000条时就会强制刷一次数据到磁盘
    log.flush.interval.ms=1000                    #每过1秒就会强制刷数据到磁盘，单位毫秒
    log.retention.hours=168                        #消息最大持久化时间，单位小时，即7天
    log.retention.bytes=1073741824          #一种持久化策略，当剩余段文件不低于这个阈值，当前段就会从日志消息中删除
    log.segment.bytes=1073741824          #一个段文件的最大值，达到最大值会另起一个新的segment
    log.retention.check.interval.ms=300000   #每300秒去检验一下段文件是否能被删除
    zookeeper.connect=192.168.152.192:12181,192.168.152.193:12181,192.168.152.194:12181  #设置zookeeper的端口
    zookeeper.connect.timeout.ms=6000    #连接zookeeper的超时时间
    
    
    分别启动各节点:
    进入bin目录下，输入命令行
    ./kafka-server-start.sh ../config/server.properties
    
    
    
    
    创建启动脚本:
    /usr/local/tanlei/kafkacluster/kafka1/bin/kafka-server-start.sh /usr/local/tanlei/kafkacluster/kafka1/config/server.properties | /usr/local/tanlei/kafkacluster/kafka2/bin/kafka-server-start.sh /usr/local/tanlei/kafkacluster/kafka2/config/server.properties
    
    
    创建topic:
    ./kafka-topics.sh --zookeeper 10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183 --topic test1 --replication-factor 1 --partitions 1 --create
    观察topic:
    ./kafka-topics.sh --zookeeper 10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183 --topic test1 --describe

kafka 文档中文翻译:

[http://orchome.com/451][http_orchome.com_451]

[http_orchome.com_451]: http://orchome.com/451