淄川网站建设yx718,wordpress xampp,做网站排在前十名要多少钱,制作网页方法文章目录 搭建Kafka监控平台合理规划Kafka部署环境合理优化Kafka集群配置优化Kafka客户端使用方式合理保证消息安全消费者防止消息重复消费 生产环境常见问题分析消息零丢失方案消息积压如何处理如何保证消息顺序 搭建Kafka监控平台
官网地址 下载efak-web-3.0.2-bin.tar.gz安… 文章目录 搭建Kafka监控平台合理规划Kafka部署环境合理优化Kafka集群配置优化Kafka客户端使用方式合理保证消息安全消费者防止消息重复消费 生产环境常见问题分析消息零丢失方案消息积压如何处理如何保证消息顺序 搭建Kafka监控平台
官网地址 下载efak-web-3.0.2-bin.tar.gz安装包之后efak需要依赖JDK和数据库。数据库支持本地化的SQLLite以及集中式的MySQL。生产环境建议使用MySQL。
数据库不需要建表初始化EFAK在执行过程中会自己完成初始化。 将efak压缩包解压
[rootworker1 ~]# tar -zxvf efak-web-3.0.2-bin.tar.gz -C /app/kafka/eagle修改efak解压目录下的conf/system-config.properties。 这个文件中提供了完整的配置下面只列出需要修改的部分。
######################################
# multi zookeeper kafka cluster list
# Settings prefixed with kafka.eagle. will be deprecated, use efak. instead
######################################
# 指向Zookeeper地址
efak.zk.cluster.aliascluster1
cluster1.zk.listworker1:2181,worker2:2181,worker3:2181######################################
# zookeeper enable acl
######################################
# Zookeeper权限控制
cluster1.zk.acl.enablefalse
cluster1.zk.acl.schemadigest
#cluster1.zk.acl.usernametest
#cluster1.zk.acl.passwordtest123######################################
# kafka offset storage
######################################
# offset选择存在kafka中。
cluster1.efak.offset.storagekafka
#cluster2.efak.offset.storagezk######################################
# kafka mysql jdbc driver address
######################################
#指向自己的MySQL服务。库需要提前创建
efak.drivercom.mysql.cj.jdbc.Driver
efak.urljdbc:mysql://worker1:3306/ke?useUnicodetruecharacterEncodingUTF-8zeroDateTimeBehaviorconvertToNull
efak.usernameroot
efak.passwordroot配置EFAK的环境变量
[rootworker1 ~]# vi ~/.bash_profile
export KE_HOME/app/kafka/eagle/efak-web-3.0.2
PATH$PATH:#KE_HOME/bin:$HOME/.local/bin:$HOME/bin[rootworker1 ~]# source ~/.bash_profile启动EFAK
配置完成后先启动Zookeeper和Kafka服务然后调用EFAK的bin目录下的ke.sh脚本启动服务
[rootworker1 bin]$ ./ke.sh start
-- 日志很长看到以下内容表示服务启动成功
[2023-06-28 16:09:43] INFO: [Job done!]
Welcome to______ ______ ___ __ __/ ____/ / ____/ / | / //_// __/ / /_ / /| | / , / /___ / __/ / ___ | / /| |
/_____/ /_/ /_/ |_|/_/ |_|
( Eagle For Apache Kafka® )Version v3.0.2 -- Copyright 2016-2022
*******************************************************************
* EFAK Service has started success.
* Welcome, Now you can visit http://192.168.232.128:8048
* Account:admin ,Password:123456
*******************************************************************
* Usage ke.sh [start|status|stop|restart|stats] /Usage
* Usage https://www.kafka-eagle.org/ /Usage
*******************************************************************EFAK管理页面
接下来就可以访问EFAK的管理页面。http://192.168.232.128:8048。 默认的用户名是admin 密码是123456
关于EFAK更多的使用方式比如EFAK服务如何集群部署等可以参考官方文档。 合理规划Kafka部署环境
机械硬盘kafka使用机械硬盘即可可以不用给Kafka固态硬盘 内存修改 bin/kafka-start-server.sh启动脚本修改JVM启动参数中的内存默认只申请了1G内存。
[operworker1 bin]$ cat kafka-server-start.sh
......
if [ x$KAFKA_HEAP_OPTS x ]; thenexport KAFKA_HEAP_OPTS-Xmx1G -Xms1G
fi
......对于主流的16核32G服务器可以适当扩大Kafka的内存。例如
export KAFKA_HEAP_OPTS‐Xmx16G ‐Xms16G ‐Xmn10G ‐XX:MetaspaceSize256M ‐XX:UseG1GC ‐XX:MaxGCPauseMillis50 ‐XX:G1HeapRegionSize16M高性能网卡Kafka本身的服务性能非常高单机就可以支持百万级的TPS在高流量冲击下网络非常有可能优先成为性能瓶颈。对于Kafka服务器建议配置高性能的网卡。成本允许的话尽量选择千兆以上的网卡。 合理优化Kafka集群配置
合理配置partition分区数量
Kafka的单个Partition读写效率是非常高的但是Kafka的Partition设计是非常碎片化。如果Partition文件过多很容易严重影响Kafka的整体性能。
尽量不要使用过多的Topic通常不建议超过3个Topic。一个分区下的副本集数量不要设置过多 将副本数设置为2就可以了。
至于Partition的数量最好根据业务情况灵活调整。partition数量设置多一些可以一定程度增加Topic的吞吐量。但是过多的partition数量还是同样会带来partition索引的压力。
Kafka提供了一个生产者的性能压测脚本可以用来衡量集群的整体性能。
# num-record表示要发送100000条压测消息record-size表示每条消息大小1KBthroughput表示限流控制设置为小于0表示不限流。
# properducer-props用来设置生产者的参数。
[operworker1 bin]$ ./kafka-producer-perf-test.sh --topic test --num-record 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.serversworker1:9092 acks1
94846 records sent, 18969.2 records/sec (18.52 MB/sec), 1157.4 ms avg latency, 1581.0 ms max latency.
133740 records sent, 26748.0 records/sec (26.12 MB/sec), 1150.6 ms avg latency, 1312.0 ms max latency.
146760 records sent, 29346.1 records/sec (28.66 MB/sec), 1051.5 ms avg latency, 1164.0 ms max latency.
137400 records sent, 27480.0 records/sec (26.84 MB/sec), 1123.7 ms avg latency, 1182.0 ms max latency.
158700 records sent, 31740.0 records/sec (31.00 MB/sec), 972.1 ms avg latency, 1022.0 ms max latency.
158775 records sent, 31755.0 records/sec (31.01 MB/sec), 963.5 ms avg latency, 1055.0 ms max latency.
1000000 records sent, 28667.259123 records/sec (28.00 MB/sec), 1030.44 ms avg latency, 1581.00 ms max latency, 1002 ms 50th, 1231 ms 95th, 1440 ms 99th, 1563 ms 99.9th.合理对数据进行压缩
在生产者的ProducerConfig中有一个配置 COMPRESSION_TYPE_CONFIG是用来对消息进行压缩的。
public static final String COMPRESSION_TYPE_CONFIG compression.type;生产者配置了压缩策略后会对生产的每个消息进行压缩从而降低Producer到Broker的网络传输也降低了Broker的数据存储压力。
所支持的几种压缩算法中zstd算法具有最高的数据压缩比但是吞吐量不高lz4在吞吐量方面的优势比较明显。压缩消息必然增加CPU的消耗如果CPU资源紧张就不要压缩了。
关于数据压缩机制可以在Broker的conf/server.properties文件中进行配置。正常情况下Broker从Producer端接收到消息后不会对其进行任何修改但是如果Broker端和Producer端指定了不同的压缩算法就会产生很多异常的表现。
compression.typeType: string
Default: producer
Valid Values: [uncompressed, zstd, lz4, snappy, gzip, producer]如果开启了消息压缩那么在消费者端自然是要进行解压缩的。
在Kafka中消息从Producer到Broker再到Consumer会一直携带消息的压缩方式这样当Consumer读取到消息集合时自然就知道了这些消息使用的是哪种压缩算法也就可以自己进行解压了。但是这时要注意的是应用中使用的Kafka客户端版本和Kafka服务端版本是否匹配。 优化Kafka客户端使用方式
合理保证消息安全
消息生产者设置好发送者应答参数
消息生产者的acks参数可以设置为0、1、all或-1Broker的min.insync.replicas参数。如果生产者的acks设置为-1或all服务端并不是强行要求所有Paritition都完成写入再返回而是可以配置多少个Partition完成消息写入后再往Producer返回消息。 生产者端的幂等性配置
// 值为 true 或 false
public static final String ENABLE_IDEMPOTENCE_CONFIG enable.idempotence;如果要开启幂等性那么
生产者消息缓存机制中的 max.in.flight.requests.per.connection 5重试次数retries0应答机制中的acks必须为all
其他补充点
如果没有设置冲突配置则默认启用幂等性。如果设置了冲突的配置并且幂等性没有显式启用则幂等性被禁用。即显示开启的幂等性又有冲突配置就抛异常 使用生产者事务机制发送消息
// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 提交事务
void commitTransaction() throws ProducerFencedException;
// 4 放弃事务类似于回滚事务的操作
void abortTransaction() throws ProducerFencedException;SpringBoot集成Kafka时使用的KafkaTemplate就提供了事务消息机制 消费者端合理使用提交方式
如果消费者要使用异步方式进行业务处理那么如果业务处理失败此时消费者已经提交了Offset这个消息就无法重试了这就会造成消息丢失。
因此在消费者端尽量不要使用异步处理方式在绝大部分场景下就能够通过Kafka的消费者重试机制保证消息安全处理。此时在消费者端需要更多考虑的问题就变成了消费重试机制造成的消息重复消费的问题。 消费者防止消息重复消费
问题的产生消费者业务处理时间较长此时消费者正常处理消息的过程中Broker端就已经等不下去了认为这个消费者处理失败了。这时就会往同组的其他消费者实例投递消息这就造成了消息重复处理。这就给消费者端带来不必要的幂等性问题。
解决方式 根据业务ID校验。比如对于订单消息消费者根据订单ID去确认一下这个订单消息有没有处理过。 统一的方式处理幂等性问题 将Offset放到Redis中自行进行管理。通过Redis中的offset来判断消息之前是否处理过。 具体实现详情请参考上文《客户端 — 客户端属性分析 — 消息重复消费问题》 生产环境常见问题分析
消息零丢失方案 1、 生产者发送消息到Broker不丢失
生产者丢失消息的原因就和ack机制有关如果将acks设置为1、all或-1 就表示接收到Broker成功将消息写入文件后的ack应答 2、Broker端保证消息不丢失
Broker端丢失消息的原因主要有两种Leader partition重新选举、服务器非正常关机 pageCache中的消息还未刷盘
《服务端的Zookeeper元数据梳理 — Partition故障恢复机制》 《服务端日志 — 合理配置刷盘频率》
消息生产者将acks设置为all可以有效避免因为Leader partition故障导致的消息丢失
而pageCache刷盘可以通过下面的参数来设定合理的配置
# 多长时间进行一次强制刷盘。默认是Long.MAX。
flush.ms# 表示当同一个Partiton的消息条数积累到这个数量时就会申请一次刷盘操作。默认是Long.MAX。
log.flush.interval.messages#当一个消息在内存中保留的时间达到这个数量时就会申请一次刷盘操作。他的默认值是空。如果这个参数配置为空则生效的是下一个参数。
log.flush.interval.ms# 日志刷新程序检查是否需要将日志刷新到磁盘的频率(以毫秒为单位),默认也是Long.MAX。
log.flush.scheduler.interval.ms3、消费者保证消息不丢失
同步处理业务逻辑同步手动提交offset 消息积压如何处理 如果业务运行正常只是因为消费者处理消息过慢造成消息加压。 增加消费者个数最多让一个消费者组下的消费者个数Partition分区数让一个Consumer负责一个分区将消费进度提升到最大。 如果还是不够那就新创建一个topic给这个topic更多的partition然后启动一批消费者将消息从旧topic中搬运到新topic中这些消费者不处理业务消息仅仅是搬运。这样就可以创建更多的消费者消费从新topic消费了 如果是消费者业务处理异常导致的消息积压。 使用一种降级方案。先启动一个Consumer将Topic下的消息先转发到其他队列中然后再慢慢分析新队列里的消息处理问题。类似于死信队列的处理方式。 如何保证消息顺序
Kafka很难保证消息的顺序因为Kafka设计的最优先重点是海量吞吐。 这个问题要分两个方面考虑
Topic下各个partition是并发处理消息的 producer要保证一组有序的消息发送到一个partition中consumer要从partition中顺序处理消息 问题一producer要保证一组有序的消息发送到一个partition中
第一种简单粗暴的方式就只为Topic创建一个partition这种方式放弃了多partition带来的高吞吐量。
第二种方式自己定义一个分区路由机制实现org.apache.kafka.clients.producer.Partitioner接口将消息分配到同一个Partition上。
此时发送方保证了发送时的消息顺序性。但是存在网络问题导致发送的一批消息中某一个消息丢失这样消息到Broker时就还是没办法保证顺序性。Kafka是通过幂等性中单调递增的sequenceNumber来保证消息是顺序因为是单调递增的所以还能判断是否存在消息丢失一旦Kafka发现Producer传过来的SequenceNumber出现了跨越那么就意味着中间有可能消息出现了丢失就会往Producer抛出一个OutOfOrderSequenceException异常。 问题二Partition中的消息有序后如何保证Consumer的消费顺序是有序的
// 一次拉取消息的数量
public static final String FETCH_MAX_BYTES_CONFIG fetch.max.bytes;
Note that the consumer performs multiple fetches in parallel.public static final int DEFAULT_FETCH_MAX_BYTES 50 * 1024 * 1024;从上方fetch.max.bytes参数的最后一个解释可以知道Consumer其实是每次并行的拉取多个Batch批次的消息进行处理的。所以在Kafka提供的客户端Consumer中是没有办法直接保证消费的消息顺序。
我们能做的就是在Consumer的处理逻辑中将消息进行排序。比如将消息按照业务独立性收集到一个集合中然后在集合中对消息进行排序。
RocketMQ中提供了顺序消息的实现。他的实现原理是先锁定一个队列消费完这一个队列后才开始锁定下一个队列并消费队列中的消息。再结合MessageQueue中的消息有序性就能保证整体消息的消费顺序是有序的。
