网站备案加链接代码,登陆wordpress忘记密码,网站建设需要php吗,中国建设银行手机银行家网站文章目录 1.1.网络和io操作线程配置优化1.2.log数据文件刷盘策略1.3.日志保留策略配置1.4.replica复制配置1.5.配置jmx服务1.6.系统I/O参数优化1.6.1.网络性能优化1.6.2.常见痛点以及优化方案1.6.4.优化参数 1.7.版本升级1.8.数据迁移1.8.1.同集群broker之间迁移1.8.2.跨集群迁… 文章目录 1.1.网络和io操作线程配置优化1.2.log数据文件刷盘策略1.3.日志保留策略配置1.4.replica复制配置1.5.配置jmx服务1.6.系统I/O参数优化1.6.1.网络性能优化1.6.2.常见痛点以及优化方案1.6.4.优化参数 1.7.版本升级1.8.数据迁移1.8.1.同集群broker之间迁移1.8.2.跨集群迁移 1.9.深度巡检1.9.1.集群状态检查1.9.2.查看消费者组1.9.3.查看消费偏移量 Kafka配置优化其实都是修改server.properties文件中参数值
1.1.网络和io操作线程配置优化
num.network.threadsxxx # broker处理消息的最大线程 num.io.threadsxxx # broker处理磁盘IO的线程数
建议配置 一般num.network.threads主要处理网络io读写缓冲区数据基本没有io等待配置线程数量为cpu核数加1. num.io.threads主要进行磁盘io操作高峰期可能有些io等待因此配置需要大些。配置线程数量为cpu核数2倍最大不超过3倍.
1.2.log数据文件刷盘策略
为了大幅度提高producer写入吞吐量需要定期批量写文件。
建议配置 每当producer写入10000条消息时刷数据到磁盘 log.flush.interval.messages10000 每间隔1秒钟时间刷数据到磁盘 log.flush.interval.ms1000
1.3.日志保留策略配置
当kafka server的被写入海量消息后会生成很多数据文件且占用大量磁盘空间如果不及时清理可能磁盘空间不够用kafka默认是保留7天。
建议配置 保留三天也可以更短 log.retention.hours72 段文件配置1GB有利于快速回收磁盘空间重启kafka加载也会加快(如果文件过小则文件数量比较多 kafka启动时是单线程扫描目录(log.dir)下所有数据文件) log.segment.bytes1073741824
1.4.replica复制配置
每个follow从leader拉取消息进行同步数据follow同步性能由这几个参数决定分别为拉取线程数(num.replica.fetchers)、最小字节数(replica.fetch.min.bytes)、最大字节数(replica.fetch.max.bytes)、最大等待时间(replica.fetch.wait.max.ms)
建议配置 num.replica.fetchers 配置多可以提高follower的I/O并发度单位时间内leader持有跟多请求相应负载会增大需要根据机器硬件资源做权衡 replica.fetch.min.bytes1 默认配置为1字节否则读取消息不及时 replica.fetch.max.bytes 5 * 1024 * 1024 默认为1MB这个值太小5MB为宜根据业务情况调整 replica.fetch.wait.max.ms follow拉取频率频率过高会导致cpu飙升因为leader无数据同步leader会积压大量无效请求情况又因为0.8.2.x版本存在bug定时器超时检查比较消耗CPU使用者需要做好权衡
1.5.配置jmx服务
kafka server中默认是不启动jmx端口的需要用户自己配置
1.6.系统I/O参数优化
vm.dirty_background_ratio是内存可以填充脏页的百分比当脏页总大小达到这个比例后系统后台进程就会开始将脏页刷磁盘vm.dirty_background_bytes类似只不过是通过字节数来设置 vm.dirty_ratio是绝对的脏数据限制内存里的脏数据百分比不能超过这个值。如果脏数据超过这个数量新的IO请求将会被阻挡直到脏数据被写进磁盘 vm.dirty_writeback_centisecs指定多长时间做一次脏数据写回操作单位为百分之一秒 vm.dirty_expire_centisecs指定脏数据能存活的时间单位为百分之一秒比如这里设置为30秒在操作系统进行写回操作时如果脏数据在内存中超过30秒时就会被写回磁盘
参考配置如下
vm.dirty_background_ratio 10
vm.dirty_background_bytes 0
vm.dirty_ratio 20
vm.dirty_bytes 0
vm.dirty_writeback_centisecs 500
vm.dirty_expire_centisecs 30001.6.1.网络性能优化
参考配置 net.core.optmem_max 增大每个套接字的缓冲区大小 参考设置81920 net.core.rmem_max 增大套接字接收缓冲区大小 参考设置513920 net.core.wmem_max 发送缓冲区大小 参考设置513920 net.ipv4.tcp_rmem 增大 TCP 接收缓冲区大小 参考设置4096 87380 16777216 net.ipv4.tcp_wmem 发送缓冲区大小 参考设置4096 65535 16777216 net.ipv4.udp_mem 增大UDP缓冲区范围 参考设置188562 251418 377124 net.ipv4.tcp_max_tw_buckets 增大TIME_WAIT 状态的连接数量 参考设置 1048576 net.netfilter.nf_conntrack_max 连接跟踪表大小 参考设置 1048576 net.ipv4.tcp_fin_timeout 减少连接超时时间 参考设置 15 net.netfilter.nf_conntrack_tcp_timeout_time_wait 缩短连接跟踪表中处于TIME_WAIT状态连接的超时时间 参考设置 30 net.ipv4.tcp_tw_reuse 开启端口复用 net.ipv4.ip_local_port_range 增大本地端口范围 参考设置 10000 65000 fs.nr_open 设置最大文件描述符数 参考设置 1048576 net.ipv4.tcp_max_syn_backlog 增加半连接数最大数量 参考设置 16384 net.ipv4.tcp_syncookies 开启SYN Cookies 参考设置 1 net.ipv4.ip_forward 开启ip转发 参考设置 1
说明 tcp_rmem 和 tcp_wmem 的三个数值分别是 mindefaultmax系统会根据这些设置自动调整 TCP 接收 / 发送缓冲区的大小 udp_mem 的三个数值分别是 minpressuremax系统会根据这些设置自动调整 UDP 发送缓冲区的大小
1.6.2.常见痛点以及优化方案
1.集群⽊桶效应broker雪崩 痛点 当整个集群当leader和follower分布不均衡时这可能导致流量分布不均衡。⼀部分节点⽐较空闲⼀部分节点负载过⾼这⾥当负载主要是磁盘IO与⽹络带宽CPU基本上不会成为Kafka的瓶颈。最后导致出现⼤量副本缺失直⾄broker挂掉后流量压⼒转移到另外⼀个broker节点很可能这个节点也会因为负载过⼤⽽被打挂。
优化⽅案 1 实现⼀套⾃动负载均衡程序⾃动⽣成均衡计划逐个topic进⾏均衡。【推荐】 2⼿动切换分区leader或者进⾏副本迁移【效率低不推荐】
2.集群扩容⽆法⾃动负载均衡 痛点 当扩容集群时topic的分区⽣产消费⽆法落在新加⼊的broker节点上。这样相当于已经存在的topic读写数据⽆法落在新broker节点上从⽽新节点⽆法得到充分利⽤。 优化⽅案 1实现⼀套⾃动负载均衡程序⾃动⽣成均衡计划逐个topic进⾏均衡。【推荐】 2⼿动切换分区leader或者进⾏副本迁移【效率低不推荐】
3.集群副本迁移影响集群稳定迁移任务不可控 痛点 当需要迁移的分区数据量⽐较⼤时单分区数据量超过100GB以上这将导致迁移任务启动的新副本会从leader所在的broker节点⼤量拉取历史数据。这可能带来以下问题 1broker的磁盘IO被持续打满 2操作系统pagecache受到污染导致⽣产消费延迟 3出现⼤量副本缺失 4迁移任务⼀旦开启便⽆法停⽌只有让其执⾏失败或者成果才能结束。
优化⽅案 1改造源码从最新偏移量开始同步数据实现增量副本迁移 2增量迁移新副本加⼊isr列表的时机可以根据当前分区是否有消费延迟和指定同步时间两个⽅⾯去考虑 3修改源码对迁移任务添加可⼿动终⽌功能
4.异常流量打挂集群 痛点 当出现⼊流量当突增或出流量当突增时可能造成broker节点负载过⼤经常时磁盘IO被持续打满到100%。以下情况容易造成流量突增 1新业务上线数据源加⼊了新的⼤业务⽣产者写⼊流量突增 2消费端程序从历史最早位置开始消费拉去⼤量历史数据 3消费端程序停⽌服务⼀段时间后重启追数 4⼤topic在消费端程序有新的消费者组加⼊出流量突增 1根据⽤户维度对各集群⽤户的出⼊流量进⾏限制保证单个broker节点上所有⽤户的出⼊流量之和在broker的处理能⼒范围内
5.⼀个业务异常影响整个集群稳定 痛点 当某个业务的部分topic异常时可能会影响到集群上的其他业务。 优化⽅案 根据不同业务线以资源组为单位对集群进⾏物理隔离让各业务线的topic都是分布在⾃⼰的资源组内。不受其他业务线影响
6.pagecache污染及优化 痛点 1⼤量拉去历史数据导致pagecache污染造成⽣产消费延迟
优化⽅案 1对pagecache参数进⾏调优⽂章地址 2修改源码对kafka的cache进⾏改造。⾃定义⼀套cache专门⽤来做消费cache。保证副本同步和拉取历史数据不会污染最近⽣产的数据。
7.磁盘故障或者坏道整个broker半死不活 痛点 1当整块磁盘故障或出现磁盘坏道情况整个broker部分分区可以读写部分分区⽆法读写broker进程不会主动推出坏道所影响的分区消费⼀直被卡住同时造成数据丢失。
优化⽅案 1当整块磁盘故障或者坏道时消费者消费会在服务端抛出⼀些特定关键字符串当异常信息⽤程序扫描异常信息检测到后把对应的磁盘从 log.dirs 中剔除
8.同⼀个消费者组消费多个topic问题 痛点 1当同⼀个消费者组消费多个topic⽆法回收消费者组下某个topic的读权限只能对整个组下⾯的topic进⾏读权限回收 2消费组内多个topic间经常出现join操作导致topic分区重新分配影响整个组下⾯topic消费
优化⽅案 1限定⼀个组只能消费⼀个topic 1.6.3.ack机制 ack不同设置的区别
acks1, 默认值1表示只要分区的leader副本成功写入就算成功acks0生产者不需要等待任何服务端的响应可能会丢失数据acks-1或acksall需要全部处于同步状态的副本确认写入成功可靠性最强性能也差
不同的ack机制可能产生的问题
ack为-1时吞吐量吞吐最低数据最安全可能发生重复ack为1时吞吐量安全性最均衡ack为0时吞吐最高数据安全性最低
ack为-1的重复问题
1.6.4.优化参数
1Broker参数配置server.properties 1、网络和io操作线程配置优化 broker处理消息的最大线程数默认为3 num.network.threadscpu核数1 broker处理磁盘IO的线程数 num.io.threadscpu核数*2
2、log数据文件刷盘策略 每当producer写入10000条消息时刷数据到磁盘 log.flush.interval.messages10000 每间隔1秒钟时间刷数据到磁盘 log.flush.interval.ms1000
3、日志保留策略配置 保留三天也可以对个别主题单独设置 log.cleaner.delete.retention.ms log.retention.hours72
4、Replica相关配置 offsets.topic.replication.factor:3 这个参数指新创建一个topic时默认的Replica数量,Replica过少会影响数据的可用性太多则会白白浪费存储资源一般建议在2~3为宜
2Producer优化producer.properties buffer.memory:33554432 (32m) 在Producer端用来存放尚未发送出去的Message的缓冲区大小。缓冲区满了之后可以选择阻塞发送或抛出异常由block.on.buffer.full的配置来决定。
compression.type:none 默认发送不进行压缩推荐配置一种适合的压缩算法可以大幅度的减缓网络压力和Broker的存储压力。
3Consumer优化 num.consumer.fetchers:1 启动Consumer的个数适当增加可以提高并发度。
fetch.min.bytes:1 每次Fetch Request至少要拿到多少字节的数据才可以返回。
fetch.wait.max.ms:100 在Fetch Request获取的数据至少达到fetch.min.bytes之前允许等待的最大时长。对应上面说到的Purgatory中请求的超时时间。
4Kafka内存调整kafka-server-start.sh 默认内存1个G生产环境尽量不要超过6个G。 export KAFKA_HEAP_OPTS“-Xms4g -Xmx4g”
1.7.版本升级
FromToAction0.8.x~a.b.xa.b n.m n.m.01. 编辑server.properties(1)0.11.0:inter.broker.protocol.version…log.messages.format.version…(2) 0.11.0:inter.broker.protocol.version…2. 停止服务, 更新代码, 重启服务3. 升级完毕, 重新编辑server.properties4. 改为n.m5. 重启服务0.8.x~0.11.x1.0.01. 编辑server.properties(1) 0.11.0:inter.broker.protocol.version…log.messages.format.version…(2) 0.11.0:inter.broker.protocol.version0.11.0log.messages.format.version0.11.02. 停止服务, 更新代码, 重启服务3. 升级完毕, 重新编辑server.properties inter.broker.protocol.version1.04. 重启服务0.8.x~0.10.x0.11.0.01. 编辑server.propertiesinter.broker.protocol.version…log.messages.format.version…2. 停止服务, 更新代码, 重启服务3. 升级完毕, 重新编辑server.propertiesinter.broker.protocol.version0.11.04. 重启服务0.8.x.x0.9.0.01. 编辑server.properties增加行:inter.broker.protocol.version0.8.2.X2.停止服务, 更新代码, 重启服务3.升级完毕, 重新编辑server.properties0.8.2.x改为0.9.0.04.重启服务0.8.10.8.2完全兼容0.8.00.8.1完全兼容0.70.8.0不兼容
1.8.数据迁移
Kafka两种迁移场景分别是同集群数据迁移、跨集群数据迁移。
1.8.1.同集群broker之间迁移 1.8.1.1.应用场景 broker 迁移 主要使用的场景是broker 上线,下线,或者扩容等.基于同一套zookeeper的操作. 实践 将需要新添加的broker 列表一并添加到kafka的集群中。启动新的kafka指定同一套zk Kafka由之前的三节点扩容至四节点 1.8.1.2.数据迁移 查询信息 四个分区分布在三台机器上
新建json文件
cat topic-to-move.json
{topics: [{topic: test-topic}],version:1}获取重新分配方案 kafka-reassign-partitions.sh --zookeeper node1:2181,node2:2181,node3:2181 --topics-to-move-json-file topics-to-move.json --broker-list “150,151,155,159” –generate
通过kafka-reassign-partitions.sh 获取重新分配方案,–broker-lsit 的参数 “150,151,155,159是指集群中每个broker的id由于我们是需要将所有topic均匀分配到扩完结点的4台机器上所以要指定。同理当业务改变为将原来的所有数据从旧节点0,5,9迁移到新节点1实现数据平滑迁移这时的参数应4” ,执行后会出现以下内容: 复制新的方案到一个json文件 assignplan.json 文件名不重要文件格式也不一定要以json为 结尾只要保证内容是json即可 ./kafka-reassign-partitions.sh --zookeeper node1:2181 --reassignment-json-file assignplan.json --execute完成后查看topic信息四个分区分布在四台机器上
1.8.2.跨集群迁移 1.8.2.1.应用场景 主要用于kafka 集群的变更. 将数据同步等操作. 相当于是实现了双打.等各项数据消费端在零误差的对接好了后,可以停掉就集群
1.8.2.2.数据迁移
方案一MirrorMaker 修改kafka配置 consumer.properties内容为原始集群信息
#config/consumer.properties 在网上看到有在此配置zookeeper的应该是之前的老版本。kafka_2.11-2.4.1中不需要
bootstrap.serverskafka-cluster1:9092,kafka-cluster1:9093 # source-cluster的broker list
group.idtest-consumer-group1 # 自定义一个消费者的group id
auto.offset.reset # latest当各分区下有已提交的offset时从提交的offset开始消费无提交的offset时消费新产生的该分区下的数据 earliest当各分区下有已提交的offset时从提交的offset开始消费无提交的offset时从头开始消费 nonetopic各分区都存在已提交的offset时从offset后开始消费只要有一个分区不存在已提交的offset则抛出异常producer.properties内容为新集群信息
#config/producer.properties 在网上看到有在此配置zookeeper的应该是之前的老版本。kafka_2.11-2.4.1中不需要
bootstrap.serverskafka-cluster2:9092,kafka-cluster2:9093 # destination-cluster的broker list
compression.typenone # 数据压缩方式none, gzip, snappy, lz4, zstd
partitioner.class # 指定分区程序路径默认为随机分区
request.timeout.ms # 请求超时时间
max.block.ms # KafkaProducer.send and KafkaProducer.partitionsFor 阻塞时间
linger.ms # 等待指定时间后批量发送
max.request.size # 发送消息最大字节数
batch.size # 单次批量处理的字节数
buffer.memory # 指定等待发送消息的缓冲区大小执行操作
./kafka-mirror-maker.sh --consumer.config ../config/consumer.properties --producer.config ../config/producer.properties --whitelist test说明 1、–num.streams: 指定流就是指定消费者所有消费者公用一个生产者。 2、–whitelist: 表明需要同步的白名单可以使用”|”来连接多个topic还可以使用正则表达式。可设置黑名单。
方案二zk迁移
zk迁移就比较简单了起新节点加入zk集群稳定后关停旧节点。新增broker加入集群将所有topic分区只分配给新broker执行分配任务后kafka将旧broker的分区数据复制到新broker新broker成为各分区的leader随后kafka删除旧broker上的分区数据整个过程中客户端应用正常生产消费消息执行结束后使用新的消费者组从头消费可以获取到全部历史消息。停止旧broker后正在运行的客户端应用正常生产消费消息新建客户端连接旧broker失败连接新broker正常
1.9.深度巡检
1.9.1.集群状态检查
1.9.1.1.查看kafka节点的集群角色 使用kafka自带的zookeeper shell工具查看
[rootdb3 bin]# ./zookeeper-shell.sh db3:2184,db3:2185,db3:2186
get /controller
{version:1,brokerid:0,timestamp:1525847148171}
#当前kafka集群的中央控制器为broker 0
quit
#退出zookeeper shell还可以使用zookeeper的zkCli工具查看
./zkCli.sh -server localhost:2184,localhost:2185,localhost:2186
get /controller
{version:1,brokerid:0,timestamp:1531967020325}
cZxid 0x1200000046
ctime Thu Jul 19 10:23:40 CST 2018
mZxid 0x1200000046
mtime Thu Jul 19 10:23:40 CST 2018
pZxid 0x1200000046
cversion 0
dataVersion 0
aclVersion 0
ephemeralOwner 0x30355dd23ee0000
dataLength 54
numChildren 0
#当前kafka集群的中央控制器为broker 0
quit
#退出zkClizookeeper事务日志中各个字段的含义 cZxid创建节点的事务id ctime创建节点的时间 mZxid修改节点的事务id mtime修改节点的时间 pZxid子节点列表最后一次修改的事务id。删除或添加子节点不包含修改子节点的数据 cversion子节点的版本号删除或添加子节点版本号会自增 dataVersion节点数据版本号数据写入操作版本号会递增 aclVersion节点ACL权限版本权限写入操作版本号会递增 ephemeralOwner临时节点创建时的事务id如果节点是永久节点则它的值为0 dataLength节点数据长度单位byte中文占3个byte numChildren子节点数量
1.9.1.2.查看kafka集群中所有topic
./kafka-topics.sh --zookeeper db3:2184,db3:2185,db3:2186 --list –force1.9.1.3.查看所有topic的分区和副本信息
./kafka-topics.sh --zookeeper db3:2184,db3:2185,db3:2186 --describe –force1.9.1.4.查看test的topic的分区和副本信息
./kafka-topics.sh --zookeeper db3:2184,db3:2185,db3:2186 --describe --force --topic test返回信息示例test队列有3个分区分别是0、1、2对应broker0、broker1、broker2每个分区都有一个broker作为读写分区的leader节点 Topic:test PartitionCount:3 ReplicationFactor:3 Configs: Topic: test Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2 Topic: test Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 0,1,2 Topic: test Partition: 2 Leader: 2 Replicas: 2,0,1 Isr: 0,1,2
1.9.1.5.从消费者查看test的topic中的所有消息
kafka-console-consumer.sh --bootstrap-server db3:9092,db3:9093,db3:9094 --from-beginning --topic test
#将输出test队列中的所有消息到控制台退出请按CtrlC1.9.1.6.查看kafka的LogSegment 一个Partition日志又被划分为多个日志段LogSegment日志段是Kafka日志对象分片的最小单位。一个日志段对应磁盘上一个“.log”的日志文件一个“.index”的消息偏移量索引文件和一个“.timeindex”的消息时间戳索引文件。
查看指定的kafka节点的syslog队列的分区0的日志段
kafka-run-class.sh kafka.tools.DumpLogSegments --files /opt/kafka_2.12-1.1.0/logs/kafka-logs/syslog-0/00000000000006239341.log --deep-iteration --print-data-log
截取部分数据日志
offset: 6239505 position: 45001 CreateTime: -1 isvalid: true keysize: -1 valuesize: 240 magic: 2 compresscodec: GZIP producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {timestamp:2018-05-11T06:01:10.654Z,beat:{name:proxy1},input_type:log,message:May 11 14:01:01 proxy1 systemd: Starting Session 1066 of user root.,offset:21415,source:/var/log/messages,tags:[syslog],type:log}1.9.2.查看消费者组
消费者组类型有zookeeper和kafka两种基于kafka类型消费者组是通过kafka中名为__consumer_offsets的topic来记录每个消费者组的OffsetMetadata信息。 相同消费者组中的不同消费者可以对topic中的消息进行负载均衡的消费单播不同的消费者组可以同时对队列中的消息进行消费多播。
1.9.2.1.查看基于zookeeper管理的消费者组 kafka-consumer-groups.sh --zookeeper db3:2184,db3:2185,db3:2186 --list 1.9.2.2.查看基于kafka管理的消费者组 kafka-consumer-groups.sh --bootstrap-server db3:9092,db3:9093,db3:9094 –list 1.9.2.3.查看消费者组的详细信息 消费者组的客户端ID、消费者ID、主机、队列、分区
./kafka-consumer-groups.sh --bootstrap-server db3:9092,db3:9093,db3:9094 --group logstash --describe –verbose1.9.3.查看消费偏移量
1.9.3.1.查看指定消费者组的所有topic的partition的消费offset kafka-consumer-groups.sh --bootstrap-server db3:9092,db3:9093,db3:9094 --describe --group logstash –offsets 1.9.3.2.查看指定topic的消费偏移量 kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list db3:9092,db3:9093,db3:9094 --topic test --time -1 1.9.3.3.查看__consumer_offsets的消息
查看__consumer_offsets主题的分区6的消息
kafka-simple-consumer-shell.sh --topic __consumer_offsets --broker-list db3:9092,db3:9093,db3:9094 --partition 6 --formatter kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter截取部分OffsetMetadata消息
[console-consumer-15932,heartbeat,0]::[OffsetMetadata[1274,NO_METADATA],CommitTime 1524656104986,ExpirationTime 1524742504986]
[console-consumer-94057,__consumer_offsets,26]::[OffsetMetadata[9,NO_METADATA],CommitTime 1525859391291,ExpirationTime 1525945791291]
[logstash,syslog,2]::[OffsetMetadata[5,NO_METADATA],CommitTime 1532407710812,ExpirationTime 1532494110812]
[logstash,syslog,0]::[OffsetMetadata[4,NO_METADATA],CommitTime 1532407710827,ExpirationTime 1532494110827]
