java kafka client_Kafka Client API 基本使用
之前讲过了[Kafka基本概念及原理][1],这次我们来看看Kafka Client的API。要使用Kafka Client的API,首先需要先部署Kafka集群,部署过程请参见[官网][2]。然后在项目中添加Kafka Client的依赖,在这里我们使用0.10.0.1版本:
org.apache.kafka
kafka-clients
0.10.0.1
Kafka有以下四个核心API:
Producer
Consumer
Streams
Connect
这里我们来讲解比较基础的Producer和Consumer。假设读者已经按照[官网][2]在本机配置好了Kafka服务,并创建了名为“test”的topic。
Producer API
Producer用来向Kafka集群中发布消息记录的Kafka客户端。Producer是线程安全的,并且通常来讲,在多个线程间共享一个producer要比每个线程都创建一个producer速度更快。Producer的API相对比较简单,下面给出一个较为简单的API实例:
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class ProducerDemo {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord("test", Integer.toString(i), Integer.toString(i)));
producer.close();
}
}
Producer由一个持有未发送消息记录的资源池和一个用来向Kafka集群发送消息记录的后台IO线程组成。使用后未关闭producer将导致这些资源泄露。
send方法是异步的。当它被调用时,它会将消息记录添加到待发送缓冲区并立即返回。使用这种方式可以使生产者聚集一批消息记录后一起发送,从而提高效率。
ack 配置项用来控制producer要求leader确认多少消息后返回调用成功。当值为0时producer不需要等待任何确认消息。当值为1时只需要等待leader确认。当值为-1或all时需要全部ISR集合返回确认才可以返回成功。
当 retries > 0 时,如果发送失败,会自动尝试重新发送数据。发送次数为retries设置的值。
buffer.memory、batch.size、linger.ms三个参数用来控制缓冲区大小和延迟发送时间,具体含义可以参考官方文档的配置。
bootstrap.servers 配置项处需要填写我们要发送到的Kafka集群地址。
key.serializer 和 value.serializer 指定使用什么序列化方式将用户提供的key和value进行序列化。
运行此程序,在$KAFKA_HOME目录下运行:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
可以看到写入Kafka的消息。
Consumer API
Consumer的API分为High-level API和Low-level API。前者提供了高度抽象的API,使用起来简单、方便。因此本文将主要讲述High-level API。Low-level API提供了更强的控制能力,但使用起来较为繁琐。下面我们来看一种最简单的方式,自动确认offset:
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class AutoCommitConsumerDemo {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
@SuppressWarnings("resource")
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test"));
while (true) {
ConsumerRecords records = consumer.poll(100);
for (ConsumerRecord record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}
}
bootstrap.servers配置项指定了consumer需要连接的服务器集群。多台服务器用“,”分隔。这个配置项就算只配置了一台server的地址,也可以通过这一台server发现集群中的其他服务器。但是为了避免这台server挂掉引发单点问题,所以把所有服务器地址列举上去是一个比较好的选择。
enable.auto.commit配置项指定了提交offset的方式为自动提交,auto.commit.interval.ms配置项配置了每次自动提交的时间间隔。
group.id即消费者组标签,本例中消费者组的名称为test。不了解消费者组的概念可以看我的这篇文章:[Kafka基本概念及原理][1]。
自动提交offset的方式非常简单,但多数情况下,我们不会使用自动提交的方式。因为不论从Kafka集群中拉取的数据是否被处理成功,offset都会被更新,也就是如果处理过程中出现错误可能会出现数据丢失的情况。所以多数情况下我们会选择手动提交方式:
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
public class ManualCommitConsumerDemo {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
@SuppressWarnings("resource")
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo", "bar"));
final int minBatchSize = 200;
List> buffer = new ArrayList<>();
while (true) {
ConsumerRecords records = consumer.poll(100);
for (ConsumerRecord record : records) {
buffer.add(record);
}
if (buffer.size() >= minBatchSize) {
insertIntoDb(buffer);
consumer.commitSync();
buffer.clear();
}
}
}
private static void insertIntoDb(List> buffer) {
// Insert into db
}
}
从对比自动提交offset的代码,我们看到 enable.auto.commit 配置项被设置为false,代表手动提交。代码中定义了一个ConsumerRecord的列表作为缓冲,当缓冲中的数据大于200条时,才一次性插入数据库中,并手动提交offset。这样,只有当数据成功插入数据库时才会更新offset,从而保证了数据不丢失。但如果在数据插入数据库后和手动提交offset之间这段时间(虽然很短,但也是有可能的)程序崩溃或服务器down机,那么再次启动会导致重复消费。所以这种方式其实是提供了 at least once 语义。
另外,consumer并不是线程安全的,所以在进行多线程操作时需要在每个线程实例化一个consumer。如果需要跨线程使用consumer,需要进行手动同步。
[1]:http://www.jianshu.com/p/97011dab6c56
[2]:http://kafka.apache.org/quickstart
java kafka client_Kafka Client API 基本使用相关推荐
- java kafka client_Kafka Java Client基本使用及整合SpringBoot
kafka-clients 添加依赖 org.apache.kafka kafka-clients 2.5.0 消费者 Consumer 代码上总体可以分为三部分:消费者的配置消费者的配置在 org. ...
- java kafka api_kafka java API的使用
Kafka包含四种核心的API: 1.Producer API支持应用将数据流发送到Kafka集群的主题 2.Consumer API支持应用从Kafka集群的主题中读取数据流 3.Streams A ...
- Java 9 揭秘(14. HTTP/2 Client API)
Tips 做一个终身学习的人. 在此章中,主要介绍以下内容: 什么是HTTP/2 Client API 如何创建HTTP客户端 如何使HTTP请求 如何接收HTTP响应 如何创建WebSocket的e ...
- mq 接口 java_Rabbitmq Java Client Api详解
AMQP AMQP协议是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现. 基础概念快速入门 每个rabbitmq-server叫做一个Broker,等着tcp连接进入. 在rabbi ...
- Memcached Java Client API详解
针对Memcached官方网站提供的java_memcached-release_2.0.1版本进行阅读分析,Memcached Java客户端lib库主要提供的调用类是SockIOPool和MemC ...
- HIDL示例-JAVA服务创建-Client验证-Android10.0 HwBinder通信原理(四)
摘要:本节主要来讲解Android10.0 JAVA层的HIDL服务创建和JAVA层的Client验证 阅读本文大约需要花费15分钟. 文章首发微信公众号:IngresGe 专注于Android系统级 ...
- java consumer.poll_kafka消费者API consumer.poll()没有错误,没有异常,只是阻止
我正在学习遵循Apache kafka文档的kafka . 我用默认配置启动它 . bin/zookeeper-server-start.sh config/zookeeper.properties ...
- 使用RESTful Client API进行GET / POST
互联网上有很多如何使用RESTful Client API的东西. 这些是基础. 但是,尽管该主题看起来微不足道,但仍然存在一些障碍,尤其是对于初学者而言. 在这篇文章中,我将尝试总结我的专业知识,以 ...
- scala rest_使用路标的Scala和Java的Twitter REST API
scala rest 如果您已经阅读了此博客上的其他文章,您可能会知道我喜欢创建各种数据集的可视化. 我刚刚开始一个小项目,在这里我想可视化来自Twitter的一些数据. 为此,我想直接从Twitte ...
最新文章
- 简易快速的开发,需要一个快速开发平台来支持
- php项目webpack打包,Vue项目webpack打包部署时Tomcat刷新报404错误问题如何处理
- 5233杨光--第十一周学习总结
- 梯度下降法(一)入门
- 机器学习物语(3):回归问题
- matlab中数组创建方法
- 在javafx中界面主题_最小的JavaFX演示文稿(在JavaFX中)
- 【HDU - 1564 】Play a game (博弈问题,找规律,奇偶博弈)
- 麒麟芯片或“绝版”,华为多系列手机涨价​;一加回应“刘作虎回归OPPO”;DBeaver 7.2 发布| 极客头条
- c gui qt 4编程第二版_一本专门学习PyQt5 GUI编程的书
- Linux命令行下WEP密码破解(通用,也可非BT平台)
- 网管员的任务与职责漫谈
- 安装系统出现Winload.exe错误0xc000000e解决方法
- 集赞生成器:朋友圈集赞不求人
- word文字怎么竖向排列_word怎么把字体竖着 如何在WORD中竖排文字
- 不同vlan实现互通
- linux26内核,基于AT91RM9200与LINUX2.6.26内核的嵌入式平台开发全过程
- 扫除知识共享障碍,天翎知识文档管理系统+群晖NAS一体化解决方案
- 树莓派Android Things物联网开发:GitHub案例程序汇总
- Mac OSX常用软件
热门文章
- Scrapy周期性爬取(解决Unknown command: crawl报错)
- 利用oc门或od门实现线与_福师《数字逻辑》在线作业二答案
- flutter 图解_Flutter 54: 图解基本生命周期
- php反序列化总结与学习
- android -------- 解决NDK开发中的 Method 'NewStringUTF' could not be resolved
- MySQL筛选重复数据
- hihoCoder #1162 : 骨牌覆盖问题·三 (矩阵快速幂,DP)
- Codeforces Round #256 (Div. 2)
- Effective C# Item33:限制类型的可见性
- 庆祝西安.NET俱乐部成立,请申请加入的朋友在这里报到