kafka集群搭建

下载地址

开始部署

  1. 进入项目前的目录 cd /home/test/
  2. 创建项目目录 mkdir kafka
  3. 进入项目目录 cd /home/test/kafka
  4. 创建kafka消息目录，主要存放kafka消息 mkdir kafkalogs
  5. 进入项目目录 cd /home/test/kafka
  6. 上传 kafka_2.10-0.10.0.0.tgz 安装包
  7. 解压 tar zxvf kafka_2.10-0.10.0.0.tgz -C ./
  8. cd /home/test/kafka/kafka_2.11-0.10.0.0/config
  9. 打开配置文件 vim server.properties
 1------------------------配置文件-------------------------------------
 2broker.id=1
 3/* 这是这台虚拟机上的值，在另外两台虚拟机上应该是2或者3，这个值是唯一的，每台虚拟机或者叫服务器不能相同。 */
 4listeners=PLAINTEXT://test01:9092
 5/设置本机IP和端口。我这里设置的是listeners，
 6也可以直接设置host.name=test01,port=9092,
 7这个IP地址也是与本机相关的，每台服务器上设置为自己的IP地址。/
 8
 9log.dirs=/home/test/kafka/kafkalogs
10
11#指定其与另外几台一样的ip
12zookeeper.connect=test01:2181,test01:2181,test02:2181
13delete.topic.enable=true
14
15num.network.threads=3 #这个是borker进行网络处理的线程数
16num.io.threads=8 #这个是borker进行I/O处理的线程数
17
18socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
19socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
20socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
21num.partitions=1 #默认的分区数，一个topic默认1个分区数
22log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天
23message.max.byte=5242880  #消息保存的最大值5M
24default.replication.factor=2  #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务
25replica.fetch.max.bytes=5242880  #取消息的最大直接数
26log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
27log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168 ），到目录查看是否有过期的消息如果有，删除
28log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能
29

启动

普模式启动

1bin/kafka-server-start.sh config/server.properties  >/dev/null 2>&1 &

守护模式启动

1bin/kafka-server-start.sh -daemon config/server.properties >/dev/null 2>&1 &