搭建kafka集群

落日映苍穹つ 2022-08-18 05:40 259阅读 0赞

**1 kafka简介**

Kafka is a distributed,partitioned,replicated commit logservice。
    它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现。
    kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,
    此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。无论是kafka集群，
    还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。

**2 下载kafka0.8**

http://ftp.tc.edu.tw/pub/Apache/kafka/0.8.2.0/kafka_2.10-0.8.2.0.tgz

**3 解压,创建软连接**

tar -zxvf kafka_2.10-0.8.2.0.tgz 
    mv -R kafka_2.10-0.8.2.0 /opt/
    ln -s kafka_2.10-0.8.2.0 kafka

**3 配置**

修改kafka/config/server.properties,修改其中broker.id,log.dirs,zookeeper.connect，port，其他项根据需要自行斟酌：
    # The id of the broker. This must be set to a unique integer for each broker.
    broker.id=1
    # The port the socket server listens on
    port=9091
    # A comma seperated list of directories under which to store log files
    log.dirs=/mnt/hd01/hadoop/kafka/logs
    # root directory for all kafka znodes.
    zookeeper.connect=db1:2181,db2:2181,db3:2181

**4 启动db1的kafka**

/opt/kafka/kafka-server-start.sh /opt/kafka/config/server.properties &

**5 复制kafka目录所有类容到其他节点**

scp -r ./kafka db2:/opt/
    scp -r ./kafka db3:/opt/

**6 修改db2,db3节点上的/config/server.properties**

只需修改两个值，其他配置和db1上的/config/server.properties保持一致：
    broker.id=2
    port=9092

**7 分别起动db1，db2，db3上的kafka**

在db1上执行：
    /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &
    ssh db2 '/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &'
    ssh db3 '/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &'

**8 创建Topic**

[root@db1 bin]# /opt/kafka/bin/kafka-topics.sh --create --zookeeper db1:2181 --replication-factor 1 --partitions 1 --topic mytopic
    Created topic "mytopic".

**9 查看topic情况**

[root@db1 bin]# /opt/kafka/bin/kafka-topics.sh --list --zookeeper db2:2181
    mytopic
    test
    参数db1，db2，db3，结果都是一致的

**10 创建发送者**

[root@db1 bin]# /opt/kafka/bin/kafka-console-producer.sh --broker-list db1:9091 --topic test
    [2015-11-10 05:54:35,370] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
    111
    [2015-11-10 05:54:38,534] INFO Closing socket connection to /192.168.10.155. (kafka.network.Processor)
    123
    qwe
    hello
    [2015-11-10 05:55:43,543] INFO Closing socket connection to /192.168.10.155. (kafka.network.Processor)

**11 创建消费者**

创建消费者之后，可以获取topic：test 发送的所有类容
    [root@db1 bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
    111
    123
    qwe
    hello