手机网站指向什么意思,烟台seo网络推广,有域名了 怎么做网站,备案号查询平台Kafka是由LinkedIn开发的一个分布式发布/订阅的消息系统和一个强大的队列#xff0c;使用Scala编写#xff0c;它以可扩展和高吞吐率而被广泛使用。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上#xff0c;并在群集内以master-flower方式实现数据同步#xff0c;… Kafka是由LinkedIn开发的一个分布式发布/订阅的消息系统和一个强大的队列使用Scala编写它以可扩展和高吞吐率而被广泛使用。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上并在群集内以master-flower方式实现数据同步从而防止数据丢失。 1、组件和角色 Producer消息生产者发布消息到 kafka 集群的终端或服务。 Consumer从 kafka 集群中消费消息的终端或服务。 Consumer grouphigh-level consumer API 中每个 consumer 都属于一个 consumer group一个partition只能被同一个 consumer group 中的一个 Consumer 消费但可以被多个不同consumer group 中的consumer消费。 Broker: 集群中的每一个kafka进程都是一个Broker通常一台服务器上部署一个broker。 Topic 每条发布到 kafka 集群的消息属于的类别即kafka是面向topic的topic是逻辑概念。 Partition每个topic包含一个或多个partition。kafka分配的单位是partitionpartition是物理概念生产者发送的消息就是保存在partition中的。 Segmentpartition物理上由多个segment组成。 offset : 每个partition都由一系列有序的、不可变的消息组成这些消息被连续的追加到partition中。partition中的每个消息都有一个连续递增的序列号叫做offset偏移量offset在每个分区中是唯一的。 replicapartition的副本保障 partition 的高可用。leader和follower统称为Replica。在kafka集群中为了防止数据丢失每个partition都会有主分区和从分区当然也可以没有从分区。每个partition有且只有一个主分区可以没有从分区也可以有一个或者多个从分区。 leaderreplica中的一个角色主分区所在的节点称为leader。在kafka集群中每个partition都有一个leaderproducer和consumer只跟leader交互leader负责数据的读写。 followerreplica中的一个角色从分区所在的节点称为follower从leader中复制fentch数据。为了防止leader与follower节点上数据不一致性的问题kafka没有使用读写分离而是只在leader节点上读写数据follower节点只是从leader节点上定期复制数据。如果leader节点异常随机选择一个follower节点成为leader节点从而防止数据丢失。 controllerkafka 集群中的一个broker用来进行 leader 选举以及各种故障转移。 zookeeperkafka 通过 zookeeper 来存储集群的 meta 信息meta信息主要包括kafka的broker列表(ip:port)、topic和partition等信息。 ARAssigned Replica 已分配的副本表示某个分区的所有副本。 ISRIn-Sync-Replica 在同步中的副本表示所有与leader副本保持一定程度同步的副本包括leader副本在内。 OSROut-of-Sync-Replica 不在同步中的副本表示所有与leader副本同步滞后过多的follower副本不包括leader副本。 AR ISR OSR 正常情况下所有的follower副本都应该与leader副本保持同步即AR ISROSR集合为空。 1controller的选举 Kafka启动时会在所有的broker集群节点中选择一个controllerleader和follower是针对分区而言的而controller是针对broker而言的。创建topic、添加分区、修改副本数量等管理任务都是由controller完成的以及Kafka分区leader的选举也是有controller决定的。 在Kafka集群启动时每个broker都会将自己注册到zookeeper上并尝试在zookeeper上抢锁抢占成功的broker就注册成为ControllerZK临时节点。只会有一个broker节点竞争成功其他的broker会注册该节点的监视器一旦该临时节点状态发生变化就可以进行相应的处理。Controller是高可用的一旦某个broker崩溃其他的broker会重新注册成为Controller。 2leader的选举 所有分区的leader选举都是由controller决定的controller会将leader的改变直接通过RPC的方式通知需为此做出响应的brokercontroller读取到当前分区的ISR只有一个replica存活时就选择这个replica作为leader否则任意选择一个replica作为leader如果该分区的所有replica都已经宕机则新的leader为-1。 为什么不通过ZK的方式选举分区的leader Kafka集群如果业务很多的情况下会存在很多的分区假设某个broker宕机就会出现很多的分区都需要重新选举leader如果使用zookeeper选举leader会给zk带来巨大的压力。因此Kafka中leader的选举不能使用zk来实现。 2、原理简介 1.一个Topic分为多个Partition来进行数据管理一个Partition中的数据是有序、不可变的使用偏移量(offset)唯一标识一条数据是一个long类型的数据。Partition接收到producer发送过来的数据后会产生一个递增的offset偏移量数据同时将数据保存到本地的磁盘文件中(文件内容以追加的方式写入数据)Partition中的数据存活时间超过参数值(log.retention.{ms,minutes,hours}默认7天)的时候进行删除(默认)。Consumer根据offset消费对应Topic的Partition中的数据(也就是每个Consumer消费的每个Topic的Partition都拥有自己的offset偏移量)。注意Kafka的数据消费是顺序读写的磁盘的顺序读写速度(600MB/sec)比随机读写速度(100k/sec)快。 2.在Kafka集群中producer生产数据并发送到对应的Topic。Producer通过push的方式将数据发送到对应Topic的分区Producer发送到Topic的数据是由key/value键值对组成的Kafka根据不同的key将数据发送到不同的Partition默认采用Hash的机制发送数据到对应Topic的不同Partition中配置参数为{partitioner.class}。也可以配置自定义分配机制自定义类实现Partitioner接口重写partition方法的方式。Producer发送数据的方式分为sync(同步)和async(异步)两种默认为同步方式 由参数{producer.type}决定当发送模式为异步发送的时候Producer提供重试机制默认失败重试发送3次。 3.如果生产者同步发消息在收到kafka的ack告知发送成功之前一直处于阻塞状态。如果生产者异步发消息发送完之后不用等待broker给回复直接执行后面的业务逻辑。可以提供回调方法让broker异步的调用callback告知生产者消息发送的结果。如果告知的结果异常再进行相应的处理操作。 4.Kafka有两种模式消费数据队列和发布订阅在队列模式下一条数据只会发送给consumer group中的一个consumer进行消费在发布订阅模式下一条数据会发送给多个consumer进行消费。Kafka中通过控制consumer的参数{group.id}来决定kafka是什么数据消费模式如果所有消费者的该参数值是相同的那么此时的kafka就是类似于队列模式数据只会发送到一个consumer此时类似于负载均衡否则就是发布订阅模式。Kafka的consumer基于offset对kafka中的数据进行消费。 5.Kafka的数据是按照分区进行排序的(插入的顺序)也就是每个分区中的数据是有序的。在Consumer进行数据消费的时候也是对分区的数据进行有序消费的 但是不保证所有数据的有序性(多个分区之间)同一个分区数据先进先出。 6.Consumer Rebalance当一个consumer group组中的消费者数量和对应Topic的分区数量一致的时候此时一个Consumer消费一个Partition的数据 如果不一致那么可能出现一个Consumer消费多个Partition的数据或者不消费数据的情况这个机制是根据Consumer和Partition的数量动态变化的。Consumer通过poll的方式主动从Kafka集群中获取数据。 7.Kafka的Replication指的是Partition的复制一个Partition的所有分区中只有一个分区是leader节点其它分区是follower节点。Replication对Kafka的吞吐率有一定的影响但是极大的增强了可靠性。Follower节点会定时的从leader节点上获取增量数据一个活跃的follower节点必须满足以下两个条件 (1)所有节点必须维护与zookeeper的连接(通过zk的heartbeat实现) (2)follower必须能够及时的将leader上的writing复制过来不能“落后太多”由参数{replica.lag.time.max.ms}和{replica.lag.max.messages}决定。 8.MessageDeliverySemantics是消息系统中数据传输的可靠性保证的一个定义主要分为三种类型 At most once最多一次消息可能会丢失但是不可能重复发送。At least once最少一次消息不可能丢失但是可能重复发送。Exactly once仅仅一次消息只发送一次但不存在消息的丢失。Kafka的Producer通过参数{request.required.acks}来确定Producer和Broker之间是哪种消息传递类型。Ack0相当于异步发送意味着producer不等待broker同步完成消息发送完毕继续发送下一批消息。提供了最低延迟但持久性最弱当broker发生故障时很可能发生数据丢失。如果leader死亡producer继续发送消息broker接收不到数据就会造成数据丢失。 Ack1producer要等待leader成功收到消息并确认才发送下一条message。提供较低的延迟性以及较好的持久性。但是如果partition下的leader死亡而follower尚未复制数据数据就会丢失。 Ack-1leader收到所有消息且follower同步完数据才发送下一条数据。延迟性最差持久性最好即可靠性最好。 三种参数设置性能递减可靠性递增。 同时Ack默认值为1此时吞吐量与可靠性折中。实际生产中可以根据实际需求进行调整。 3、常用参数介绍 1kafka的server.properties配置文件中参数 broker.id0  #当前机器在集群中的唯一标识和zookeeper的myid性质一样 port9092 #当前kafka对外提供服务的端口默认是9092 host.name192.168.7.100 #这个参数默认是关闭的在0.8.1有个bugDNS解析问题失败率的问题。改成自己centos的ip地址。 num.network.threads3 #这个是borker进行网络处理的线程数 num.io.threads8 #这个是borker进行I/O处理的线程数 log.dirs/opt/kafka/kafkalogs/ #消息存放的目录这个目录可以配置为“”逗号分割的表达式上面的num.io.threads要大于这个目录的个数这个目录如果配置多个目录新创建的topic他把消息持久化的地方是当前以逗号分割的目录中那个分区数最少就放那一个 socket.send.buffer.bytes102400 #发送缓冲区buffer大小数据不是一下子就发送的先回存储到缓冲区了到达一定的大小后再发送能提高性能 socket.receive.buffer.bytes102400 #kafka接收缓冲区大小当数据到达一定大小后在序列化到磁盘 socket.request.max.bytes104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数这个值不能超过java的堆栈大小 num.partitions1 #默认的分区数一个topic默认1个分区数 log.retention.hours168 #默认消息的最大持久化时间168小时7天 message.max.byte5242880  #消息保存的最大值5M default.replication.factor2  #kafka保存消息的副本数如果一个副本失效了另一个还可以继续提供服务 replica.fetch.max.bytes5242880  #取消息的最大直接数 log.segment.bytes1073741824 #这个参数是因为kafka的消息是以追加的形式落地到文件当超过这个值的时候kafka会新起一个文件 log.retention.check.interval.ms300000 #每隔300000毫秒去检查上面配置的log失效时间log.retention.hours168 到目录查看是否有过期的消息如果有删除 log.cleaner.enablefalse #是否启用log压缩一般不用启用启用的话可以提高性能 zookeeper.connect192.168.7.100:2181,192.168.7.101:2181,192.168.7.107:2181/kafka #设置zookeeper的连接端口在集群配置时要把所有机器的ip地址都要写上这里以三个机器为例。如果是单机部署只需要写一个ip地址就行了。 注意在zookeeper.connect的最后加上/kafka是因为kafka需要依赖zookeeper在kafka启动之后默认会在zookeeper服务所在节点的根目录下创建很多与kafka有关的目录这样就会导致zookeeper服务所在节点的根目录下的文件很多很乱。另外如果多个kafka共用一个zookeeper就会导致zookeeper服务的根目录下各个kafka文件更加混乱。所以在zookeeper.connect的最后加上/kafka是为了在kafka启动时将创建的文件都放到zookeeper节点根目录下的/kafka子目录下。多个kafka共用一个zookeeper时可以分别配置自己的子目录以示区分。 启动zookeeper和kafka之后会自动在zookeeper节点上创建/kafka目录。 2生产者producer.properties配置文件中的参数 1.bootstrap.servershost1:port1,host2:port2  // 用于生产者与kafka集群建立连接 2. acks表示Producer需要Leader确认的模式。 1acks 0: 表示Producer请求立即返回不需要等待Leader的任何确认。这种方案有最高的吞吐率但是不保证消息是否真的发送成功。 2acks -1: 表示分区Leader必须等待消息被成功写入到所有的ISR副本(同步副本)中才认为Producer请求成功。这种方案提供最高的消息持久性保证但是理论上吞吐率也是最差的。 3acks 1: 表示Leader副本必须应答此Producer请求并写入消息到本地日志之后Producer请求被认为成功。如果此时Leader副本应答请求之后挂掉了消息会丢失。这个方案提供了不错的持久性保证和吞吐。 3. compression.typenone  // 压缩类型目前支持none(不压缩), gzip, snappy, lz4, zstd 4. partitioner.class kafka.producer.DefaultPartitioner  // 分区的策略默认是取模 5. request.timeout.ms10000  // 消息发送的最长等待时间 6. linger.ms0  //这个值是为了延迟发送来收集更多的消息一批发送Producer是按照batch进行发送的但是还要看linger.ms的值默认是0表示不延迟。为了减少网络IO提升整体的性能建议设置5-100ms。 7. batch.size16384   // Producer按照batch进行发送通过这个参数来设置批量提交的数据大小默认是16KB当积压的消息达到这个值的时候就会统一发送发往同一分区的消息。 8. buffer.memory33554432  //该参数用于指定Producer端用于缓存消息的缓冲区大小单位为字节默认值为33554432即32MB。发送的消息会先进入到本地缓冲区32MB生产者会跑一个线程该线程去缓冲区中取16KB的数据发送到kafka如果到10毫秒数据没取满16KB也会发送一次。异步的时候假如设置了缓存消息数量为200但是一直没有200条数据那么不可能一直等下去就会取16KB大小的数据直接发不够16KB也会发。 3消费者consumer.properties配置文件中的参数 1.bootstrap.servers host1:port1,host2:port2 ...  // 用于消费者与kafka集群建立连接 2. group.idtest-consumer-group  // 标记消费者所属的消费者组 3. key.deserializer和value.deserializer指定接收消息的key和value的反序列化类型。一定要写全类名。 4. enable.auto.commit默认值为true消费者会自动周期性地向服务器提交偏移量。 5. auto.commit.interval.ms如果设置了 enable.auto.commit 的值为true 则该值定义了消费者偏移量向Kafka提交的频率默认5s。 6. auto.offset.reset当Kafka中没有初始偏移量或当前偏移量在服务器中不存在如数据被删除了读取数据偏移量的处理方式 1earliest自动重置偏移量到最早的偏移量。 2latest默认自动重置偏移量为最新的偏移量。 3none如果消费组原来的偏移量不存在则向消费者抛异常。 7. max.poll.records一次poll拉取数据返回消息的最大条数默认500条。 8.offsets.topic.num.partitions__consumer_offsets的分区数默认是50个分区。 9.heartbeat.interval.msKafka消费者和coordinator之间的心跳时间默认3s。该条目的值必须小于 session.timeout.ms 也不应该高于 session.timeout.ms 的1/3。 10.session.timeout.msKafka消费者和coordinator之间连接超时时间默认45s。超过该值该消费者被移除消费者组执行再平衡。 11.max.poll.interval.ms消费者处理消息的最大时长默认是5分钟。超过该值该消费者被移除消费者组执行再平衡。 12.fetch.min.bytes默认1个字节。消费者获取服务器端一批消息最小的字节数。 13.fetch.max.wait.ms默认500ms。如果没有从服务器端获取到一批数据的最小字节数。该时间到仍然会返回数据。 14.fetch.max.bytes默认值: 52428800字节即50MB。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值仍然可以拉取回来这批数据因此这不是一个绝对最大值。一批次的大小受message.max.bytes broker configor max.message.bytes topic config影响。