最近雾霾又有点严重了，再加上北山门拆除的建筑垃圾，一晚上拉土车疯狂咆哮，车实在是脏的不能再看了，花四十五洗一洗，还是那么靓丽。

好了， 废话不多说，今天就看一下kafka集群的搭建，当然介于囊中羞涩，还是以一台服务器搭一个伪集群，如下是kafka集群的工作过程

接下来我们就看一下搭建过程

• 第一步，搭建zookeeper集群，这个zookeeper集群的搭建请参考我之前的文章zookeeper集群搭建，当然你也可以使用kafka自带的zookeeper，我们使用zookeeper来管理broker

• 第二步，去官网下载kafka2.13版本：https://www.apache.org/dyn/closer.cgi?path=/kafka/3.0.0/kafka_2.13-3.0.0.tgz

• 第三步，将其压缩包copy至linux目录usr/local/kafka-cluster目录并进行解压，解压后赋值三份并重新命名kafka_2.13-{1~3}
  
  

  然后修改配置，文件路径config/server.properties，在这里还是建议使用VSCode修改
  
  主要修改的几个参数如下，broker.id唯一，不能重复，zookeeper.connect设置我们搭建的zookeeper集群地址，注意这里我使用的是域名，域名在我们的etc/hosts文件中
  

• # The id of the broker. This must be set to a unique integer for each broker.
  broker.id=0
  
  # The address the socket server listens on. It will get the value returned from 
  # java.net.InetAddress.getCanonicalHostName() if not configured.
  #   FORMAT:
  #     listeners = listener_name://host_name:port
  #   EXAMPLE:
  #     listeners = PLAINTEXT://your.host.name:9092
  listeners=PLAINTEXT://:9091
  
  # A comma separated list of directories under which to store log files
  log.dirs=/usr/local/kafka-cluster/kafka_2.13-1/kafka-logs
  
  # Zookeeper connection string (see zookeeper docs for details).
  # This is a comma separated host:port pairs, each corresponding to a zk
  # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  # You can also append an optional chroot string to the urls to specify the
  # root directory for all kafka znodes.
  zookeeper.connect=bruce:2181,bruce:2182,bruce:2183

  
  

  修改另外两个目录kafka配置文件，broker.id=1或2, listeners=PLAINTEXT://:9092或9093

• 第四步，接下来我们开始启动，三个全部启动，启动之前要确保zookeeper集群没有问题
  

• /usr/local/kafka-cluster/kafka_2.13-1/bin/kafka-server-start.sh -daemon ../config/server.properties
  /usr/local/kafka-cluster/kafka_2.13-2/bin/kafka-server-start.sh -daemon ../config/server.properties
  /usr/local/kafka-cluster/kafka_2.13-3/bin/kafka-server-start.sh -daemon ../config/server.properties

  完了以后如果没有报错，表明我们启动成功，使用命令jps查看创建的集群，我们发现三个broker都已经全部启动
  
  使用命令在第一台机器创建一个名为techarticle的topic试试

• /usr/local/kafka-cluster/kafka_2.13-1/bin/kafka-topics.sh --create --bootstrap-server bruce:9091 --replication-factor 1 --partitions 1 --topic techarticle

  创建完成后，我们查看一下，三个broker都有了这个topic
  

  接下来我们就简单测试一下，发送几条消息试试，命令如下

• /usr/local/kafka-cluster/kafka_2.13-1/bin/kafka-console-producer.sh --broker-list 172.26.233.95:9091 --topic techarticle

  
  

  这个命令完成后，你可以键入几条消息，我们再使用命令消费一下看看，鉴于阿里云服务器1核2G，已经卡死了，说不定zookeeper也挂了，所以我就不截图了。
  

• /usr/local/kafka-cluster/kafka_2.13-1/bin/kafka-console-consumer.sh --bootstrap-server 172.26.233.95:9091 --topic techarticle --from-beginning