做网站设计累吗,广告发布服务属于什么服务,藤县建设局网站,h5电子商城网站开发目录 一、 Kafka介绍1、kafka定义2、传统消息队列应用场景3、kafka特点和优势4、kafka角色介绍5、分区和副本的优势6、kafka 写入消息的流程 二、Kafka单机部署1、基础环境2、iptables -L -n配置3、下载并解压kafka部署包至/usr/local/目录4、修改server.properties5、修改/etc… 目录 一、 Kafka介绍1、kafka定义2、传统消息队列应用场景3、kafka特点和优势4、kafka角色介绍5、分区和副本的优势6、kafka 写入消息的流程 二、Kafka单机部署1、基础环境2、iptables -L -n配置3、下载并解压kafka部署包至/usr/local/目录4、修改server.properties5、修改/etc/profile6、执行/etc/profile7、启动kafka 三、Kafka集群部署1、基础环境2、集群3、iptables -L -n配置4、下载并解压kafka部署包至/usr/local/目录cnode1执行5、修改/etc/profilecnode1执行6、执行/etc/profilecnode1执行7、创建存储目录cnode1执行8、克隆出cnode2和cnode39、修改zookeeper配置文件所有节点执行10、cnode1新增myid配置cnode1执行11、cnode2新增myid配置cnode2执行12、cnode2新增myid配置cnode3执行13、修改cnode1节点kafka配置14、修改cnode2节点kafka配置15、修改cnode3节点kafka配置16、启动zookeeper所有节点执行17、启动kafka所有节点执行18、也可以用脚本启动/停止所有节点执行 四、Kafka基础命令一、KAFKA启停命令二、Topic 相关命令三、消息相关命令四、消费者Group五、其他 一、 Kafka介绍
1、kafka定义
Kafka是由Apache软件基金会开发的一个开源流处理平台由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统它可以处理消费者在网站中的所有动作流数据。 这种动作网页浏览搜索和其他用户的行动是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统但又要求实时处理的限制这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理也是为了通过集群来提供实时的消息。
2、传统消息队列应用场景
削峰填谷 诸如电商业务中的秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲或因没做相应的保护而导致系统超负荷甚至崩溃或因限制太过导致请求大量失败而影响用户体验消息队列可提供削峰填谷的服务来解决该问题。
异步解耦 交易系统作为淘宝等电商的最核心的系统每笔交易订单数据的产生会引起几百个下游业务系统的关注包括物流、购物车、积分、流计算分析等等整体业务系统庞大而且复杂消息队列可实现异步通信和应用解耦确保主站业务的连续性。
顺序收发 细数日常中需要保证顺序的应用场景非常多例如证券交易过程时间优先原则交易系统中的订单创建、支付、退款等流程航班中的旅客登机消息处理等等。与先进先出FIFOFirst In First Out原理类似消息队列提供的顺序消息即保证消息FIFO。
分布式事务一致性 交易系统、支付红包等场景需要确保数据的最终一致性大量引入消息队列的分布式事务既可以实现系统之间的解耦又可以保证最终的数据一致性。
分布式缓存同步 电商的大促各个分会场琳琅满目的商品需要实时感知价格变化大量并发访问数据库导致会场页面响应时间长集中式缓存因带宽瓶颈限制了商品变更的访问流量通过消息队列构建分布式缓存实时通知商品数据的变化。
3、kafka特点和优势
特点
分布式: 多机实现,不允许单机分区: 一个消息.可以拆分出多个分别存储在多个位置多副本: 防止信息丢失可以多来几个备份多订阅者: 可以有很多应用连接kafkaZookeeper: 早期版本的Kafka依赖于zookeeper 2021年4月19日Kafka 2.8.0正式发布此版本包括了很多重要改动最主要的是kafka通过自我管理的仲裁来替代ZooKeeper即Kafka将不再需要ZooKeeper
优势
高吞吐量即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。支持通过Kafka 服务器分区消息。分布式 Kafka 基于分布式集群实现高可用的容错机制可以实现自动的故障转移。顺序保证在大多数使用场景下数据处理的顺序都很重要。大部分消息队列本来就是排序的并且能保证数据会按照特定的顺序来处理。 Kafka保证一个Partiton内的消息的有序性分区间数据是无序的如果对数据的顺序有要求应将在创建主题时将分区数partitions设置为1。支持 Hadoop 并行数据加载。通常用于大数据场合,传递单条消息比较大而Rabbitmq 消息主要是传输业务的指令数据,单条数据较小。
4、kafka角色介绍
1Producer Producer即生产者消息的产生者是消息的入口。负责发布消息到Kafka broker 2Consumer 消费者用于消费消息即处理消息 BrokerBroker是kafka实例每个服务器上可以有一个或多个kafka的实例假设每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号如: broker-0、broker-1等…… 3Topic 消息的主题可以理解为消息的分类一个Topic相当于数据库中的一张表,一条消息相当于关系数据库的一条记录一个Topic或者相当于Redis中列表类型的一个Key一条消息即为列表中的一个元素。kafka的数据就保存在topic。在每个broker上都可以创建多个topic。物理上不同 topic 的消息分开存储在不同的文件夹逻辑上一个 topic的消息虽然保存于一个或多个broker 上, 但用户只需指定消息的topic即可生产或消费数据而不必关心数据存于何处topic 在逻辑上对record(记录、日志)进行分组保存消费者需要订阅相应的topic 才能消费topic中的消息 4Consumer group: 每个consumer 属于一个特定的consumer group可为每个consumer 指定 group name若不指定 group name 则属于默认的group同一topic的一条消息只能被同一个consumer group 内的一个consumer 消费类似于一对一的单播机制但多个consumer group 可同时消费这一消息类似于一对多的多播机制 5Partition 是物理上的概念每个topic 分割为一个或多个partition即一个topic切分为多份.创建 topic时可指定 partition 数量partition的表现形式就是一个一个的文件夹,该文件夹下存储该partition的数据和索引文件分区的作用还可以实现负载均衡提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的,一般Partition数不要超过节点数注意同一个partition数据是有顺序的但不同的partition则是无序的 6Replication: 同样数据的副本包括leader和follower的副本数,基本于数据安全,建议至少2个,是Kafka的高可靠性的保障和ES的副本有所不同Kafka中的副本数包括主分片数,而ES中的副本数不包括主分片数
为了实现数据的高可用比如将分区 0 的数据分散到不同的kafka 节点每一个分区都有一个 broker 作为 Leader 和一个 broker 作为Follower类似于ES中的主分片和副本分片。
假设分区为 3, 即分三个分区0-2副本为3即每个分区都有一个 leader再加两个follower分区 0 的leader为服务器A则服务器 B 和服务器 C 为 A 的follower而分区 1 的leader为服务器B则服务器 A 和C 为服务器B 的follower而分区 2 的leader 为C则服务器A 和 B 为C 的follower。
AR Assigned Replicas分区中的所有副本的统称包括leader和 followerAR lSR OSR lSRln Sync Replicas所有与leader副本保持同步的副本 follower和leader本身组成的集合包括leader和 follower是AR的子集 OSRout-of-Sync Replied所有与leader副本同步不能同步的 follower的集合是AR的子集
5、分区和副本的优势
实现存储空间的横向扩容即将多个kafka服务器的空间组合利用提升性能多服务器并行读写实现高可用每个分区都有一个主分区即 leader 分布在不同的kafka 服务器并且有对应follower 分布在和leader不同的服务器上
6、kafka 写入消息的流程
生产者producter先从kafka集群获取分区的leader生产者producter将消息发送给leaderleader将消息写入本地文件followers从leader pull消息followers将消息写入本地后向leader发送ACKleader收到所有副本的ACK后向producter发送ACK
二、Kafka单机部署
1、基础环境
查看centos版本
[rootlocalhost config]# cat /etc/redhat-release
CentOS Linux release 7.9.2009 (Core)
查看java版本
[rootlocalhost config]# java -version
openjdk version 1.8.0_412
OpenJDK Runtime Environment (build 1.8.0_412-b08)
OpenJDK 64-Bit Server VM (build 25.412-b08, mixed mode)2、iptables -L -n配置
target prot opt source destination
ACCEPT all -- 0.0.0.0/0 0.0.0.0/0
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:22
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:80
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:443
ACCEPT icmp -- 0.0.0.0/0 0.0.0.0/0 icmptype 8
ACCEPT all -- 0.0.0.0/0 0.0.0.0/0 state RELATED,ESTABLISHED
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:3306
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:21
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9090
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9104
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:3000
ACCEPT icmp -- 0.0.0.0/0 0.0.0.0/0 icmptype 8
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9092
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:21813、下载并解压kafka部署包至/usr/local/目录
tar -zxvf kafka_2.12-3.1.1.tgz -C /usr/local/4、修改server.properties
vim /usr/local/kafka_2.12-3.1.1/config/server.properties修改以下内容
listenersPLAINTEXT://192.168.15.128:9092
advertised.listenersPLAINTEXT://192.168.15.128:9092
log.dirs/data/kafka/logs
zookeeper.connectlocalhost:2181local改成192.168.15.128会报错[2024-12-03 11:17:06,427] INFO [ZooKeeperClient Kafka server] Closing. (kafka.zookeeper.ZooKeeperClient)5、修改/etc/profile
vim /etc/profile新增
export KAFKA_HOME/usr/local/kafka_2.12-3.1.1
export PATH$KAFKA_HOME/bin:$PATH6、执行/etc/profile
source /etc/profile7、启动kafka
先启动zookeeper
/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-start.sh /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties查看是否启动
netstat -tuln | grep 2181再启动kafka
/usr/local/kafka_2.12-3.1.1/bin/kafka-server-start.sh /usr/local/kafka_2.12-3.1.1/config/server.properties查看是否启动
netstat -tuln | grep 9092
jps #有kafka则为启动后台启动
/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-start.sh -daemon /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-start.sh -daemon /usr/local/kafka_2.12-3.1.1/config/server.properties三、Kafka集群部署
1、基础环境
查看centos版本
[rootlocalhost config]# cat /etc/redhat-release
CentOS Linux release 7.9.2009 (Core)
查看java版本
[rootlocalhost config]# java -version
openjdk version 1.8.0_412
OpenJDK Runtime Environment (build 1.8.0_412-b08)
OpenJDK 64-Bit Server VM (build 25.412-b08, mixed mode)2、集群
IP主机名192.168.15.130cnode1192.168.15.131cnode2192.168.15.132cnode3
3、iptables -L -n配置
target prot opt source destination
ACCEPT all -- 0.0.0.0/0 0.0.0.0/0
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:22
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:80
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:443
ACCEPT icmp -- 0.0.0.0/0 0.0.0.0/0 icmptype 8
ACCEPT all -- 0.0.0.0/0 0.0.0.0/0 state RELATED,ESTABLISHED
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:3306
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:21
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9090
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9104
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:3000
ACCEPT icmp -- 0.0.0.0/0 0.0.0.0/0 icmptype 8
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:9092
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:2181
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:3888
ACCEPT tcp -- 0.0.0.0/0 0.0.0.0/0 tcp dpt:28884、下载并解压kafka部署包至/usr/local/目录cnode1执行
tar -zxvf kafka_2.12-3.1.1.tgz -C /usr/local/5、修改/etc/profilecnode1执行
vim /etc/profile新增
export KAFKA_HOME/usr/local/kafka_2.12-3.1.1
export PATH$KAFKA_HOME/bin:$PATH6、执行/etc/profilecnode1执行
source /etc/profile7、创建存储目录cnode1执行
mkdir -p /opt/software/kafka/log
mkdir -p /opt/software/kafka/zookeeper
mkdir -p /opt/software/kafka/zookeeper/log8、克隆出cnode2和cnode3
9、修改zookeeper配置文件所有节点执行
进入配置目录
cd /usr/local/kafka_2.12-3.1.1/config/
备份zookeeper配置文件
mv zookeeper.properties zookeeper.properties.bak
新建zookeeper配置文件
vi /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties
输入以下内容
dataDir/opt/software/kafka/zookeeper
dataLogDir/opt/software/kafka/zookeeper/log
clientPort2181
maxClientCnxns100
tickTimes2000
initLimit10
syncLimit5
server.0192.168.56.101:2888:3888
server.1192.168.56.103:2888:3888
server.2192.168.56.104:2888:388810、cnode1新增myid配置cnode1执行
vim/opt/software/kafka/zookeeper/myid
输入以下内容
011、cnode2新增myid配置cnode2执行
vim/opt/software/kafka/zookeeper/myid
输入以下内容
112、cnode2新增myid配置cnode3执行
vim/opt/software/kafka/zookeeper/myid
输入以下内容
213、修改cnode1节点kafka配置
进入配置目录
cd /usr/local/kafka_2.12-3.1.1/config/
备份kafka配置文件
mv server.properties server.properties.bak
新建server配置文件broker.id0
listenersPLAINTEXT://192.168.15.130:9092
advertised.listenersPLAINTEXT://192.168.15.130:9092
num.network.threads3
num.io.threads8
socket.send.buffer.bytes102400
socket.receive.buffer.bytes102400
socket.request.max.bytes104857600
log.dirs/tmp/kafka-logs
num.partitions1
num.recovery.threads.per.data.dir1
offsets.topic.replication.factor1
transaction.state.log.replication.factor1
transaction.state.log.min.isr1
log.retention.hours168
log.retention.check.interval.ms300000
zookeeper.connect192.168.15.130:2181,192.168.15.131:2181,192.168.15.132:2181
zookeeper.connection.timeout.ms18000
group.initial.rebalance.delay.ms014、修改cnode2节点kafka配置
进入配置目录
cd /usr/local/kafka_2.12-3.1.1/config/
备份kafka配置文件
mv server.properties server.properties.bak
新建server配置文件broker.id1
listenersPLAINTEXT://192.168.15.131:9092
advertised.listenersPLAINTEXT://192.168.15.131:9092
num.network.threads3
num.io.threads8
socket.send.buffer.bytes102400
socket.receive.buffer.bytes102400
socket.request.max.bytes104857600
log.dirs/tmp/kafka-logs
num.partitions1
num.recovery.threads.per.data.dir1
offsets.topic.replication.factor1
transaction.state.log.replication.factor1
transaction.state.log.min.isr1
log.retention.hours168
log.retention.check.interval.ms300000
zookeeper.connect192.168.15.130:2181,192.168.15.131:2181,192.168.15.132:2181
zookeeper.connection.timeout.ms18000
group.initial.rebalance.delay.ms015、修改cnode3节点kafka配置
进入配置目录
cd /usr/local/kafka_2.12-3.1.1/config/
备份kafka配置文件
mv server.properties server.properties.bak
新建server配置文件broker.id2
listenersPLAINTEXT://192.168.15.132:9092
advertised.listenersPLAINTEXT://192.168.15.132:9092
num.network.threads3
num.io.threads8
socket.send.buffer.bytes102400
socket.receive.buffer.bytes102400
socket.request.max.bytes104857600
log.dirs/tmp/kafka-logs
num.partitions1
num.recovery.threads.per.data.dir1
offsets.topic.replication.factor1
transaction.state.log.replication.factor1
transaction.state.log.min.isr1
log.retention.hours168
log.retention.check.interval.ms300000
zookeeper.connect192.168.15.130:2181,192.168.15.131:2181,192.168.15.132:2181
zookeeper.connection.timeout.ms18000
group.initial.rebalance.delay.ms016、启动zookeeper所有节点执行
/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-start.sh /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties查看是否启动
netstat -tuln | grep 218117、启动kafka所有节点执行
/usr/local/kafka_2.12-3.1.1/bin/kafka-server-start.sh /usr/local/kafka_2.12-3.1.1/config/server.properties查看是否启动
netstat -tuln | grep 909218、也可以用脚本启动/停止所有节点执行
启动脚本
[rootlocalhost config]# cat /usr/local/kafka_2.12-3.1.1/kafkaStart.sh
/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-start.sh /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties
sleep 10
/usr/local/kafka_2.12-3.1.1/bin/kafka-server-start.sh /usr/local/kafka_2.12-3.1.1/config/server.properties
停止脚本
[rootlocalhost config]# cat /usr/local/kafka_2.12-3.1.1/kafkaStop.sh
/usr/local/kafka_2.12-3.1.1/bin/zookeeper-server-stop.sh /usr/local/kafka_2.12-3.1.1/config/zookeeper.properties
sleep 10
/usr/local/kafka_2.12-3.1.1/bin/kafka-server-stop.sh /usr/local/kafka_2.12-3.1.1/config/server.properties
将脚本复制到cnode2和cnode3节点
scp kafkaStart.sh kafkaStop.sh rootcnode2:$PWD
scp kafkaStart.sh kafkaStop.sh rootcnode3:$PWD
执行以下脚本启动
sh /usr/local/kafka_2.12-3.1.1/kafkaStart.sh
执行以下脚本停止
sh /usr/local/kafka_2.12-3.1.1/kafkaStop.sh 四、Kafka基础命令
一、KAFKA启停命令
1. 前台启动(先启动zookeeper)
bin/zookeeper-server-start.sh config/zookeeper.propertiesbin/kafka-server-start.sh config/server.properties
2. 后台启动
bin/zookeeper-server-start.sh -daemon config/zookeeper.propertiesbin/kafka-server-start.sh -daemon config/server.properties
或者
nohup bin/kafka-server-start.sh config/server.properties
指定 JMX port 端口启动指定 jmx可以方便监控 Kafka 集群
JMX_PORT9991 /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
3. 停止命令
bin/zookeeper-server-stop.shbin/kafka-server-stop.sh二、Topic 相关命令
1.创建 Topic
--topic 指定 Topic 名
-–partitions 指定分区数
-–replication-factor 指定备份(副本)数注意指定副本因子的时候不能大于broker实例个数否则报错
bin/kafka-topics.sh --create --bootstrap-server 192.168.15.130:9092 --replication-factor 1 --partitions 1 --topic demo22.查询Topic列表
/usr/local/kafka_2.12-3.1.1/bin/kafka-topics.sh --bootstrap-server 192.168.15.130:9092 --list3.查询 Topic 详情
bin/kafka-topics.sh --describe --bootstrap-server 192.168.15.130:9092 --topic demo4. 增加 Topic 的 partition 数
bin/kafka-topics.sh --bootstrap-server 192.168.15.130:9092 --alter --topic demo --partitions 25. 查看 topic 指定分区 offset 的最大值或最小值
time 为 -1 时表示最大值为 -2 时表示最小值
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic test_kafka_topic --time -1 --broker-list 192.168.15.130:9092 --partitions 0 6.删除Topic
说明在${KAFKA_HOME}/config/server.properties中配置 delete.topic.enable 为 true这样才能生效删除指定的 topic主题
bin/kafka-topics.sh --bootstrap-server 192.168.15.130:9092 --delete --topic demo2三、消息相关命令
1.发送消息
bin/kafka-console-producer.sh --broker-list 192.168.15.130:9092 --topic demo
消费消息从头开始
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --from-beginning --topic demo
2.消费消息从尾开始
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --topic demo --offset latest
3.消费消息从尾开始指定分区
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --topic demo --offset latest --partition 0
4.消费消息指定分区指定偏移量
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --topic demo --partition 0 --offset 100
5.指定分组-消费消息
注意给客户端命名之后,如果之前有过消费那么–from-beginning就不会再从头消费了
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --from-beginning --topic demo --group t1
6.取指定个数
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --topic demo --offset latest --partition 0 --max-messages 1
7.指定分组从头开始消费消息应该会指定偏移量
bin/kafka-console-consumer.sh --bootstrap-server 192.168.15.130:9092 --topic demo -group t2 --from-beginning四、消费者Group
1.消费者Group列表
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.15.130:9092 --list
2.查看Group详情
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.15.130:9092 --group test_group --describe
3.删除 Group中Topic
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.15.130:9092 --group test_group --topic demo --delete
4.删除Group
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.15.130:9092 --group test_group --delete五、其他
1.平衡leader
bin/kafka-preferred-replica-election.sh --bootstrap-server 192.168.15.130:9092
2.自带压测工具
bin/kafka-producer-perf-test.sh --topic test --num-records 100 --record-size 1 --throughput 100 --producer-props bootstrap.servers192.168.15.130:9092
