kafka急速入门与核心API解析
Posted 尘恍若梦
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了kafka急速入门与核心API解析相关的知识,希望对你有一定的参考价值。
kafka急速入门与核心API
kafka急速入门与核心API解析
kafka环境安装
上一节课我们已经对kafka的基本概念、核心思想有了一定的了解和认知,并且掌握了kafka在实际工作中的一些主要的应用场景。那么接下来,我们就一起快速进入kafka的安装吧。
-
kafka下载地址:http://kafka.apache.org/downloads.html
-
kafka安装步骤:首先kafka安装需要依赖与zookeeper,所以小伙伴们先准备好zookeeper环境(三个节点即可),然后我们来一起构建kafka broker。
## 解压命令: tar -zxvf kafka_2.12-2.1.0.tgz -C /usr/local/ ## 改名命令: mv kafka_2.12-2.1.0/ kafka_2.12 ## 进入解压后的目录,修改server.properties文件: vim /usr/local/kafka_2.12/config/server.properties ## 修改配置: broker.id=0 port=9092 host.name=192.168.11.51 advertised.host.name=192.168.11.51 log.dirs=/usr/local/kafka_2.12/kafka-logs num.partitions=2 zookeeper.connect=192.168.11.111:2181,192.168.11.112:2181,192.168.11.113:2181 ## 建立日志文件夹: mkdir /usr/local/kafka_2.12/kafka-logs ##启动kafka: /usr/local/kafka_2.12/bin/kafka-server-start.sh /usr/local/kafka_2.12/config/server.properties &
kafka常用命令
我们接下来一起了解几个非常重要的命令,通过这些命令我们对kafka topic partition 进行查看和操作。
-
常用命令:
## 简单操作: #(1)创建topic主题命令:(创建名为test的topic, 1个分区分别存放数据,数据备份总共1份) kafka-topics.sh --zookeeper 192.168.11.111:2181 --create --topic topic1 --partitions 1 --replication-factor 1 ## --zookeeper 为zk服务列表 ## --create 命令后 --topic 为创建topic 并指定 topic name ## --partitions 为指定分区数量 ## --replication-factor 为指定副本集数量 #(2)查看topic列表命令: kafka-topics.sh --zookeeper 192.168.11.111:2181 --list #(3)kafka命令发送数据:(然后我们就可以编写数据发送出去了) kafka-console-producer.sh --broker-list 192.168.11.51:9092 --topic topic1 #(4)kafka命令接受数据:(然后我们就可以看到消费的信息了) kafka-console-consumer.sh --bootstrap-server 192.168.11.51:9092 --topic topic1 --from-beginning #(5)删除topic命令: kafka-topics.sh --zookeeper 192.168.11.111:2181 --delete --topic topic1 #(6)kafka查看消费进度:(当我们需要查看一个消费者组的消费进度时,则使用下面的命令) kafka-consumer-groups.sh --bootstrap-server 192.168.11.51:9092 --describe --group group1 ## --describe --group 为订阅组, 后面指定 group name
急速入门
下面我们一起使用kafka最基本的API来对kafka进行操作!
-
kafka依赖包:
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.12</artifactId> </dependency>
-
kafka生产者:
package com.bfxy.mix.kafka; import java.util.Properties; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import com.alibaba.fastjson.JSON; public class CollectKafkaProducer { // 创建一个kafka生产者 private final KafkaProducer<String, String> producer; // 定义一个成员变量为topic private final String topic; // 初始化kafka的配置文件和实例:Properties & KafkaProducer public CollectKafkaProducer(String topic) { Properties props = new Properties(); // 配置broker地址 props.put("bootstrap.servers", "192.168.11.51:9092"); // 定义一个 client.id props.put("client.id", "demo-producer-test"); // 其他配置项: // props.put("batch.size", 16384); //16KB -> 满足16KB发送批量消息 // props.put("linger.ms", 10); //10ms -> 满足10ms时间间隔发送批量消息 // props.put("buffer.memory", 33554432); //32M -> 缓存提性能 // kafka 序列化配置: props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 创建 KafkaProducer 与 接收 topic this.producer = new KafkaProducer<>(props); this.topic = topic; } // 发送消息 (同步或者异步) public void send(Object message, boolean syncSend) throws InterruptedException { try { // 同步发送 if(syncSend) { producer.send(new ProducerRecord<>(topic, JSON.toJSONString(message))); } // 异步发送(callback实现回调监听) else { producer.send(new ProducerRecord<>(topic, JSON.toJSONString(message)), new Callback() { @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (e != null) { System.err.println("Unable to write to Kafka in CollectKafkaProducer [" + topic + "] exception: " + e); } } }); } } catch (Exception e) { e.printStackTrace(); } } // 关闭producer public void close() { producer.close(); }