kafka

[复制链接]
发表于 2026-2-10 17:02:45 | 显示全部楼层 |阅读模式

马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有账号?立即注册

×
kafka带入集群当中,实现elkfk
zookeeper集群+kafka集群
1、zookeeper

1.1概念

zookeeper是一个开源的、分布式的,为分布式架构提供和谐的apache的项目
1.2作用

生存元数据
1.3zookeeper的工作机制

观察者模式计划的分布式服务器管理架构
负责存储和管理元数据,记载集群的变革,生存集群变革的信息
1.4zookeeper的特点
  1. 1、在集群中分为领导者和追随者,组成的集群
  2. 2、只要有半数以上的节点正常工作,整个zookeeper就可以正常工作。zookeeper集群在部署时一般选择奇数台
  3. 3、全局数据一致,每个zookeeper不论是领导者还是追随者,在访问他们的数据时都是一致的
  4. 4、数据更新的原子性,一次更新数据,要么都成功,要么都失败
  5. 5、数据更新的实时性
  6. 6、领导者和追随者根据投票产生
复制代码
1.5推举机制
  1. 1、服务器A启动,发起一次选举,A会投自己一票。A有一票不够半数,选举无法完成,A进入looking状态
  2. 2、服务器B启动,再发起一次选举,服务器B也投自己一票,服务器A和B会做个比较,谁的myid大,如果A比B小,A会把票改投给B,B有两票,B自动当选为leader
  3. 3、C启动,自动成为追随者,A也会成为追随者
复制代码
1.6摆设zookeeper
  1. mysql1、2、3
  2. systemctl stop firewalld
  3. setenforce 0
  4. #时间同步
  5. yum -y install ntpdate
  6. ntpdate ntp.aliyun.com
  7. date
  8. #安装java环境
  9. yum -y install java
  10. java -version
  11. #zookeeper安装包解压
  12. tar -xf apache-zookeeper-3.5.7-bin.tar.gz
  13. mv apache-zookeeper-3.5.7-bin /opt/zookeeper
  14. cd zookeeper
  15. cd conf
  16. cp zoo_sample.cfg zoo.cfg
  17. #修改配置文件
  18. vim zoo.cfg
  19. 2行 ticktime=2000
  20. #通信心跳时间,zookeeper服务端和客户端之间通信的间隔时间,单位是毫秒
  21. 5行 initlimit=10
  22. #leader和follower初始连接时最多能容忍的心跳数,单位是秒
  23. 8行 synclimit=5
  24. #leader和followe之间同步通信的超时时间,5*2的时间发生超时,leader就认为follower死了,会从集群中将其删除
  25. 12行 datadir=/opt/zookeeper/data
  26. #zookeeper的数据保存目录
  27. 13行 dataLogDir=/opt/zookeeper/logs
  28. #日志日志保存目录
  29. 15行 clientPort=2181
  30. #客户端连接服务器的指定端口
  31. 16行添加
  32. server.1=192.168.230.21:3188:3288
  33. server.2=192.168.230.22:3188:3288
  34. server.3=192.168.230.23:3188:3288
  35. server.1:数字id,也就是服务器对应的myid
  36. 192.168.230.21:服务器的ip地址
  37. 3188:zookeeper集群内部通信的端口
  38. 3288:重新选举端口,万一leader挂了,就用这个端口进行内部通信,选举新的leader
  39. -------------------------------------------------------------------------------------------------------------
  40. #创建数据目录和日志日志目录
  41. mkdir /opt/zookeeper/data
  42. mkdir /opt/zookeeper/logs
  43. #在每个节点的dataDir指定的目录下创建一个 myid 的文件
  44. echo 1 > /mkdir /opt/zookeeper/data/myid
  45. echo 2 > /mkdir /opt/zookeeper/data/myid
  46. echo 3 > /mkdir /opt/zookeeper/data/myid
  47. #配置zookeeper启动脚本
  48. vim /etc/init.d/zookeeper
  49. #设置开机自启
  50. chmod +x /etc/init.d/zookeeper
  51. chkconfig --add zookeeper
  52. #启动zookeeper
  53. service zookeeper start
  54. #查看zookeeper状态
  55. service zookeeper status
复制代码
2、kafka

2.1概念

消息队列:MQ
在高并发情况下,同步的哀求来不及处理处罚,哀求太对会造成壅闭
好比大量哀求并发到数据库,出现too many connection报错,消息队列使用异步处理处罚方式,可以缓解体系处理处罚哀求的压力
2.2kafka的作用
  1. 1、异步处理
  2. 2、系统解耦
  3. 每个系统之间独立运行,互相之间没有必然的依赖关系。
  4. 微服务架构中的通信对于解耦来说至关重要,各个微服务之间独立运行,分别处理各自的请求和消息,提高整个系统的吞吐量和处理能力。比如电商的订单系统,网站的工单系统,是典型的一个消息队列场景
  5. 3、负载均衡
  6. 消息队列的负载均衡,把任务发送到多个消费者,多个消费者可以并行处理队列中的消息
  7. 4、流量控制和限流
  8. 通过延迟方法,处理生产速率和消费者的处理速度(代码控制)
  9. 5、数据同步和分发
  10. 跨系统的系统同步和日志日志收集
  11. 6、任务调度和定时任务
  12. 7、实时数据处理
  13. 8、备份和恢复
复制代码
2.3消息队列的模式
  1. 1、点对点,一对一,消费者消费完数据之后,生产者会自动清除已消毒的数据。一个生产者对应一个消费者(淘汰)
  2. 2、发布/订阅模式(一对多模式,观察者模式):消费者数据在消费完之后不会被清除(保留一段时间)。生产者发布一个消息,可以是一个消费者使用,也可以是多个消费者同时使用(主流)
  3. kafka就是发布/订阅模式的消息队列,用于大数据的实时处理领域。RABBITMQ也是发布/订阅模式的消息队列,小集群内部使用
复制代码
2.4kafka的特性
  1. 1、高吞吐量,低延迟。每秒可以处理几十万条数据,延迟只有几毫秒
  2. 2、集群的可扩展性(支持热扩展)
  3. 3、消息的持久化:生产者发布的消息可以保存到吸盘当中,防止数据丢失(有时间限制)
  4. 3、容错性:挂了一个可以继续使用
  5. 4、高并发:数千个客户端可以同时读写
复制代码
2.5kafka的组件
  1. 1、topic:主题,kafka的基本单元,所有生产者发布的消息都是发到主题,消费者订阅主题,然后消费生产者发布的消息
  2. 2、生产者:把消息发布到主题
  3. 3、消费者:订阅主题消费,消费生产者发布的消息
  4. 4、分区:每个主题都可以分成多个分区,每个分区都是数据的有序子集
  5. 分区当中保留数据,按照偏移量来有序的存储数据,消费者可以根据偏分区当中的偏移量来消费指定分区当中的消息(一般不用)
  6. 分区还有备份的作用,在创建主题时创建分区,创建分区时要在指定副本数。
  7. 分区和执行的集群机器数量一般是保持一致的
  8. 副本:备份分区中的消息,最少要2个,互为备份
  9. 4、偏移量:消息在分区当中的唯一标识,跟踪和定位消息所在的位置。消费者可以根据偏移量来处理信息/5
  10. 5、经纪人broker:处理生产者和消费者的请求(kafka自己),元数据的经纪人/6
  11. 6、zookeeper:保存元数据
复制代码
2.6kafka的工作流程

生产者将消息发布到指定的主题,每个消息都附带一个key和value
主题是有多个分区的,生产者把消息写入一个分区(带偏移量)
经纪人(kafka)分配和处理处罚生产者的发布哀求,偏移量也是经纪人分配(在分区中是唯一的)
斲丧者订阅主题,获取全量的斲丧者的斲丧信息(默认模式),也可以从指定的分区获取消息(代码来完成,一样平常不消)
生产者发布的消息会在本地生存一段时间,防止斲丧者有耽误大概处理处罚速率过慢导致没有乐成斲丧。生存时间默认是7天
2.7摆设kafka
  1. #kafka安装包解压
  2. tar -xf kafka_2.13-3.4.1.tgz
  3. mv kafka_2.13-3.4.1 /usr/local/kafka
  4. cd /usr/local/kafka/config
  5. #配置文件备份
  6. cp server.properties server.properties.bak
  7. #修改配置文件
  8. vim server.properties
  9. 24行 broker.id=0(经纪人id要不一样)
  10. 34行 listeners=PLAINTETX://192.168.230.21:9092
  11. #监听本地地址
  12. 44行 num.network.threads=3
  13. #broker处理网络请求的线程数,一般不用修改
  14. 48行 num.io.threads=8
  15. #数值一定要大于磁盘数,处理磁盘读写的线程数
  16. 52行 socket.send.buffer.bytes=102400
  17. #发送套接字缓冲区的大小
  18. 56行 socket.receive.buffer.bytes=102400
  19. #接收套接字缓冲区的大小
  20. 60行 socket.request.buffer.bytes=102400
  21. #请求套接字缓冲区的大小
  22. 67行 log.dirs=/usr/local/kafka/logs
  23. #kafka日志存放的路径
  24. 72行 num.partitions=1
  25. #创建主题时经纪人指定的分区数,如果指定的分区数不同,这个值可以被覆盖
  26. 111行 log.retention.hours=168
  27. #消息在本地持久化保留的时间是168小时
  28. 118行 log.segent.bytes=1073741824
  29. #保存的持久化的文件的大小,超过这个值的也会被删除
  30. 132 zookeeper.connect=192.168.230.21:2181,192.168.230.22:2181,192.168.230.23:2181
  31. #指定zookeeper集群的ip地址
  32. #声明变量
  33. vim /etc/profile
  34. export KAFKA_HOME=/usr/local/kafka
  35. export PATH=$PATH:$KAFKA_HOME/bin
  36. #立即生效
  37. source /etc/profile
  38. #主机名映射
  39. vim /etc/hosts
  40. 192.168.230.21 mysql1
  41. 192.168.230.22 mysql2
  42. 192.168.230.23 mysql3
  43. #配置kafka启动脚本
  44. vim /etc/init.d/kafka
  45. #设置开机自启
  46. chmod +x /etc/init.d/kafka
  47. chkconfig --add kafka
  48. #启动kafka
  49. service kafka start
  50. #查看端口是否开启
  51. netstat -antp | grep 9092
  52. #创建主题
  53. kafka-topics.sh --create --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092 --replication-factor 2 --partitions 3 --topic test1
  54. --replication-factor 2:创建分区的副本数
  55. --partitions:分区数
  56. --topic:指定分区的名称
  57. #查看topic数
  58. kafka-topics.sh --list --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092
  59. #模拟发布消息
  60. kafka-console-producer.sh --broker-list 192.168.230.21;9092,192.168.230.22:9092,192.168.230.23:9092 --topic test1
  61. test1
  62. test2
  63. test3
  64. #消费者订阅主题
  65. kafka-console-consumer.sh --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092 --topic test1 --from-beginning
  66. #查看主题的详细信息
  67. kafka-topics.sh --describe --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092
  68. Topic: test1        TopicId: 3cMz2ZiYSr2rxy9jzGCyeQ        PartitionCount: 3        ReplicationFactor: 2        Configs:
  69.         Topic: test1        Partition: 0        Leader: 2        Replicas: 2,1        Isr: 2,1
  70.         Topic: test1        Partition: 1        Leader: 1        Replicas: 1,0        Isr: 1,0
  71.         Topic: test1        Partition: 2        Leader: 0        Replicas: 0,2        Isr: 0,2
  72. Topic:主题名称
  73. Partition:分区,后面的0 1 2是偏移量
  74. Leader:分区的领导者用来处理分区的读写操作,只有在指定写分区和指定读分区的时候才有工作,如果不是指定,全量展示则无意义
  75. Replicas:分区的副本,0 1 2对应的是broker id
  76. Isr:表示当前与领导者同步的副本
  77. -----------------------------------------------------------------------------------------------------------------------------------------------------------
  78. #删除主题
  79. kafka-topics.sh --delete --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092 --topic test1
复制代码
2.8kafka的消息堆积该怎样办理
  1. 产生的原因:
  2. 消费者出现了延迟或者处理能力太差,导致消息堆积。
  3. 方法:
  4. 1、减少kafka持久化的保存时间
  5. 2、修改主题的分区数,扩大分区的数量,提高消费者获取的通道
  6. 修改分区数:kafka-topics.sh --bootstrap-server 192.168.230.21:9092,192.168.230.22:9092,192.168.230.23:9092 --alter --topic test1 --partitions 6
  7. 3、可以指定多个消费者共同工作,处理消息的挤压
复制代码
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!qidao123.com:ToB企服之家,中国第一个企服评测及软件市场,开放入驻,技术点评得现金
回复

使用道具 举报

登录后关闭弹窗

登录参与点评抽奖  加入IT实名职场社区
去登录
快速回复 返回顶部 返回列表