Kafka是一种分布式的基于发布/订阅的消息系统,它的高吞吐量、灵活的offset是其它消息系统所没有的。
kafka优缺点
优点:
- 基于磁盘的数据存储
- 高伸缩性
- 高性能
- 应用场景 : 收集指标和日志 提交日志 流处理
缺点:
- 运维难度大
- 偶尔有数据混乱的情况
- 对zookeeper强依赖
- 多副本模式下对带宽有一定要求
一、下载Kafka
访问 Apache Kafka 下载最新的 Kafka 版本并解压
解压后如下:
kafka依赖于zookeeper,但是我们不必去下载zk,已经内置了一个
二、启动kafka
2.1 启动kafka前得先启动zk
E:\study\kafka_2.12-3.4.0\bin\windows>zookeeper-server-start.bat ../../config/zookeeper.properties
2.2 启动kafka
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-server-start.bat ../../config/server.properties
三、操作Kafka
3.1 创建 Kafka 主题(Topic)
我们可执行如下命令用于查看已经存在的主题:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --list --bootstrap-server localhost:9092
用于刚起,所以并不存在任何主题 。
参数描述–-bootstrap-server <String: server toconnect to>连接的 Kafka Broker 主机名称和端口号–-topic <String: topic>操作的 topic 名称–-create创建主题–-delete 删除主题–-alter 修改主题-–list 查看所有主题–-describe查看主题详细描述-–partitions <Integer: # of partitions> 设置分区数-–replication-factor<Integer: replication factor>设置分区副本-–config <String: name=value>更新系统默认的配置
我们来创建一个名为
quick_start_topic
的主题:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --create --topic quick_start_topic --bootstrap-server localhost:9092
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic quick_start_topic.
再来查看存在的主题:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --list --bootstrap-server localhost:9092
quick_start_topic
查看详细信息:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-topics.bat --describe --topic quick_start_topic --bootstrap-server localhost:9092
Topic: quick_start_topic TopicId: TYIVCN3zSaWSvKDhhPvZRQ PartitionCount: 1 ReplicationFactor: 1 Configs:
Topic: quick_start_topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0
3.2 将信息写入主题(Topic)
Kafka 客户端通过网络与 Kafka 代理通信以写入(或读取)事件。一旦收到,代理将以持久和容错的方式存储事件,只要您需要——甚至永远。
运行控制台生产者客户端以将一些事件写入到指定主题。默认情况下,输入的每一行都会产生一个单独的事件写入到主题
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-producer.bat --topic quick_start_topic --bootstrap-server localhost:9092
>this is the first message
>this is the second message
>终止批处理操作吗(Y/N)? y
ctrl +c 按照提示实在输入y退出命令交互模式
此时我们已经成功的将两条消息发布到了kafla的 quick_start_topic主题中
参数描述–bootstrap-server <String: server toconnect to>连接的 Kafka Broker 主机名称和端口号。–topic <String: topic>操作的 topic 名称。
3.3 读取信息
打开另一个终端会话并运行消费者客户端以读取刚刚创建的事件:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message
参数描述–bootstrap-server <String: server toconnect to>连接的 Kafka Broker 主机名称和端口号–topic <String: topic>操作的 topic 名称–from-beginning从头开始消费。–group <String: consumer group id>指定消费者组名称。
命令行的简单演示到此结束,我们来试下java demo
四、Java实践-三种发送消息的方式
正式进入生产者代码实践之前,首先列举出生产者方大致的参数列表如下:
参数解释说明bootstrap.servers生产者连接集群所需的 broker 地 址 清 单 。 例 如hadoop102:9092,hadoop103:9092,hadoop104:9092,可以设置 1 个或者多个,中间用逗号隔开。注意这里并非需要所有的 broker 地址,因为生产者从给定的 broker里查找到其他 broker 信息。key.serializer 和 value.serializer指定发送消息的 key 和 value 的序列化类型。一定要写全类名。buffer.memoryRecordAccumulator 缓冲区总大小,默认 32m。batch.size缓冲区一批数据最大值,默认 16k。适当增加该值,可以提高吞吐量,但是如果该值设置太大,会导致数据传输延迟增加。linger.ms如果数据迟迟未达到 batch.size,sender 等待 linger.time之后就会发送数据。单位 ms,默认值是 0ms,表示没有延迟。生产环境建议该值大小为 5-100ms 之间。acks0:生产者发送过来的数据,不需要等数据落盘应答。1:生产者发送过来的数据,Leader 收到数据后应答。-1(all):生产者发送过来的数据,Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1,-1 和all 是等价的。max.in.flight.requests.per.connection允许最多没有返回 ack 的次数,默认为 5,开启幂等性要保证该值是 1-5 的数字。retries当消息发送出现错误的时候,系统会重发消息。retries表示重试次数。默认是 int 最大值,2147483647。如果设置了重试,还想保证消息的有序性,需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候,其他的消息可能发送成功了retry.backoff.ms两次重试之间的时间间隔,默认是 100ms。enable.idempotence是否开启幂等性,默认 true,开启幂等性。compression.type生产者发送的所有数据的压缩方式。默认是 none,也就是不压缩。支持压缩类型:none、gzip、snappy、lz4 和 zstd。
使用boot搭建项目并勾选kafka或者手动引入相关依赖
引入kafka依赖
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
发送消息前,我们使用命令开启消费者监听
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
4.1 异步发送-无回调
发送并忘记的方式本质上也是一种异步的方式,只是它不会获取消息发送的返回结果,这种方式的吞吐量是最高的,但是无法保证消息的可靠性:
package com.cjian.kafkademo.demo;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
* @Author: cjian
* @Date: 2023/5/16 20:01
* @Des: 异步发送-无回调
*/
public class AsyncProducerNoCallback {
public static void main(String[] args) {
// 1. 创建kafka生产者配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息:bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// key,value 序列化(必须):key.serializer,value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
// 4. 调用 send 方法,异步发送消息,无回调接口
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "producerAsyncMessage-" + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
}
运行后,观察“消费者”窗口:
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
4.2 异步发送-有回调
异步发送+回调函数(消息以异步的方式发送,通过回调函数返回消息发送成功/失败)
在调用send方法发送消息的同时,指定一个回调函数,服务器在返回响应时会调用该回调函数,通过回调函数能够对异常情况进行处理,当调用了回调函数时,只有回调函数执行完毕生产者才会结束,否则一直会阻塞
package com.cjian.kafkademo.demo;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
/**
* @Author: cjian
* @Date: 2023/5/16 20:09
* @Des: 异步发送-有回调
*/
public class AsyncProducerWithCallback {
public static void main(String[] args) throws InterruptedException {
// 1. 创建Kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// 3. key 序列化 key.serializer,value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 4. value 序列化 value.serializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 5. 创建kafka生产者对象
KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "asyncCallbackMessage-" + i), new Callback() {
// 该方法在Producer 收到 ack 时调用,为异步调用
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e == null) {
// 没有异常,输出信息到控制台
System.out.println("topic:" + recordMetadata.topic() + " -> " + " 分区 " + recordMetadata.partition());
}else {
e.printStackTrace();
}
}
});
// 延迟发送是否会发送到不同分区
Thread.sleep(2);
}
// 5. 关闭资源
kafkaProducer.close();
}
}
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4
4.3 同步发送
以同步的方式发送消息时,一条一条的发送,对每条消息返回的结果判断, 可以明确地知道每条消息的发送情况,但是由于同步的方式会阻塞,只有当消息通过get返回future对象时,才会继续下一条消息的发送
package com.cjian.kafkademo.demo;
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.clients.producer.RecordMetadata;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
/**
* @Author: cjian
* @Date: 2023/5/16 20:12
* @Des: 同步发送
*/
public class SyncProducer {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 1. 创建kafka生产者配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息:bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// key,value 序列化(必须):key.serializer,value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
// 4. 调用 send 方法,发送消息
for (int i = 0; i < 5; i++) {
RecordMetadata recordMetadata = kafkaProducer.send(new ProducerRecord<>("quick_start_topic", "syncMessage-" + i)).get();
System.out.println("send response result,topic:" + recordMetadata.topic() + ", " + " 分区:" + recordMetadata.partition());
}
// 5. 关闭资源
kafkaProducer.close();
}
}
E:\study\kafka_2.12-3.4.0\bin\windows>kafka-console-consumer.bat --topic quick_start_topic --from-beginning --bootstrap-server localhost:9092
this is the first message
this is the second message
producerAsyncMessage-0
producerAsyncMessage-1
producerAsyncMessage-2
producerAsyncMessage-3
producerAsyncMessage-4
asyncCallbackMessage-0
asyncCallbackMessage-1
asyncCallbackMessage-2
asyncCallbackMessage-3
asyncCallbackMessage-4
syncMessage-0
syncMessage-1
syncMessage-2
syncMessage-3
syncMessage-4
版权归原作者 cj_eryue 所有, 如有侵权,请联系我们删除。