点击关注，与你共同成长！
Kafka 是由 LinkedIn公司开发的，它是一个分布式的，支持多分区、多副本，基于 Zookeeper 的分布式消息流平台，它同时也是一款开源的基于发布订阅模式的消息引擎系统。
(资料图片仅供参考)
以上介绍参考Kafka官方文档。
Kafka核心APIKafka有4个核心API
应用程序使用Producer API发布消息到1个或多个Topics中；应用程序使用ConsumerAPI来订阅1个或多个Topics，并处理产生的消息；应用程序使用Streams API充当一个流处理器，从1个或多个Topics消费输入流，并产生一个输出流到1个或多个Topics，有效地将输入流转换到输出流；Connector API允许构建或运行可重复使用的生产者或消费者，将Topic链接到现有的应用程序或数据系统。Kafka为何如此之快Kafka 实现了零拷贝原理来快速移动数据，避免了内核之间的切换。Kafka 可以将数据记录分批发送，从生产者到文件系统（Kafka 主题日志）到消费者，可以端到端的查看这些批次的数据。批处理能够进行更有效的数据压缩并减少 I/O 延迟，Kafka 采取顺序写入磁盘的方式，避免了随机磁盘寻址的浪费。
总结一下其实就是四个要点：
顺序读写；零拷贝；消息压缩；分批发送。案例项目创建：
Dependencies：
构建工具为Maven，Maven的依赖如下：
Kafka Producerorg.apache.kafka kafka_2.12 1.0.0 provided org.apache.kafka kafka-clients 1.0.0 org.apache.kafka kafka-streams 1.0.0
packagecn.com.codingce.module;importjava.util.Properties;importjava.util.Random;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;publicclassProducer{//定义主题publicstaticStringtopic="codingce_test";publicstaticvoidmain(String[]args)throwsInterruptedException{Propertiesp=newProperties();//bootstrap.servers:kafka的地址,多个地址用逗号分割p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.31.150:9092");//acks:消息的确认机制，默认值是0.acks=0:如果设置为0，生产者不会等待kafka的响应;acks=1:这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应//acks=all:这个配置意味着leader会等待所有的follower同步完成.这个确保消息不会丢失,除非kafka集群中所有机器挂掉.这是最强的可用性保证.p.put("acks","all");//retries:配置为大于0的值的话,客户端会在消息发送失败时重新发送.p.put("retries",0);//batch.size:当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求.这会提高client和生产者的效率.p.put("batch.size",16384);//key.serializer:键序列化，默认org.apache.kafka.common.serialization.StringDeserializer.p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);//value.deserializer:值序列化，默认org.apache.kafka.common.serialization.StringDeserializer.p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);KafkaProducerkafkaProducer=newKafkaProducer<>(p);try{do{Stringmsg="后端码匠,"+newRandom().nextInt(100);ProducerRecordrecord=newProducerRecord<>(topic,msg);kafkaProducer.send(record);System.out.println("======消息发送成功:"+msg+"======");Thread.sleep(1000L);}while(true);}finally{kafkaProducer.close();}}} output
======消息发送成功:后端码匠,97============消息发送成功:后端码匠,35============消息发送成功:后端码匠,81============消息发送成功:后端码匠,46============消息发送成功:后端码匠,62============消息发送成功:后端码匠,53============消息发送成功:后端码匠,42============消息发送成功:后端码匠,56============消息发送成功:后端码匠,99============消息发送成功:后端码匠,46============消息发送成功:后端码匠,49============消息发送成功:后端码匠,35============消息发送成功:后端码匠,17============消息发送成功:后端码匠,78============消息发送成功:后端码匠,66============消息发送成功:后端码匠,4============消息发送成功:后端码匠,9============消息发送成功:后端码匠,69============消息发送成功:后端码匠,52============消息发送成功:后端码匠,2============消息发送成功:后端码匠,8============消息发送成功:后端码匠,86============消息发送成功:后端码匠,12============消息发送成功:后端码匠,67============消息发送成功:后端码匠,91============消息发送成功:后端码匠,8============消息发送成功:后端码匠,56============消息发送成功:后端码匠,89============消息发送成功:后端码匠,37============消息发送成功:后端码匠,39============消息发送成功:后端码匠,71======Kafka Consumer
packagecn.com.codingce.module;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.clients.consumer.ConsumerRecords;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.common.serialization.StringDeserializer;importjava.util.Collections;importjava.util.Properties;publicclassConsumer{privatestaticfinalStringGROUPID="codingce_consumer_a";publicstaticvoidmain(String[]args){Propertiesp=newProperties();//bootstrap.servers:kafka的地址,多个地址用逗号分割p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.31.150:9092");//消费者所属的分组id,组名不同组名可以重复消费.例如你先使用了组名A消费了Kafka的1000条数据,但是你还想再次进行消费这1000条数据,//并且不想重新去产生,那么这里你只需要更改组名就可以重复消费了.p.put(ConsumerConfig.GROUP_ID_CONFIG,GROUPID);//是否自动提交,默认为true.p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");//从poll(拉)的回话处理时长p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");//超时时间p.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"30000");//一次最大拉取的条数p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1000);//消费规则,默认earliestp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//key.serializer:键序列化,默认org.apache.kafka.common.serialization.StringDeserializer.p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);//value.deserializer:值序列化,默认org.apache.kafka.common.serialization.StringDeserializer.p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);KafkaConsumerkafkaConsumer=newKafkaConsumer<>(p);//订阅消息kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));do{//订阅之后,再从kafka中拉取数据ConsumerRecordsrecords=kafkaConsumer.poll(100);for(ConsumerRecordrecord:records){System.out.printf("-----topic:%s,offset:%d,消息:%s-----\n",record.topic(),record.offset(),record.value());}}while(true);}} output
-----topic:codingce_test,offset:289,消息:后端码匠,97----------topic:codingce_test,offset:290,消息:后端码匠,35----------topic:codingce_test,offset:291,消息:后端码匠,81----------topic:codingce_test,offset:292,消息:后端码匠,46----------topic:codingce_test,offset:293,消息:后端码匠,62----------topic:codingce_test,offset:294,消息:后端码匠,53----------topic:codingce_test,offset:295,消息:后端码匠,42----------topic:codingce_test,offset:296,消息:后端码匠,56----------topic:codingce_test,offset:297,消息:后端码匠,99----------topic:codingce_test,offset:298,消息:后端码匠,46----------topic:codingce_test,offset:299,消息:后端码匠,49----------topic:codingce_test,offset:300,消息:后端码匠,35----------topic:codingce_test,offset:301,消息:后端码匠,17----------topic:codingce_test,offset:302,消息:后端码匠,78----------topic:codingce_test,offset:303,消息:后端码匠,66----------topic:codingce_test,offset:304,消息:后端码匠,4----------topic:codingce_test,offset:305,消息:后端码匠,9----------topic:codingce_test,offset:306,消息:后端码匠,69----------topic:codingce_test,offset:307,消息:后端码匠,52----------topic:codingce_test,offset:308,消息:后端码匠,2----------topic:codingce_test,offset:309,消息:后端码匠,8----------topic:codingce_test,offset:310,消息:后端码匠,86----------topic:codingce_test,offset:311,消息:后端码匠,12----------topic:codingce_test,offset:312,消息:后端码匠,67----------topic:codingce_test,offset:313,消息:后端码匠,91----------topic:codingce_test,offset:314,消息:后端码匠,8----------topic:codingce_test,offset:315,消息:后端码匠,56----------topic:codingce_test,offset:316,消息:后端码匠,89----------topic:codingce_test,offset:317,消息:后端码匠,37----------topic:codingce_test,offset:318,消息:后端码匠,39----------topic:codingce_test,offset:319,消息:后端码匠,71-----
本次采用Docker 搭建的单机 Kafka、Zookeeper，Kafka介绍参考官方文档：http://kafka.apache.org/intro
项目地址：https://gitee.com/codingce/codingce-leetcode
【Java】线程池梳理
【C++】const关键字
【Java】原子类
以上，便是今天的分享，希望大家喜欢，觉得内容不错的，欢迎「分享」「赞」或者点击「在看」支持，谢谢各位。