Kafka自带zookeeper---集群安装部署

2023-09-18 11:40:47

kafka简介

kafka官网：http://kafka.apache.org/
kafka下载页面：http://kafka.apache.org/downloads
kafka配置快速入门：http://kafka.apache.org/quickstart

首先让我们看几个基本的消息系统术语：

•Kafka将消息以topic为单位进行归纳。
•将向Kafka topic发布消息的程序成为producers.
•将预订topics并消费消息的程序成为consumer.
•Kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker.
producers通过网络将消息发送到Kafka集群，集群向消费者提供消息，如下图所示：

客户端和服务端通过TCP协议通信。Kafka提供了Java客户端，并且对多种语言都提供了支持。

环境准备

kafka版本：kafka_2.11-1.0.0.tgz
三台主机IP：172.16.10.91、172.16.10.92、172.16.10.93

安裝配置工作

1、kafka安裝

　　下载后解压kafka，将其解压到/usr/local目录下，删除压缩包即可：

tar xzvf kafka_2.11-1.0.0.tgz  #解压

　　进入到kafka的config目录

　　自定义目录：

首先新建kafka的日志目录和zookeeper数据目录，因为这两项默认放在tmp目录，而tmp目录中内容会随重启而丢失,所以我们自定义以下目录:

mkdir /usr/local/kafka_2.-1.0./zookeeper        #创建zookeeper数据目录

mkdir /usr/local/kafka_2.-1.0./log              #创建日志目录

mkdir /usr/local/kafka_2.-1.0./log/zookeeper    #创建zookeeper日志目录

mkdir /usr/local/kafka_2.-1.0./log/kafka        #创建kafka日志目录

2、zookeeper配置

　　> 修改 zookeeper.properties

进入config目录下，修改关键配置如下，3台服务器的zookeeper.properties配置文件都一样

#修改为自定义的zookeeper数据目录
dataDir=/usr/local/kafka_2.-1.0./zookeeper
#修改为自定义的zookeeper日志目录
dataLogDir=/usr/local/kafka_2.-1.0./log/zookeeper
# 端口
clientPort=
#注释掉
#maxClientCnxns=
#设置连接参数，添加如下配置
tickTime=2000　　　　#为zk的基本时间单元，毫秒
initLimit=10　　　　 #Leader-Follower初始通信时限 tickTime*10
syncLimit=5　　　　　#Leader-Follower同步通信时限 tickTime*5
#设置broker Id的服务地址
server.=172.16.10.91::
server.=172.16.10.92::
server.=172.16.10.93::

> zookeeper数据目录添加myid配置

在各台服务器的zookeeper数据目录【/usr/local/kafka_2.11-1.0.0/zookeeper】添加myid文件，写入服务broker.id属性值

如这里的目录是/usr/local/kafka_2.11-1.0.0/zookeeper

注：3台服务器都要添加myid文件，但值的內容不一样，棘突关系请见下图：

例如：第一台broker.id为0的服务到该目录【/usr/local/kafka_2.11-1.0.0/zookeeper】下执行以下命令，会生成myid文件內容

[root@mmc config]# echo 0 > myid(无法写入myid值)

采用Vi  myid，写入“0”

3、kafka配置

　　进入config目录下，修改server.properties文件

注：3台服务器都要配置，其中【broker.id、advertised.listeners】不一樣，其它配置都一样

############################# Server Basics #############################
# broker 的全局唯一编号，不能重复
broker.id=1
 
 
############################# Socket Server Settings #############################
 
 
# 配置监听,修改为本机ip
 
 
advertised.listeners=PLAINTEXT://172.16.10.91:9092
 
 
# 处理网络请求的线程数量，默认
 
 
num.network.threads=
 
 
# 用来处理磁盘IO的线程数量，默认
 
 
num.io.threads=
 
 
# 发送套接字的缓冲区大小，默认
 
 
socket.send.buffer.bytes=
 
 
# 接收套接字的缓冲区大小，默认
 
 
socket.receive.buffer.bytes=
 
 
# 请求套接字的缓冲区大小，默认
 
 
socket.request.max.bytes=
 
 
############################# Zookeeper #############################
 
 
# 配置三台服务zookeeper连接地址
 
 
zookeeper.connect=172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181
 
 
 
 
############################# Log Basics #############################
 
 
# kafka 运行日志存放路径
 
 
log.dirs=/usr/local/kafka_2.11-1.0.0/log/kafka
 
 
# topic 在当前broker上的分片个数，与broker保持一致
 
 
num.partitions=3
 
 
# 用来恢复和清理data下数据的线程数量，默认
 
 
num.recovery.threads.per.data.dir=
 
 
############################# Log Retention Policy #############################
 
 
# segment文件保留的最长时间，超时将被删除，默认
 
 
log.retention.hours=
 
 
# 滚动生成新的segment文件的最大时间，默认
 
 
log.roll.hours=
 

 

4、kafka启动
 

　　kafka启动时先启动zookeeper，再启动kafka；关闭时相反，先关闭kafka，再关闭zookeeper
 

  • 进入kafka根目錄
 

 
 
[root@localhost kafka_2.-1.0.]# cd /usr/local/kafka_2.-1.0./
 

 

  • 启动zookeeper
 

　　输入命令：
 

 
 
[root@localhostkafka_2.-1.0.]# bin/zookeeper-server-start.sh config/zookeeper.properties &
 
 
 
 
#后台启动命令：
  
 
 
[root@localhostkafka_2.-1.0.]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties >log/zookeeper/zookeeper.log 2>1 &
 

 

查看log/zookeeper/zookeeper.log 文件，第1、2台服务器启动的時候，会报错Connection refused，因為另外1、2台服务器还没有启动，先不用管它
 

 

查看log/zookeeper/zookeeper.log 文件，等3台服务器都启动后，就不在报错了
 

  • 启动kafka
 

　　输入命令：
 

 
 
[root@localhostkafka_2.-1.0.]# bin/kafka-server-start.sh config/server.properties &
 
 
 
 
#后台启动命令：
 

 

备注：nohup命令
 

nohup 英文全称 no hang up（不挂起），用于在系统后台不挂断地运行命令，退出终端不会影响程序的运行。
 

 
 
[root@localhostkafka_2.-1.0.]# nohup bin/kafka-server-start.sh config/server.properties >log/kafka/kafka.log 2>1 &
 

 

　　查看log/kafka/kafka.log 文件，启动过程过程中没有报错，且有以下信息输出，说明启动成功
 

 

5、kafka测试验证
 

　　分别启动三台机器的zookeeper，三台机器的zookeeper都启动成功后，再分别启动三台机器的kafka。
 

  • 進入kafka根目錄
 

 
 
[root@localhost kafka_2.-1.0.]# cd /usr/local/kafka_2.-1.0./
 

 

  • 在某臺機器創建topic，名稱為test
 

 
 
[root@localhost kafka_2.11-1.0.0]# bin/kafka-topics.sh --create --zookeeper 172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181 --replication-factor 3 --partitions 3 --topic test
 
 
命令解析：
 
 
--create：　　　　　　　　　 #指定创建topic动作 
 
 
--topic：　　　　　　　　　　#指定新建topic的名称 
 
 
--zookeeper： 　　　　　　　#指定kafka连接zk的连接url，该值和server.properties文件中的配置项{zookeeper.connect}一样
 
 
--partitions 3　　　　　　 #指定当前创建的kafka分区数量，默认为1个
 
 
--replication-factor 　　#指定每个分区的副本集数量，不能大于 broker 节点数量，多了也没用，1个节点放>=2个副本挂了都完蛋。
 

 

 

  • 查看topic，确认topic创建成功
 

 
 
[root@localhost kafka_2.-1.0.]# bin/kafka-topics.sh --list --zookeeper 172.16.10.91:2181
 

 

 

  • 查看topic，详细信息
 

 
 
[root@localhost kafka_2.-1.0.]# bin/kafka-topics.sh --describe --zookeeper 172.16.10.93:2181 --topic test
 

 

可以描述为：Topic分区数/副本数/副本Leader/副本ISR等信息：
 

 

　　“leader”：该节点负责该分区的所有的读和写，每个节点的leader都是随机选择的。
 　　“replicas”：备份的节点列表，无论该节点是否是leader或者目前是否还活着，只是显示。
 　　“isr”：同步备份”的节点列表，也就是活着的节点并且正在同步leader
   　　　其中Replicas和Isr中的1,2,0就对应着3个broker他们的broker.id属性！
 

  • 在某台服务器上创建生产者(注意默认端口号是9092)
 

 
 
[root@localhost kafka_2.-1.0.]# bin/kafka-console-producer.sh --broker-list 172.16.10.91: 9092,172.16.10.92: 9092,172.16.10.93: 9092 --topic test
 

 

 

  • 在另外两台服务器创建消费者(注意默认端口号是9092)
 

 
 
[root@localhost kafka_2.-1.0.]# bin/kafka-console-consumer.sh --bootstrap-server 172.16.10.91:9092,172.16.10.92: 9092,172.16.10.93: 9092 --topic test --from-beginning
 

 

　　
 

  • 生产者服务器截图
 

 

  • 消費者服务器截图
 

 

  • 修改topic信息
 

 
 
bin/kafka-topics.sh --zookeeper 192.168.187.146: --alter --topic test --config max.message.bytes=
  
 
 
bin/kafka-topics.sh --zookeeper 192.168.187.146: --alter --topic test --delete-config max.message.bytes
  
 
 
bin/kafka-topics.sh --zookeeper 192.168.187.146: --alter --topic test --partitions
  
 
 
bin/kafka-topics.sh --zookeeper 192.168.187.146: --alter --topic test --partitions  ## Kafka分区数量只允许增加，不允许减少
 

 

  • 删除topic
 

注意，只是删除Topic在zk的元数据，日志数据仍需手动删除
 

 
 
[root@localhost kafka_2.-1.0.]# bin/kafka-topics.sh -delete --zookeeper 172.16.10.91:,172.16.10.92:,172.16.10.93: --topic test
 

 

 

Note: This will have no impact if delete.topic.enable is not set to true.## 默认情况下，删除是标记删除，没有实际删除这个Topic；如果运行删除Topic，两种方式：
   方式一：通过delete命令删除后，手动将本地磁盘以及zk上的相关topic的信息删除即可
   方式二：配置server.properties文件，给定参数delete.topic.enable=true，重启kafka服务，此时执行delete命令表示允许进行Topic的删除
 

  • 查看kafka未消费的消息
 

[root@localhost kafka_2.11-1.0.0]# bin/kafka-console-consumer.sh --bootstrap-server 172.16.10.91:9092,172.16.10.92:9092,172.16.10.93:9092 --topic logstash-busi-2 --from-beginning
 

 

  • 查看kafka消费進度
 

 
 
命令：bin/kafka-consumer-groups.sh --bootstrap-server <Kafka broker 连接信息> --describe --group <group 名称>
 
 
bin/kafka-consumer-groups.sh --bootstrap-server 172.16.10.91:9092,172.16.10.92:9092,172.16.10.93:9092 --describe --group terminalMonitor
 

 

 

 6、kafka测试集群的容错能力
 　　Kafka是支持容错的，下面我们来对Kafka的容错性进行测试，测试步骤如下
 

(1).查看topic的详细信息，观察那个blocker的角色是leader，那些blocker的角色是follower
 

 
 
[hadoop@Master ~]$ kafka-topics.sh --describe --zookeeper localhost: --topic my-replicated-topic
  
 
 
Topic:my-replicated-topic   PartitionCount:    ReplicationFactor: Configs:
  
 
 
    Topic: my-replicated-topic  Partition:     Leader: 2   Replicas: ,, Isr: ,,
 

 

(2).手工kill掉任意一个状态是follower的borker，测试生成和消费信息是否正确
 

步骤1中可以看到 2 为leader,1 和 3为 follower，将follower为1的进程kill掉
 

 

启动生产和消费者测试信息是否正确
 

 

结论：kill掉任意一个状态是follower的broker，生成和消费信息正确，不受任何影响
 

(3).手工kill掉状态是leader的borker，测试生产和消费的信息是否正确
 

borker2的角色为leader，将它kill掉，borker 3变成了leader
 

 

启动生产和消费者测试信息是否正确
 

 

结论：kill掉状态是leader的borker，生产和消费的信息正确

                

        

        

    

    

    

      
    

    

  





        
更多推荐




  

    

      AOSP源码中Android.mk文件中的反斜杠符号(\)的作用和使用
    

    

      简介在AOSP（AndroidOpenSourceProject）源码中的Android.mk文件中，反斜杠符号（\）的主要作用是将一行代码拆分成多行，以提高可读性并帮助组织较长的代码块。这对于定义复杂的构建规则和变量时特别有用。以下是\符号在Android.mk文件中的作用以及如何使用它的示例：多行命令：Androi
    

  

  
  
  





  

    

      Nacos使用教程（三）——nacos注册中心(2)
    

    

      文章目录什么是注册中心注册中心的作用1.服务注册2.服务发现3.负载均衡4.故障恢复注册中心的解决的问题1.服务管理问题2.服务调用问题3.负载均衡问题4.故障恢复问题服务的发现与注册的实现模式服务注册表Nacos注册中心的部署与使用什么是Nacos注册中心Nacos注册中心的部署下载Nacos安装包解压安装包修改配置
    

  

  
  
  





  

    

      【图像处理】怎么让图片背景变透明？怎么改变图片中线条的颜色？在线PS软件
    

    

      文章目录前言一、打开图二、背景透明处理三、改变线条颜色总结前言我们处理图片将会使用到PS软件，如果安装了PS那直接打开使用。如果没有安装PS，可使用在线PS软件：https://www.uupoop.com/#/一、打开图第一步，从这个入口打开图片。第二步：打开图片，我们打开一个MySQL数据库的图标。二、背景透明处理
    

  

  





  

    

      【AIGC】提示词 Prompt 分享
    

    

      提示词工程是什么？Promptengineering（提示词工程）是指在使用语言模型进行生成性任务时，设计和调整输入提示（prompts）以改善模型生成结果的过程。它是一种优化技术，旨在引导模型产生更加准确、相关和符合预期的输出。在生成性任务中，输入提示是指提供给语言模型的初始文本或问题，用以引导其生成后续的文本或回答
    

  

  
  
  





  

    

      无CDN场景下的传统架构接入阿里云WAF防火墙的配置实践
    

    

      文章目录1.配置网站接入WAF防火墙1.1.配置网站接入方式1.2.填写网站的信息1.3.WAF防火墙生成CNAME地址2.配置WAF防火墙HTTPS证书3.修改域名DNS解析记录到WAF防火墙4.验证网站是否接入WAF防火墙传统架构接入WAF防火墙非常简单，配置完WAF网站接入后，将得到CNAME地址配置在域名DNS
    

  

  
  
  





  

    

      Linux（Centos7）中安装Docker和DockerCompose
    

    

      一、安装DockerDocker分为CE和EE两大版本。CE即社区版（免费，支持周期7个月），EE即企业版，强调安全，付费使用，支持周期24个月。DockerCE分为`stable``test`和`nightly`三个更新频道。官方网站上有各种环境下的https://docs.docker.com/install/，这
    

  

  
  
  





  

    

      【Rust 基础篇】Rust Newtype模式：类型安全的包装器
    

    

      导言Rust是一种以安全性和高效性著称的系统级编程语言，其设计哲学是在不损失性能的前提下，保障代码的内存安全和线程安全。在Rust中，Newtype模式是一种常见的编程模式，用于创建类型安全的包装器。Newtype模式通过定义新的结构体包装器来包装现有的类型，从而在不引入运行时开销的情况下提供额外的类型安全性。本篇博客
    

  

  
  
  





  

    

      xshell---git上传文件到gitee远程仓库配置
    

    

      1.git下载如果没有xshell下没有下载过git，可以参考这篇的教程：Linux配置安装git详细教程下载后可以通过git--version查看git的版本号，验证是否安装成功2.新建仓库首先需要在gitee上注册一个账号然后再主页面点击右上边框的+号，选择新建仓库，建立一个仓库：然后填写新建仓库的名称，系统会根据
    

  

  
  
  





  

    

      GIT使用需知，哪些操作会导致本地代码变动
    

    

      系列文章目录手把手教你安装Git，萌新迈向专业的必备一步GIT命令只会抄却不理解？看完原理才能事半功倍！常用GIT命令详解，手把手让你登堂入室GIT实战篇，教你如何使用GIT可视化工具GIT使用需知，哪些操作会导致本地代码变动系列文章目录一、本地代码变动的本质1.远程跟踪分支2.贮藏区(stash)二、分支切换三、分支
    

  

  
  
  





  

    

      (vue的入门
    

    

      vue的入门一.Vue是什么二.Vue的特点及优势三.使用Vue的详细步骤四.Vue的基本语法五.Vue的生命周期一.Vue是什么Vue（发音为/“vjuː”/，类似于"view"）是一套用于构建用户界面的渐进式JavaScript框架。它是一个开源的、轻量级的MVVM（模型-视图-视图模型）框架，专注于实现数据驱动的
    

  

  
  
  





  

    

      vue +element 删除按钮操作 （删除单个数据 +删除页码处理 ）
    

    

      1.配置接口deleteItemById:"/api/goods/deleteItemById",//删除商品操作2.get请求接口//删除接口后台给我们返iddeleteItemById(params){returnaxios.get(base.deleteItemById,{params})}3.异步请求接口asy
    

  

  
  
  



      

      

        
        

          
热文推荐


    
  
  
  1. 
    
      
    
        PCIE基础知识-3
      
    
      
    
        PCIE三种传输方式：IO中断，DMA，peertopeer中断：PCI设备需要向内存（SDRAM）中写入一些数据，该PCI设备会向CPU请求一个中断，然后CPU首先先通过PCI总线把该PCI设备的数据读取到CPU内部的寄存器中，然后再把数据从内部寄存器写入到内存（SDRAM）中。DMA：直接内存访问（DMA，Dire
      
    
        
  
    2. 
  
  
  2. 
    
      
    
        ​CVPR 2023 | STAR Loss：减少人脸关键点标注歧义，实现人脸关键点SOTA精度
      
    
      
    
        论文链接：https://arxiv.org/pdf/2306.02763.pdf代码链接：https://github.com/ZhenglinZhou/STAR要解决的问题：人脸关键点检测标注中存在语义歧义问题。语义歧义是指不同的标注者对同一个面部特征点的位置有不同的理解，导致标注结果不一致，影响模型的收敛和准确性
      
    
        
  
    3. 
  
  
  3. 
    
      
    
        设计模式-代理模式
      
    
      
    
        “阁下有什么问题可以和我的代理律师谈即可”，为什么会有律师这个职业呢？随着法律法规的逐步完善，日益复杂，导致大部分的普通民众掌握的法律知识明显不足，进而无法在合适的时间点进行维权、规避风险。可见，律师的作用就是利用自身的专业知识帮助案件当事人处理其无法处理的事情。不仅只有律师，生活中处处可见这种代理模式的存在，比如婚庆
      
    
        
  
    4. 
  
  
  4. 
    
      
    
        麦肯锡：中国生成式AI市场现状和未来发展趋势
      
    
      
    
        本文来自《麦肯锡中国金融业CEO季刊》，版权归麦肯锡所有。该季刊主要围绕生成式AI（以下简称“GenAI”）主题，通过4大章节共8篇文章，全面深入分析了GenAI对各主要行业的影响、价值链投资机会、中国GenAI市场现状和未来趋势以及企业如何布局GenAI，从而真正挖掘其价值。随着ChatGPT的火爆出圈，GenAI成
      
    
        
  
    5. 
  
  
  5. 
    
      
    
        2023年8月京东空调行业品牌销售排行榜（京东数据报告）
      
    
      
    
        鲸参谋监测的京东平台8月份空调市场销售数据已出炉！鲸参谋数据显示，今年8月份，京东平台上空调的销量将近146万，环比降低约44%，同比降低约37%；销售额为41亿+，环比下降约45%，同比下降约40%。可以看到，8月份空调市场整体下滑。*数据源于鲸参谋-行业趋势分析（来自公开渠道获取，数据仅供参考）空调市场中，格力品牌
      
    
        
  
    6. 
  
  
  6. 
    
      
    
        Python编程指南：利用HTTP和HTTPS适配器实现智能路由
      
    
      
    
        目录HTTP和HTTPS适配器什么是智能路由利用HTTP和HTTPS适配器实现智能路由总结在Python编程中，利用HTTP和HTTPS适配器实现智能路由是一项非常实用的技能。智能路由可以根据不同的条件选择不同的路由，从而提高网络性能和用户体验。在本文中，我们将介绍如何使用Python编程语言和HTTP/HTTPS适配
      
    
        
  
    7. 
  
  
  7. 
    
      
    
        非对称加密、解密原理及openssl中的RSA示例代码
      
    
      
    
        一、【原理简介】非对称加密非对称加密，也被称为公钥加密，其中使用一对相关的密钥：一个公钥和一个私钥。公钥用于加密数据，私钥用于解密数据。公钥可以公开分享，而私钥必须保密。密钥生成:当一个用户或设备希望使用非对称加密时，要生成一对密钥：一个公钥和一个私钥。这两个密钥是数学上相关的，但从公钥中计算出私钥在计算上是不可行的。
      
    
        
  
    8. 
  
  
  8. 
    
      
    
        【操作系统笔记】内存寻址
      
    
      
    
        物理寻址主存（内存）计算机主存也可以称为物理内存，内存可以看成由若干个连续字节大小的单元组成的数组每个字节都有一个唯一的物理地址（PhysicalAddress）CPU访问内存前，先拿到内存地址，然后，通过内存地址访问内存中数据指令总线的分工数据总线：负责传输实际数据的地址总线：负责传输数据地址的，用来确定到底把数据传
      
    
        
  
    9. 
  
  
  9. 
    
      
    
        FPGA千兆网 UDP 网络视频传输，基于88E1518 PHY实现，提供工程和QT上位机源码加技术支持
      
    
      
    
        目录1、前言版本更新说明免责声明2、我这里已有的以太网方案3、设计思路框架视频源选择OV5640摄像头配置及采集动态彩条UDP协议栈UDP视频数据组包UDP协议栈数据发送UDP协议栈数据缓冲IP地址、端口号的修改TriModeEthernetMAC介绍以及移植注意事项88E1518PHYQT上位机和源码4、vivado
      
    
        
  
    10.