宁波网站建设策划公司排名,网站软文发多了会影响,含山县建设局网站,1小时前俄乌战况消息kafka简介Kafka是最初由Linkedin公司开发#xff0c;是一个分布式、支持分区的#xff08;partition#xff09;、多副本的#xff08;replica#xff09;#xff0c;基于zookeeper协调的分布式消息系统#xff0c;它的最大的特性就是可以实时的处理大量数据以满足各种需…kafka简介Kafka是最初由Linkedin公司开发是一个分布式、支持分区的partition、多副本的replica基于zookeeper协调的分布式消息系统它的最大的特性就是可以实时的处理大量数据以满足各种需求场景比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎web/nginx日志、访问日志消息服务等等用scala语言编写Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。关键词分布式流处理平台。在系统之间构建实时数据流管道。以topic分类对记录进行存储每个记录包含key-valuetimestamp每秒钟百万消息吞吐量。安装kafka0.选择三台主机安装kafka
1.准备zk
略
2.jdk
略
3.tar文件
4.环境变量
略
5.配置kafka
[kafka/config/server.properties]
...
broker.id201
...
listenersPLAINTEXT://:9092
...
log.dirs/home/centos/kafka/logs
...
zookeeper.connects201:2181,s202:2181,s203:21816.分发server.properties同时修改每个文件的broker.id7.启动kafka服务器
a)先启动zk
b)启动kafka
[s202 ~ s204]
$bin/kafka-server-start.sh -daemon config/server.propertiesc)验证kafka服务器是否启动
$netstat -anop | grep 90928.创建主题
$bin/kafka-topics.sh --create --zookeeper s201:2181 --replication-factor 3 --partitions 3 --topic test9.查看主题列表
$bin/kafka-topics.sh --list --zookeeper s201:218110.启动控制台生产者
$bin/kafka-console-producer.sh --broker-list s202:9092 --topic test11.启动控制台消费者
$bin/kafka-console-consumer.sh --bootstrap-server s202:9092 --topic test --from-beginning --zookeeper s202:218112.在生产者控制台输入hello worldkafka 的使用场景埋点日志的收集一个公司可以用Kafka可以收集各种服务的log。消息系统解耦和生产者和消费者、缓存消息等。用户活动跟踪Kafka经常被用来记录web用户或者app用户的各种活动如浏览网页、搜索、点击等活动这些活动信息被各个服务器发布到kafka的topic中然后订阅者通过订阅这些topic来做实时的监控分析或者装载到hadoop、数据仓库中做离线分析和挖掘运营指标Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据生产各种操作的集中反馈比如报警和报告。流式处理比如spark streaming和flink事件源kafka如何保证的消息数据不丢失当讨论这个问题的时候首先需要考量kafka的运行机制。kafka主要分为三个组件producer、consumer、broker。所以也必须从三个方面去考量producer、consumer、broker端数据不丢失。一、producer端如何保证数据不丢失1.ack的配配置策略acks 0
生产者发送消息之后 不需要等待服务端的任何响应它不管消息有没有发送成功如果发送过程中遇到了异常
导致broker端没有收到消息消息也就丢失了。实际上它只是把消息发送到了socketBuffer(缓存)中
而socketBuffer什么时候被提交到broker端并不关心它不担保broker端是否收到了消息
但是这样的配置对retry是不起作用的因为producer端都不知道是否发生了错误
而且对于offset的获取永远都是-1因为broker端可能还没有开始写数据。
这样不保险的操作为什么还有这样的配置kafka对于收集海量数据
如果在收集某一项日志时是允许数据量有一定丢失的话是可以用这种配置来收集日志。
acks 1(默认值)
生产者发送消息之后只要分区的leader副本成功写入消息那么它就会收到来自服务端的成功响应。
其实就是消息只发给了leader leader收到消息后会返回ack到producer端。
如果消息无法写入leader时(选举、宕机等情况时)生产都会收到一个错误的响应为了避免消息丢失
生产者可以选择重发消息如果消息成功写入在被其它副本同步数据时leader 崩溃,那么此条数据
还是会丢失因为新选举的leader是没有收到这条消息ack设置为1是消息可靠性和吞吐量折中的方案。
acks all (或-1)
生产者在发送消息之后需要等待ISR中所有的副本都成功写入消息之后才能够收到来自服务端的成功响应
在配置环境相同的情况下此种配置可以达到最强的可靠性。即在发送消息时需要leader 向fllow
同步完数据之后也就是ISR队列中所有的broker全部保存完这条消息后才会向ack发送消息表示发送成功。2.retries的配置策略在kafka中错误分为2种一种是可恢复的另一种是不可恢复的。 可恢复性的错误 如遇到在leader的选举、网络的抖动等这些异常时如果我们在这个时候配置的retries大于0的 也就是可以进行重试操作那么等到leader选举完成后、网络稳定后这些异常就会消息错误也就可以恢复 数据再次重发时就会正常发送到broker端。需要注意retries(重试)之间的时间间隔 以确保在重试时可恢复性错误都已恢复。 不可恢复性的错误 如超过了发送消息的最大值(max.request.size)时这种错误是不可恢复的如果不做处理 那么数据就会丢失因此我们需要注意在发生异常时把这些消息写入到DB、缓存本地文件中等等 把这些不成功的数据记录下来等错误修复后再把这些数据发送到broker端。如何选择高可用型配置acks allretries 0 retry.backoff.ms100(毫秒) (并根据实际情况设置retry可能恢复的间隔时间) 优点这样保证了producer端每发送一条消息都要成功如果不成功并将消息缓存起来等异常恢复后再次发送。 缺点这样保证了高可用但是这会导致集群的吞吐量不是很高因为数据发送到broker之后leader要将数据同步到fllower上如果网络带宽、不稳定等情况时ack响应时间会更长折中型配置acks 1 retries 0 retries 时间间隔设置 (并根据实际情况设置retries可能恢复的间隔时间) 优点保证了消息的可靠性和吞吐量是个折中的方案 缺点性能处于2者中间3.高吞吐型 高效率配置acks 0 优点可以相对容忍一些数据的丢失吞吐量大可以接收大量请求 缺点不知道发送的消息是 否成功每种配置都有对应的生产用途视情况而定。。二、consumer端如何保证数据不丢失consumer端配置1、group.id: consumer group 分组的一个id消费者隶属消费组的名称kafka的每个partition值允许同一个group的一个consumer消费。这样做的目的是为了保证kafka的高吞吐量2、auto.offset.reset earliest(最早) /latest(最晚)设置从哪个位置开始消费3、enable.auto.commit true/false(默认true)当设置为true时意味着由kafka的consumer端自己间隔一定的时间会自动提交offset如果设置成了fasle也就是由客户端(自己写代码)来提交那就还得控制提交的时间间隔auto.commit.interval.ms当enabe.auto.commit设置为true时才生效表示开启自动提交消费位移功能时自动提交消费位移的时间间隔。在consumer阶段如果设置为true意味着会自动提交offset比如说当你pull了30条数据但是当处理20条数据的时候自动提交了commit当处理21条数据的时候系统崩了那当你再去拉取数据的时候就会从30开始啦那就会丢失21-30的数据如果设置为false可以手动提交你可以处理一条提交一次也可以处理一批提交一批但是consumer在消费数据的时候是以batch的模式去pull数据的假设pull了30条数据你在处理30条数据的时候没处理一条就提交一次的话会非常影响消费能力你可以还是按照一批来处理设置一个累加器处理一条加1如果在处理数据时发生了异常那就把当前处理失败的offset进行提交(放在finally代码块中)注意一定要确保offset的正确性当下次再次消费的时候就可以从提交的offset处进行再次消费。consumer 保证确保消息只被处理一次处理同时确保幂等性需要结合具体的业务来看 比如你拿个数据要写库先根据主键查一下如果这数据都有了你就别插入了update一下好吧 比如你是写redis那没问题了反正每次都是set天然幂等性 比如你不是上面两个场景那做的稍微复杂一点你需要让生产者发送每条数据的时候里面加一个全局唯一的id类似订单id之类的东西然后你这里消费到了之后先根据这个id去比如redis里查一下之前消费过吗如果没有消费过你就处理然后这个id写redis。如果消费过了那你就别处理了保证别重复处理相同的消息即可。 还有比如基于数据库的唯一键来保证重复数据不会重复插入多条拿到数据的时候每次重启可能会有重复因为kafka消费者还没来得及提交offset重复数据拿到了以后我们插入的时候因为有唯一键约束了所以重复数据只会插入报错不会导致数据库中出现脏数据 三、broker端是如何保证数据不丢失的1.replication-factor 3 在创建topic时会通过replication-factor来创建副本的个数它提高了kafka的高可用性同时它允许n-1台broker挂掉设置好合理的副本因子对kafka整体性能是非常有帮助的通常是3个极限是5个如果多了也会影响开销。2.min.insync.replicas 2 分区ISR队列集合中最少有多少个副本默认值是1 3.unclean.leader.election.enable false 是否允许从ISR队列中选举leader副本默认值是false,如果设置成true,则可能会造成数据丢失。leader选举造成的数据丢失3个replica分别为0 1 20为leader数据都能完全同步到100在某一时刻分别有2个fllow挂掉了此时有producer往0 的replica上发送50条数据完后此时的leader挂掉了而此时刚好的1个fllow起来了它没有向leader上feach数据因为leader已经不存在了此时有2种处理方法重新起来的fllow可以成为1个leader需要通过 unclean.leader.election.enabletrue,这样做保证了高可用但是这样做的弊端是新起来的fllow成为了leader,但是它会丢失部分数据虽然这样保证了高可用。另一种情况是设置为false,不让fllow竞选leader但是这样也会造成数据的丢失。假如在ISR的队列里面只有0 1但此时replica 1 没有来得及向leader feach数据leader挂掉了这样也会造成数据的丢失。broker配置策略min.insync.replica在一个topic中1个分区 有3个副本在创建时设置了min.insync.replica2,假如此时在ISR中只有leader副本(1个)存在在producer端生产数据时,此时的acksall,这也就意味着在producer向broker端写数据时必须保证ISR中指定数量的副本(包含leader、fllow副本)全部同步完成才算写成功这个数量就是由min.insync.replica来控制的这样producer端向broker端写数据是不成功因为ISR中只有leader副本min.insync.replica要求2个副本此时的producer生产数据失败(异常)当然consumer端是可以消费数据的只不过是没有新数据产生而已.这样保证了数据的一致性但这样会导致高可用性降低了。一般的配置是按 n/2 1 来配置min.insync.replicas 的数量的同时也要将unclean.leader.election.enablefalseunclean.leader.election.enable假如现在有leader 0 fllow 1 fllow 2 三个副本存储的数据量分别是10 9 8此时的broker的配置是min.insync.replica2 acksall,leader的数据更新到了15在没有同步到fllow 1 fllow 2时挂掉了此时的ISR队列中是有fllow 1 和fllow 2的如果unclean.leader.election.enable设置的是true,表示在ISR中的副本是可以竞选leader这样就会造成9-15或8-15之间的数据丢失所以unclean.leader.election.enable必须设置成成false这样整个kafka cluster都不读写了这样就保证了数据的高度一致性.kafka中topic设计原理因为consumer group 中所有的consumer一定会消费topic中的partition而一个partition只能同时被同一group中的一个consumer消费所以最优的设计就是consumer group下的consumer thread的数量等于partition数量这样效率是最高的。一个Topic的Partition数量大于等于Broker的数量可以提高吞吐率参考文章https://www.cnblogs.com/MrRightZhao/p/11498952.html
