文档章节

Kafka Producer Consumer

编程SHA
 编程SHA
发布于 01/18 11:21
字数 2124
阅读 17
收藏 0

Producer API

org.apache.kafka.clients.producer.KafkaProducer

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring,MyBatis,Netty源码分析的朋友可以加我的Java高级交流:854630135,群里有阿里大牛直播讲解技术,以及Java大型互联网技术的视频免费分享给大家。

 
1 props.put("bootstrap.servers", "192.168.1.128:9092"); 2 props.put("acks", "all"); 3 props.put("retries", 0); 4 props.put("batch.size", 16384); 5 props.put("linger.ms", 1); 6 props.put("buffer.memory", 33554432); 7 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 8 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 9 10 Producer<String, String> producer = new KafkaProducer<String, String>(props); 11 for (int i = 0; i < 10; i++) { 12 producer.send(new ProducerRecord<String, String>("foo", Integer.toString(i), Integer.toString(i)), new Callback() { 13 @Override 14 public void onCompletion(RecordMetadata recordMetadata, Exception e) { 15 if (null != e) { 16 e.printStackTrace(); 17 }else { 18 System.out.println("callback: " + recordMetadata.topic() + " " + recordMetadata.offset()); 19 } 20 } 21 }); 22 } 23 producer.close(); 

producer由一个缓冲池组成,这个缓冲池中维护着那些还没有被传送到服务器上的记录,而且有一个后台的I/O线程负责将这些记录转换为请求并将其传送到集群上去。

send()方法是异步的。当调用它以后就把记录放到buffer中并立即返回。这就允许生产者批量的发送记录。

acks配置项控制的是完成的标准,即什么样的请求被认为是完成了的。本例中其值设置的是"all"表示客户端会等待直到所有记录完全被提交,这是最慢的一种方式也是持久化最好的一种方式。

如果请求失败了,生产者可以自动重试。因为这里我们设置retries为0,所以它不重试。

生产者对每个分区都维护了一个buffers,其中放的是未被发送的记录。这些buffers的大小是通过batch.size配置项来控制的。

默认情况下,即使一个buffer还有未使用的空间(PS:buffer没满)也会立即发送。如果你想要减少请求的次数,你可以设置linger.ms为一个大于0的数。这个指令将告诉生产者在发送请求之前先等待多少毫秒,以希望能有更多的记录到达好填满buffer。在本例中,我们设置的是1毫秒,表示我们的请求将会延迟1毫秒发送,这样做是为了等待更多的记录到达,1毫秒之后即使buffer没有被填满,请求也会发送。(PS:稍微解释一下这段话,producer调用send()方法只是将记录放到buffer中,然后由一个后台线程将buffer中的记录传送到服务器上。这里所说的请求指的是从buffer到服务器。默认情况下记录被放到buffer以后立即被发送到服务器,为了减少请求服务器的次数,可以通过设置linger.ms,这个配置项表示等多少毫秒以后再发送,这样做是希望每次请求可以发送更多的记录,以此减少请求次数)

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring,MyBatis,Netty源码分析的朋友可以加我的Java高级交流:854630135,群里有阿里大牛直播讲解技术,以及Java大型互联网技术的视频免费分享给大家。

buffer.memory控制的是总的buffer内存数量

key.serializer 和 value.serializer表示怎样将key和value对象转成字节

从kafka 0.11开始,KafkaProducer支持两种模型:the idempotent producer and the transactional producer(幂等producer和事务producer)。幂等producer强调的是至少一次精确的投递。事务producer允许应用程序原子的发送消息到多个分区或者主题。

为了启用幂等性,必须将enable.idempotence这个配置的值设为true。如果你这样设置了,那么retries默认是Integer.MAX_VALUE,并且acks默认是all。为了利用幂等producer的优势,请避免应用程序级别的重新发送。

为了使用事务producer,你必须配置transactional.id。如果transactional.id被设置,幂等性自动被启用。

 
1 Properties props = new Properties(); 2 props.put("bootstrap.servers", "192.168.1.128:9092"); 3 props.put("transactional.id", "my-transactional-id"); 4 5 Producer<String, String> producer = new KafkaProducer<String, String>(props, new StringSerializer(), new StringSerializer()); 6 7 producer.initTransactions(); 8 9 try { 10 producer.beginTransaction(); 11 12 for (int i = 11; i < 20; i++) { 13 producer.send(new ProducerRecord<String, String>("bar", Integer.toString(i), Integer.toString(i))); 14 } 15 // This method will flush any unsent records before actually committing the transaction 16 producer.commitTransaction(); 17 } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { 18 producer.close(); 19 } catch (KafkaException e) { 20 // By calling producer.abortTransaction() upon receiving a KafkaException we can ensure 21 // that any successful writes are marked as aborted, hence keeping the transactional guarantees. 22 producer.abortTransaction(); 23 } 24 25 producer.close(); 

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring,MyBatis,Netty源码分析的朋友可以加我的Java高级交流:854630135,群里有阿里大牛直播讲解技术,以及Java大型互联网技术的视频免费分享给大家。

Consumer API

org.apache.kafka.clients.consumer.KafkaConsumer

Offsets and Consumer Position

对于分区中的每条记录,kafka维护一个数值偏移量。这个偏移量是分区中一条记录的唯一标识,同时也是消费者在分区中的位置。例如,一个消费者在分区中的位置是5,表示它已经消费了偏移量从0到4的记录,并且接下来它将消费偏移量为5的记录。相对于消费者用户来说,这里实际上有两个位置的概念。

消费者的position表示下一条将要消费的记录的offset。每次消费者通过调用poll(long)接收消息的时候这个position会自动增加。

committed position表示已经被存储的最后一个偏移量。消费者可以自动的周期性提交offsets,也可以通过调用提交API(e.g. commitSync and commitAsync)手动的提交position。

Consumer Groups and Topic Subscriptions

Kafka用"consumer groups"(消费者组)的概念来允许一组进程分开处理和消费记录。这些处理在同一个机器上进行,也可以在不同的机器上。同一个消费者组中的消费者实例有相同的group.id

组中的每个消费者可以动态设置它们想要订阅的主题列表。Kafka给每个订阅的消费者组都投递一份消息。这归功于消费者组中所有成员之间的均衡分区,以至于每个分区都可以被指定到组中精确的一个消费者。假设一个主题有4个分区,一个组中有2个消费者,那么每个消费者将处理2个分区。

消费者组中的成员是动态维护的:如果一个消费者处理失败了,那么分配给它的分区将会被重新分给组中其它消费者。

在概念上,你可以把一个消费者组想象成一个单个的逻辑订阅者,并且每个逻辑订阅者由多个进程组成。作为一个多订阅系统,Kafka天生就支持对于给定的主题可以有任意数量的消费者组。

Automatic Offset Committing

 
1 Properties props = new Properties(); 2 props.put("bootstrap.servers", "192.168.1.128:9092"); 3 props.put("group.id", "test"); 4 props.put("enable.auto.commit", "true"); 5 props.put("auto.commit.interval.ms", "1000"); 6 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 7 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 8 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); 9 consumer.subscribe(Arrays.asList("foo", "bar")); 10 while (true) { 11 ConsumerRecords<String, String> records = consumer.poll(100); 12 for (ConsumerRecord<String, String> record : records) { 13 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 14 } 15 } 

设置enable.auto.commit意味着自动提交已消费的记录的offset

Manual Offset Control

代替消费者周期性的提交已消费的offsets,用户可以控制什么时候记录被认为是已经消费并提交它们的offsets。

 
1 Properties props = new Properties(); 2 props.put("bootstrap.servers", "localhost:9092"); 3 props.put("group.id", "test"); 4 props.put("enable.auto.commit", "false"); 5 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 6 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 7 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 8 consumer.subscribe(Arrays.asList("foo", "bar")); 9 final int minBatchSize = 200; 10 List<ConsumerRecord<String, String>> buffer = new ArrayList<>(); 11 while (true) { 12 ConsumerRecords<String, String> records = consumer.poll(100); 13 for (ConsumerRecord<String, String> record : records) { 14 buffer.add(record); 15 } 16 if (buffer.size() >= minBatchSize) { 17 insertIntoDb(buffer); 18 consumer.commitSync(); 19 buffer.clear(); 20 } 21 } 

代码演示

服务器端

 

客户端

 

 

 

 

如果想学习Java工程化、高性能及分布式、深入浅出。微服务、Spring,MyBatis,Netty源码分析的朋友可以加我的Java高级交流:854630135,群里有阿里大牛直播讲解技术,以及Java大型互联网技术的视频免费分享给大家。

© 著作权归作者所有

共有 人打赏支持
编程SHA
粉丝 9
博文 106
码字总数 259470
作品 0
长沙
私信 提问
kafka的安装和使用

一、安装 1、安装scala 解压 tar -zxvf scala-2.10.4.tgz 2、安装kafka 解压 tar -zxvf kafka_2.11-0.9.0.1.tgz 二、修改kafka/config下的server.properties配置文件如下内容 broker.id=1 不......

xiaozhou18
2016/12/15
155
0
Kafka部署

简介 kafka (官网地址:http://kafka.apache.org)是一款分布式消息发布和订阅的系统,具有高性能和高吞吐率。 i. 消息的发布(publish)称作producer,消息的订阅(subscribe)称作consume...

follitude
2016/09/21
17
0
Kafka入门简介

本文简单的介绍下kafka,主要包含以下部分: 什么是Kafka Kafka的基本概念 Kafka分布式架构 配置单机版Kafka 实验一:kafka-python实现生产者消费者 实验二:消费组实现容错性机制 实验三:o...

强哥
2017/12/06
0
0
Python操作分布式流处理系统Kafka

專 欄 ❈强哥,Python中文社区专栏作者,曾供职于摩根士丹利(Morgan Stanley)和eBay。❈ 什么是Kafka Kafka是一个分布式流处理系统,流处理系统使它可以像消息队列一样publish或者subscri...

bf02jgtrs00xktcx
2017/12/08
0
0
apache kafka消息服务

apache kafka中国社区QQ群:162272557 apache kafka参考 http://kafka.apache.org/documentation.html 消息队列分类: 点对点: 消息生产者生产消息发送到queue中,然后消息消费者从queue中取...

老先生二号
2017/08/06
0
0

没有更多内容

加载失败,请刷新页面

加载更多

PHP is_numeric 检测变量是否为数字或数字字符串

bool is_numeric ( mixed $var ) 如果 var 是数字和数字字符串则返回 TRUE,否则返回 FALSE。 For example 1: <?php $v = is_numeric ('58635272821786587286382824657568871098287278......

james_laughing
28分钟前
1
0
聊聊flink的NetworkEnvironmentConfiguration

序 本文主要研究一下flink的NetworkEnvironmentConfiguration NetworkEnvironmentConfiguration flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnv......

go4it
今天
4
0
极路由4刷OpenWrt(LEDE)

申请开发者模式 登录路由器后台 进入插件中间 选中路由器信息 申请开发者权限 安装开发者插件 刷 Bootloader 下载 极路由4 的 Breed 通过SSH登录路由器 (必须安装开发者插件,端口 1022 账号...

dingdayu
今天
3
0
浅淡个人学习嵌入式Linux过程

我专业是电子信息工程,在初入大学的时候,我们的班主任便要我们多多去了解一些关于电子方面的知识。后来我了解到了嵌入式,继而了解到了嵌入式Linux。其实我们学习linux差不多就学习linux内...

linux-tao
今天
7
0
SpringBoot使用GraphQL简单学习-1

官网 一、GraphQL简介 1.GraphQL是什么? GraphQL 既是一种用于 API 的查询语言也是一个满足你数据查询的运行时。 GraphQL 对你的 API 中的数据提供了一套易于理解的完整描述,使得客户端能够...

wind2012
今天
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部