摄影学习网站,创新型的福州网站建设,html背景图片代码,网站开发语言总结有哪些目录
一、Kafka
1、Message Queue是什么#xff1f;
2、Kafka 基础架构
3、Kafka安装
4、Offset自动控制
5、Acks Retries
6、幂等性
7、事务控制
8、数据同步机制
9、Kafka-Eagle
二、Maven项目测试
1、Topic API
2、生产者消费者 一、Kafka Kafka是…目录
一、Kafka
1、Message Queue是什么
2、Kafka 基础架构
3、Kafka安装
4、Offset自动控制
5、Acks Retries
6、幂等性
7、事务控制
8、数据同步机制
9、Kafka-Eagle
二、Maven项目测试
1、Topic API
2、生产者消费者 一、Kafka Kafka是由Apache软件基金会开发的一个开源流处理平台由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统它可以收集并处理用户在网站中的所有动作流数据以及物联网设备的采样信息。 Apache Kafka是Apache软件基金会的开源的流处理平台该平台提供了消息的订阅与发布的消息队列一般用作系统间解耦、异步通信、削峰填谷等作用。同时Kafka又提供了Kafka streaming插件包实现了实时在线流处理。相比较一些专业的流处理框架不同Kafka Streaming计算是运行在应用端具有简单、入门要求低、部署方便等优点。 - 消息队列Message Queue - Kafka Streaming 流处理
Kafka官网
Kafka官网文档
1、Message Queue是什么 消息队列是一种在分布式和大数据开发中不可或缺的中间件。在分布式开发或者大数据开发中通常使用消息队列进行缓冲、系统间解耦和削峰填谷等业务场景常见的消息队列工作模式大致会分为两大类
至多一次消息生产者将数据写入消息系统然后由消费者负责去拉取消息服务器中的消息一旦消息被确认消费之后 由消息服务器主动删除队列中的数据这种消费方式一般只允许被一个消费者消费并且消息队列中的数据不允许被重复消费。没有限制同上诉消费形式不同生产者发不完数据以后该消息可以被多个消费者同时消费并且同一个消费者可以多次消费消息服务器中的同一个记录。主要是因为消息服务器一般可以长时间存储海量消息。
2、Kafka 基础架构 Kafka集群以Topic形式负责分类集群中的Record每一个Record属于一个Topic。每个Topic底层都会对应一组分区的日志用于持久化Topic中的Record。同时在Kafka集群中Topic的每一个日志的分区都一定会有1个Borker担当该分区的Leader其他的Broker担当该分区的followerLeader负责分区数据的读写操作follower负责同步该分区的数据。这样如果分区的Leader宕机该分区的其他follower会选取出新的leader继续负责该分区数据的读写。其中集群的中Leader的监控和Topic的部分元数据是存储在Zookeeper中。
Topics and Logs Kafka中所有消息是通过Topic为单位进行管理每个Kafka中的Topic通常会有多个订阅者负责订阅发送到该Topic中的数据。Kafka负责管理集群中每个Topic的一组日志分区数据。 生产者将数据发布到相应的Topic。负责选择将哪个记录分发送到Topic中的哪个Partition。例如可以round-robin方式完成此操作然而这种仅是为了平衡负载。也可以根据某些语义分区功能例如基于记录中的Key进行此操作。 每组日志分区是一个有序的不可变的的日志序列分区中的每一个Record都被分配了唯一的序列编号称为是offsetKafka 集群会持久化所有发布到Topic中的Record信息该Record的持久化时间是通过配置文件指定,默认是168小时。
log.retention.hours168 Kafka底层会定期的检查日志文件然后将过期的数据从log中移除由于Kafka使用硬盘存储日志文件因此使用Kafka长时间缓存一些日志文件是不存在问题的。 在消费者消费Topic中数据的时候每个消费者会维护本次消费对应分区的偏移量消费者会在消费完一个批次的数据之后会将本次消费的偏移量提交给Kafka集群因此对于每个消费者而言可以随意的控制改消费者的偏移量。因此在Kafka中消费者可以从一个topic分区中的任意位置读取队列数据由于每个消费者控制了自己的消费的偏移量因此多个消费者之间彼此相互独立。Kafka中对Topic实现日志分区的有以下目的
首先它们允许日志扩展到超出单个服务器所能容纳的大小。每个单独的分区都必须适合托管它的服务器但是一个Topic可能有很多分区因此它可以处理任意数量的数据。其次每个服务器充当其某些分区的Leader也可能充当其他分区的Follwer因此群集中的负载得到了很好的平衡。
生产者消费者 消费者使用Consumer Group名称标记自己并且发布到Topic的每条记录都会传递到每个订阅Consumer Group中的一个消费者实例。如果所有Consumer实例都具有相同的Consumer Group那么Topic中的记录会在改ConsumerGroup中的Consumer实例进行均分消费如果所有Consumer实例具有不同的ConsumerGroup则每条记录将广播到所有Consumer Group进程。 更常见的是我们发现Topic具有少量的Consumer Group每个Consumer Group可以理解为一个“逻辑的订阅者”。每个Consumer Group均由许多Consumer实例组成以实现可伸缩性和容错能力。这无非就是发布-订阅模型其中订阅者是消费者的集群而不是单个进程。这种消费方式Kafka会将Topic按照分区的方式均分给一个Consumer Group下的实例如果ConsumerGroup下有新的成员介入则新介入的Consumer实例会去接管ConsumerGroup内其他消费者负责的某些分区同样如果一下ConsumerGroup下的有其他Consumer实例宕机则由改ConsumerGroup其他实例接管。 由于Kafka的Topic的分区策略因此Kafka仅提供分区中记录的有序性也就意味着相同Topic的不同分区记录之间无顺序。因为针对于绝大多数的大数据应用和使用场景 使用分区内部有序或者使用key进行分区策略已经足够满足绝大多数应用场景。但是如果您需要记录全局有序则可以通过只有一个分区Topic来实现尽管这将意味着每个ConsumerGroup只有一个Consumer进程。
高性能之道-顺序写mmap Kafka的特性之一就是高吞吐率但是Kafka的消息是保存或缓存在磁盘上的一般认为在磁盘上读写数据是会降低性能的但是Kafka即使是普通的服务器Kafka也可以轻松支持每秒百万级的写入请求超过了大部分的消息中间件这种特性也使得Kafka在日志处理等海量数据场景广泛应用。Kafka会把收到的消息都写入到硬盘中防止丢失数据。为了优化写入速度Kafka采用了两个技术顺序写入和MMFile 。 因为硬盘是机械结构每次读写都会寻址-写入其中寻址是一个“机械动作”它是最耗时的。所以硬盘最讨厌随机I/O最喜欢顺序I/O。为了提高读写硬盘的速度Kafka就是使用顺序I/O。这样省去了大量的内存开销以及节省了IO寻址的时间。但是单纯的使用顺序写入Kafka的写入性能也不可能和内存进行对比因此Kafka的数据并不是实时的写入硬盘中 。 Kafka充分利用了现代操作系统分页存储来利用内存提高I/O效率。Memory Mapped Files(后面简称mmap)也称为内存映射文件在64位操作系统中一般可以表示20G的数据文件它的工作原理是直接利用操作系统的Page实现文件到物理内存的直接映射。完成MMP映射后用户对内存的所有操作会被操作系统自动的刷新到磁盘上极大地降低了IO使用率。
Zero Copy Kafka服务器在响应客户端读取的时候底层使用ZeroCopy技术直接将磁盘无需拷贝到用户空间而是直接将数据通过内核空间传递输出数据并没有抵达用户空间。 传统IO操作
1.用户进程调用read等系统调用向操作系统发出IO请求请求读取数据到自己的内存缓冲区中。自己进入阻塞状态。2.操作系统收到请求后进一步将IO请求发送磁盘。3.磁盘驱动器收到内核的IO请求把数据从磁盘读取到驱动器的缓冲中。此时不占用CPU。当驱动器的缓冲区被读满后向内核发起中断信号告知自己缓冲区已满。4.内核收到中断使用CPU时间将磁盘驱动器的缓存中的数据拷贝到内核缓冲区中。5.如果内核缓冲区的数据少于用户申请的读的数据重复步骤3跟步骤4直到内核缓冲区的数据足够多为止。6.将数据从内核缓冲区拷贝到用户缓冲区同时从系统调用中返回。完成任务 DMA读取
1.用户进程调用read等系统调用向操作系统发出IO请求请求读取数据到自己的内存缓冲区中。自己进入阻塞状态。2.操作系统收到请求后进一步将IO请求发送DMA。然后让CPU干别的活去。3.DMA进一步将IO请求发送给磁盘。4.磁盘驱动器收到DMA的IO请求把数据从磁盘读取到驱动器的缓冲中。当驱动器的缓冲区被读满后向DMA发起中断信号告知自己缓冲区已满。4.DMA收到磁盘驱动器的信号将磁盘驱动器的缓存中的数据拷贝到内核缓冲区中。此时不占用CPU。这个时候只要内核缓冲区的数据少于用户申请的读的数据内核就会一直重复步骤3跟步骤4直到内核缓冲区的数据足够多为止。5.当DMA读取了足够多的数据就会发送中断信号给CPU。6.CPU收到DMA的信号知道数据已经准备好于是将数据从内核拷贝到用户空间系统调用返回。 跟IO中断模式相比DMA模式下DMA就是CPU的一个代理它负责了一部分的拷贝工作从而减轻了CPU的负担。DMA的优点就是中断少CPU负担低。
网络IO
一般方案
1、文件在磁盘中数据被copy到内核缓冲区 2、从内核缓冲区copy到用户缓冲区 3、用户缓冲区copy到内核与socket相关的缓冲区。 4、数据从socket缓冲区copy到相关协议引擎发送出去
Zero拷贝
1、文件在磁盘中数据被copy到内核缓冲区 2、从内核缓冲区copy到内核与socket相关的缓冲区。 3、数据从socket缓冲区copy到相关协议引擎发送出去 3、Kafka安装
Docker安装Kafka 4、Offset自动控制
Kafka消费者默认对于未订阅的topic的offset的时候也就是系统并没有存储该消费者的消费分区的记录信息默认Kafka消费者的默认首次消费策略latest
auto.offset.resetlatestearliest - 自动将偏移量重置为最早的偏移量
latest - 自动将偏移量重置为最新的偏移量
none - 如果未找到消费者组的先前偏移量则向消费者抛出异常
Kafka消费者在消费数据的时候默认会定期的提交消费的偏移量这样就可以保证所有的消息至少可以被消费者消费1次,用户可以通过以下两个参数配置
enable.auto.commit true 默认
auto.commit.interval.ms 5000 默认
如果用户需要自己管理offset的自动提交可以关闭offset的自动提交手动管理offset提交的偏移量注意用户提交的offset偏移量永远都要比本次消费的偏移量1因为提交的offset是kafka消费者下一次抓取数据的位置。
5、Acks Retries
Kafka生产者在发送完一个的消息之后要求Broker在规定的时间Ack应答如果没有在规定时间内应答Kafka生产者会尝试n次重新发送消息。
acks1 默认
acks1 - Leader会将Record写到其本地日志中但会在不等待所有Follower的完全确认的情况下做出响应。在这种情况下如果Leader在确认记录后立即失败但在Follower复制记录之前失败则记录将丢失。 acks0 - 生产者根本不会等待服务器的任何确认。该记录将立即添加到套接字缓冲区中并视为已发送。在这种情况下不能保证服务器已收到记录。 acksall - 这意味着Leader将等待全套同步副本确认记录。这保证了只要至少一个同步副本仍处于活动状态记录就不会丢失。这是最有力的保证。这等效于acks -1设置。
如果生产者在规定的时间内并没有得到Kafka的Leader的Ack应答Kafka可以开启reties机制。
request.timeout.ms 30000 默认
retries 2147483647 默认
6、幂等性 HTTP/1.1中对幂等性的定义是一次和多次请求某一个资源对于资源本身应该具有同样的结果网络超时等问题除外。也就是说其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。 Methods can also have the property of “idempotence” in that (aside from error or expiration issues) the side-effects of N 0 identical requests is the same as for a single request. Kafka在0.11.0.0版本支持增加了对幂等的支持。幂等是针对生产者角度的特性。幂等可以保证上生产者发送的消息不会丢失而且不会重复。实现幂等的关键点就是服务端可以区分请求是否重复过滤掉重复的请求。要区分请求是否重复的有两点 唯一标识要想区分请求是否重复请求中就得有唯一标识。例如支付请求中订单号就是唯一标识 记录下已处理过的请求标识光有唯一标识还不够还需要记录下那些请求是已经处理过的这样当收到新的请求时用新请求中的标识和处理记录进行比较如果处理记录中有相同的标识说明是重复记录拒绝掉。 幂等又称为exactly once。要停止多次处理消息必须仅将其持久化到Kafka Topic中仅仅一次。在初始化期间kafka会给生产者生成一个唯一的ID称为Producer ID或PID。 PID和序列号与消息捆绑在一起然后发送给Broker。由于序列号从零开始并且单调递增因此仅当消息的序列号比该PID / TopicPartition对中最后提交的消息正好大1时Broker才会接受该消息。如果不是这种情况则Broker认定是生产者重新发送该消息。
enable.idempotence false 默认
注意:在使用幂等性的时候要求必须开启retriestrue和acksall
7、事务控制 Kafka的幂等性只能保证一条记录的在分区发送的原子性但是如果要保证多条记录多分区之间的完整性这个时候就需要开启kafk的事务操作。 在Kafka0.11.0.0除了引入的幂等性的概念同时也引入了事务的概念。通常Kafka的事务分为 生产者事务Only、消费者生产者事务。一般来说默认消费者消费的消息的级别是read_uncommited数据这有可能读取到事务失败的数据所有在开启生产者事务之后需要用户设置消费者的事务隔离级别。
isolation.level read_uncommitted 默认 该选项有两个值read_committed|read_uncommitted如果开始事务控制消费端必须将事务的隔离级别设置为read_committed 开启的生产者事务的时候只需要指定transactional.id属性即可一旦开启了事务默认生产者就已经开启了幂等性。但是要求transactional.id的取值必须是唯一的同一时刻只能有一个transactional.id存储在其他的将会被关闭。
8、数据同步机制 Kafka的Topic被分为多个分区分区是是按照Segments存储文件块。分区日志是存储在磁盘上的日志序列Kafka可以保证分区里的事件是有序的。其中Leader负责对应分区的读写、Follower负责同步分区的数据0.11 版本之前Kafka使用highwatermarker机制保证数据的同步但是基于highwatermarker的同步数据可能会导致数据的不一致或者是乱序。在Kafka数据同步有以下概念。 LEOlog end offset 标识的是每个分区中最后一条消息的下一个位置分区的每个副本都有自己的LEO. HW: high watermarker称为高水位线所有HW之前的的数据都理解是已经备份的,当所有节点都备份成功Leader会更新水位线。 ISR:In-sync-replicas,kafka的leader会维护一份处于同步的副本集和如果在replica.lag.time.max.ms时间内系统没有发送fetch请求或者已然在发送请求但是在该限定时间内没有赶上Leader的数据就被剔除ISR列表。在Kafka-0.9.0版本剔除replica.lag.max.messages消息个数限定因为这个会导致其他的Broker节点频繁的加入和退出ISR。
9、Kafka-Eagle
kafka-eagle源码地址
kafka-eagle官网--下载地址
Kafka-EagleKafka软件的监控 二、Maven项目测试
创建Maven项目
pom
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.lwz/groupIdartifactIdkafka_mq/artifactIdversion1.0-SNAPSHOT/versiondependenciesdependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion2.2.0/version/dependencydependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-streams/artifactIdversion2.2.0/version/dependencydependencygroupIdlog4j/groupIdartifactIdlog4j/artifactIdversion1.2.17/version/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdversion1.7.25/version/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-log4j12/artifactIdversion1.7.25/version/dependencydependencygroupIdorg.apache.commons/groupIdartifactIdcommons-lang3/artifactIdversion3.9/version/dependency/dependenciesbuildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-compiler-plugin/artifactIdconfigurationsource8/sourcetarget8/target/configuration/plugin/plugins/build/project
1、Topic API
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;public class TopicDMLDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {// 配置连接参数Properties props new Properties();props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.0.113:9092);KafkaAdminClient adminClient (KafkaAdminClient) KafkaAdminClient.create(props);// 创建TopicsListNewTopic newTopics Arrays.asList(new NewTopic(topic01, 1, (short) 1),new NewTopic(topic02, 1, (short) 1));CreateTopicsResult createTopics adminClient.createTopics(newTopics);createTopics.all().get();// 同步创建// 查询topicsKafkaFutureSetString nameFutures adminClient.listTopics().names();for (String name : nameFutures.get()) {System.out.println(name);}System.out.println(-----------------------------------------);// 删除TopicDeleteTopicsResult deleteTopics adminClient.deleteTopics(Arrays.asList(topic02));deleteTopics.all().get();// 同步删除nameFutures adminClient.listTopics().names();// 查询topicsfor (String name : nameFutures.get()) {System.out.println(name);}// 查看Topic详情DescribeTopicsResult describeTopics adminClient.describeTopics(Arrays.asList(topic01));MapString, TopicDescription tdm describeTopics.all().get();for (Map.EntryString, TopicDescription entry : tdm.entrySet()) {System.out.println(entry.getKey() \t entry.getValue());}adminClient.close();}
}
运行结果
INFO 2023-08-03 22:58:45 org.apache.kafka.clients.admin.AdminClientConfig - AdminClientConfig values: bootstrap.servers [192.168.0.113:9092]client.dns.lookup defaultclient.id connections.max.idle.ms 300000metadata.max.age.ms 300000metric.reporters []metrics.num.samples 2metrics.recording.level INFOmetrics.sample.window.ms 30000receive.buffer.bytes 65536reconnect.backoff.max.ms 1000reconnect.backoff.ms 50request.timeout.ms 120000retries 5retry.backoff.ms 100sasl.client.callback.handler.class nullsasl.jaas.config nullsasl.kerberos.kinit.cmd /usr/bin/kinitsasl.kerberos.min.time.before.relogin 60000sasl.kerberos.service.name nullsasl.kerberos.ticket.renew.jitter 0.05sasl.kerberos.ticket.renew.window.factor 0.8sasl.login.callback.handler.class nullsasl.login.class nullsasl.login.refresh.buffer.seconds 300sasl.login.refresh.min.period.seconds 60sasl.login.refresh.window.factor 0.8sasl.login.refresh.window.jitter 0.05sasl.mechanism GSSAPIsecurity.protocol PLAINTEXTsend.buffer.bytes 131072ssl.cipher.suites nullssl.enabled.protocols [TLSv1.2, TLSv1.1, TLSv1]ssl.endpoint.identification.algorithm httpsssl.key.password nullssl.keymanager.algorithm SunX509ssl.keystore.location nullssl.keystore.password nullssl.keystore.type JKSssl.protocol TLSssl.provider nullssl.secure.random.implementation nullssl.trustmanager.algorithm PKIXssl.truststore.location nullssl.truststore.password nullssl.truststore.type JKSINFO 2023-08-03 22:58:45 org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.2.0
INFO 2023-08-03 22:58:45 org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 05fcfde8f69b0349
topic01
topic02
example
-----------------------------------------
topic01
example
topic01 (nametopic01, internalfalse, partitions(partition0, leader192.168.0.113:9092 (id: 1001 rack: null), replicas192.168.0.113:9092 (id: 1001 rack: null), isr192.168.0.113:9092 (id: 1001 rack: null)))2、生产者消费者
生产者
import java.util.Properties;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;public class KafkaProducerDemo {public static void main(String[] args) throws InterruptedException {// 1.创建连接参数Properties props new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.0.113:9092);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 2.创建生产者KafkaProducerString, String producer new KafkaProducerString, String(props);// 3.发送消息队列for (Integer i 0; i 10; i) {Thread.sleep(100);ProducerRecordString, String record new ProducerRecord(topic01, key i, value i);producer.send(record);}producer.close();}
}
消费者
import java.time.Duration;
import java.util.Iterator;
import java.util.Properties;
import java.util.regex.Pattern;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;public class KafkaConsumerDemo {public static void main(String[] args) {// 1.创建Kafka链接参数Properties props new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.0.113:9092);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.GROUP_ID_CONFIG, group01);// 2.创建Topic消费者KafkaConsumerString, String consumer new KafkaConsumerString, String(props);// 3.订阅topic开头的消息队列consumer.subscribe(Pattern.compile(^topic.*$));while (true) {ConsumerRecordsString, String consumerRecords consumer.poll(Duration.ofSeconds(1));IteratorConsumerRecordString, String recordIterator consumerRecords.iterator();while (recordIterator.hasNext()) {ConsumerRecordString, String record recordIterator.next();String key record.key();String value record.value();long offset record.offset();int partition record.partition();System.out.println(key: key ,value: value ,partition: partition ,offset: offset);}}}
}
先启动消费者再启动生产者进行测试。 Spring-Kafka官网 分布式协调服务--zookeeper 不是你觉的悟到的东西给了你你也接不住
