# Kafka_深入探秘者（1）：初识 kafka

Kafka_深入探秘者（1）：初识 kafka

一、kafka 特性

1、Kafka ：最初是由 Linkedln 公司采用 Scala 语言开发的一个多分区、多副本并且基于 ZooKeeper 协调的分布式消息系统，现在已经捐献给了 Apache 基金会。目前 Kafka 已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

2、Apache Kafka 是一个分布式的发布-订阅消,息系统，能够支撑海量数据的数据传递。在离线和实时的消息处理业务系统中，Kafka 都有广泛的应用。Kafka 将消息持久化到磁盘中，并对消息创建了备份保证了数据的安全。Kafka 在保证了较高的处理速度的同时，又能保证数据处理的低延迟和数据的零丢失。

3、示例图：

在这里插入图片描述

4、kafka 特性

  • (1)高吞吐量、低延迟: kafka 每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个主题可以分多个分区,消费组对分区进行消费操作;

  • (2)可扩展性: kafka 集群支持热扩展;

  • (3)持久性、可靠性: 消息被持久化到本地磁盘，并且支持数据备份防止数据丢失;

  • (4)容错性: 允许集群中节点失败(若副本数量为n,则允许n-1个节点失败);

  • (5)高并发: 支持数千个客户端同时读写;

5、kafka 使用场景

  • (1)日志收集:

一个公司可以用 Kafka 可以收集各种服务的 l0g，通过kafka以统一接口服务的方式开放给各种 consumer，例如 Hadoop、Hbase、Solr 等;

  • (2)消息系统:

解耦和生产者和消费者、缓存消息等:

  • (3)用户活动跟踪:

Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 Hadoop、数据仓库中做离线分析和挖掘;

  • (4)运营指标:

Kafka 也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

  • (5)流式处理:

比如 spark streaming 和 storm。

6、技术优势

1）可伸缩性: Kafka 的两个重要特性造就了它的可伸缩性。

  • 1、Kafka 集群在运行期间可以轻松地扩展或收缩(可以添加或删除代理)，而不会宕机。
  • 2、可以扩展一个 Kafka 主题来包含更多的分区。由于一个分区无法扩展到多个代理，所以它的容量受到代理磁盘空间的限制。能够增加分区和代理的数量意味着单个主题可以存储的数据量是没有限制的。

2）容错性和可靠性:

Kafka 的设计方式使某个代理的故障能够被集群中的其他代理检测到。由于每个主题都可以在多个代理上复制，所以集群可以在不中断服务的情况下从此类故障中恢复并继续运行。

3）吞吐量: 代理能够以超快的速度有效地存储和检索数据

二、kafka 概念详解

1、kafka 官网： https://kafka.apache.org/

2、kafka 概念 流程图：

2.png

3、Producer

生产者即数据的发布者，该角色将消息发布到 Kafka 的 topic 中。broker 接收到生产者发送的消息后，broker 将该消息追加到当前用于追加数据的 segment 文件中。生产者发送的消息，存储到一个 partition 中，生产者也可以指定数据存储的 partition。

4、Consumer

消费者可以从 broker 中读取数据。消费者可以消费多个 topic 中的数据。

5、Topic

在 Kafka 中，使用一个类别属性来划分数据的所属类，划分数据的这个类称为 topic。如果把Kafka看做为一个数据库，topic 可以理解为数据库中的一张表，topic 的名字即为表名。

6、Partition

topic 中的数据分割为一个或多个 partition。每个 topic 至少有一个 partition。每个 partition 中的数据使用多个 segment 文件存储。partition 中的数据是有序的，partition 间的数据丢失了数据的顺序。如果 topic 有多个 partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将 partition 数目设为 1。

7、Partition offset

每条消息都有一个当前 Partition 下唯一的64字节的 offset，它指明了这条消息的起始位置。

8、Replicas of partition

副本是一个分区的备份。副本不会被消费者消费，副本只用于防止数据丢失，即消费者不从为 folower 的 partition 中消费数据，而是从为 leader 的 partition 中读取数据。副本之间是一主多从的关系。

9、Broker

Kafka 集群包含一个或多个服务器，服务器节点称为 broker。broker 存储 topic 的数据。如果某 topic 有N个 partition，集群有N个 broker，那么每个 broker 存储该 topic 的一个 partition。如果某 topic 有N个 partition，集群有(N+M)个 broker，那么其中有N个 broker 存储该 topic 的一个 partition，剩下的M个 broker 不存储该 topic 的 partition 数据。如果某 topic 有N个 partition，集群中 broker 数目少于N个，那么一个 broker 存储该 topic 的一个或多个 partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致 Kafka 集群数据不均衡。

10、Leader

每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责数据的读写的 partition。

11、Follower

Follower 跟随 Leader，所有写请求都通过 Leader 路由，数据变更会广播给所有 Follower，Follower 与 Leader 保持数据同步。如果 Leader 失效，则从 Follower 中选举出一个新的Leader。当 Follower 与 Leader 挂掉、卡住或者同步太慢，leader 会把这个 follower 从 “in syncreplicas"(ISR) 列表中删除，重新创建一个 Follower。

12、Zookeeper

Zookeeper 负责维护和协调 broker。当 Kafka 系统中新增了 broker 或者某个 broker 发生故障失效时，由 ZooKeeper 通知生产者和消费者。生产者和消费者依据 Zookeeper 的 broker 状态信息与 broker 协调数据的发布和订阅任务。

13、AR(Assigned Replicas)

分区中所有的副本统称为 AR。

14、ISR(In-Sync Replicas)

所有与 Leader 部分保持一定程度的副(包括 Leader 副本在内)本组成 ISR。

15、OSR(Out-of-Sync-Replicas)

与 Leader 副本同步滞后过多的副本。

16、HW(High Watermark)

高水位，标识了一个特定的 offset，消费者只能拉取到这个 offset 之前的消息。

17、LEO(Log End offset)

即日志末端位移(log end offset)，记录了该副本底层日志(l0g)中下一条消息的位移值。注意是下一条消息!也就是说，如果 LEO=10，那么表示该副本保存了10条消息，位移值范围是[0,9]。

在这里插入图片描述

三、kafka 环境配置 jdk、zookeeper 下载安装

1、 jdk-12 下载安装

2、jdk-12 环境变量配置


sudo vim /etc/profile # 添加 jdk 环境变量配置
export JAVA_HOME=/opt/java/jdk-12.0.1
export JRE_HOME-${JAVA_HOME}/jre
#export MAVEN_HOME=/opt/maven/apache-maven-3.5.0
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
#export PATH=S{JAVA_HOME}/bin:${MAVEN_HOME}/bin:$PATH
export PATH=S{JAVA_HOME}/bin:$PATH# 测试 jdk 是否安装配置成功
java -version

3、zookeeper 下载安装

1）Zookeeper 是安装 Kafka 集群的必要组件，Kafka 通过 Zookeeper 来实施对元数据信息的管理，包括集群、主题分区等内容。

2）同样在官网下载安装包到指定目录解压缩，步骤如下:

ZooKeeper 官网: http://zookeeper.apache.org

3）把 zookeeper 上传至服务器，并解压


# 切换目录：
cd /usr/local/zookeeper/
# 解压即安装
tar -zxvf apache-zookeeper-3.6.3-bin.tar.gz -C /usr/local/zookeeper/

4）修改 Zookeeper 的配置文件，首先进入安装路径 conf 目录，并将 zoo_sample.cfg 文件修改为 zoo.cfg，并对核心参数进行配置。文件内容如下:


# 切换目录
cd /usr/local/zookeeper/apache-zookeeper-3.6.3-bin/conf/# 将 zoo_sample.cfg 文件修改为 zoo.cfg
mv zoo_sample.cfg zoo.cfg# 对核心参数进行配置
vim zoo.cfg# The number of milliseconds of each tick
#zk服务器的心跳时间
tickTime-2080
# The number of ticks that the initial
#synchronization phase can take
#投票选举新Leader的初始化时间
initlimit=10
# The number of ticks that can pass between
# sendidg a request and getting an acknowledgementsyncLimit-5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.# 数据目录(需要新建此目录)
dataDir=/usr/local/zookeeper/apache-zookeeper-3.6.3-bin/data# 日志目录
dataLogDir=/usr/local/zookeeper/apache-zookeeper-3.6.3-bin/log# the port at which the clients will connect
#Zookeeper对外服务端口，保持默认
clientPort-2181

4、启动 Zookeeper 命令: bin/zkServer.sh start


# 切换目录
cd /usr/local/zookeeper/apache-zookeeper-3.6.3-bin/
# 启动 Zookeeper
bin/zkServer.sh start
# 查询 zookeeper 是否启动成功
ps -ef | grep zookeeper 
# 或者 
jps -l

四、kafka 环境配置 kafka 下载安装

1、 下载安装解压 kafka

官网下载地址： https://kafka.apache.org/downloads


# 切换目录
cd /usr/local/kafka/# 解压即安装
tar -zxvf kafka-2.12.2.8.0.tar.gz -C /usr/local/kafka/

2、配置 kafka 参数


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 修改 kafka 配置参数：
vim config/server.properties# server.properties 配置中需要关注以下几个参数:# 表示 broker 的编号，如果集群中有多个 broker，则每个 broker 的编号需要设置的不同
broker.id=0 # brokder 对外提供的服务入口地址（ kafka 监听地址：你的虚拟机 IP 地址）
# listeners=PLAINTEXT://127.0.0.1:9092 
listeners=PLAINTEXT://172.18.30.110:9092 # 设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log # Kafka 所需 Zookeeper 集群地址
zookeeper.connect=lolalhost:2181

3、启动 kafka


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 启动 kafka
bin/kafka-server-start.sh config/server.properties

4、查询 kafka 是否启动成功：


# 重新打开一个终端# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 查询 kafka 是否启动成功
ps -ef | grep kafka 
# 或者 
jps -l

五、kafka 消息的生产与消费

1、创建一个主题

1）命令： bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1

2）参数说明：

–zookeeper ：指定了 kafka 所连接的 zookeeper 服务地址。
–topic : 指定了所要创建主题的名称
–partitions : 指定了分区个数
–replication-factor : 指定了副本因子
–create : 创建主题的动作指令。


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 创建一个名为 heima 的 主题
bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --create --topic heima --partitions 2 --replication-factor 1

2、展示出当前所有主题


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 展示出当前所有主题
bin/kafka-topics.sh --zookeeper 1ocalhost:2181 --1ist# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --1ist

3、 查看主题详情：


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 查看主题详情
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --describe --topic heima

4、启动消费端接收消息：

命令: bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima

参数说明：
–bootstrap-server 指定了连接 Kafka 集群的地址
–topic 指定了消费端订阅的主题


# 重新打开一个终端：# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 启动消费端接收消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-console-consumer.sh --bootstrap-server 172.18.30.110:9092 --topic heima

5、生产端发送消息

命令: bin/kafka-console-producer.sh --broker-list localhost:9092 --topic heima

参数说明：
–broker-list 指定了连接的 Kafka 集群的地址
–topic 指定了发送消息时的主题


# 重新打开一个终端：# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 生产端发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-console-producer.sh --broker-list 172.18.30.110:9092 --topic heima

在这里插入图片描述

在这里插入图片描述

六、kafka java 第一个程序 001

1、打开 idea 创建 artifactId 名为 kafka_learn 的 maven 工程。

--> idea --> File --> New --> Project --> Maven Project SDK: ( 1.8(java version "1.8.0_131" ) --> Next --> Groupld : ( djh.it )Artifactld : ( kafka_learn )Version : 1.0-SNAPSHOT--> Name: ( kafka_learn )Location: ( ...\kafka_learn\ )	--> Finish

2、在 kafka_learn 工程的 pom.xml 文件中导入依赖坐标。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>djh.it</groupId><artifactId>kafka_learn</artifactId><version>1.0-SNAPSHOT</version><name>kafka_learn</name><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.8.RELEASE</version><relativePath></relativePath></parent><properties><java.version>8</java.version>
<!--        <scala.version>2.11</scala.version>--><scala.version>2.12</scala.version><slf4j.version>1.7.21</slf4j.version>
<!--        <kafka.version>2.0.0</kafka.version>--><kafka.version>2.8.0</kafka.version><lombok.version>1.18.8</lombok.version><junit.version>4.11</junit.version><gson.version>2.2.4</gson.version><protobuff.version>1.5.4</protobuff.version>
<!--        <spark.version>2.3.1</spark.version>--><spark.version>2.4.8</spark.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>${kafka.version}</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_${scala.version}</artifactId><version>${kafka.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>${slf4j.version}</version></dependency><dependency><groupId>com.google.code.gson</groupId><artifactId>gson</artifactId><version>${gson.version}</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>${junit.version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><scope>provided</scope></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-core</artifactId><version>${protobuff.version}</version></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-runtime</artifactId><version>${protobuff.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.9.4</version></dependency><dependency><groupId>com.fasterxml.jackson.module</groupId><artifactId>jackson-module-scala_2.11</artifactId><version>2.9.5</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.version}</artifactId><version>${spark.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql-kafka-0-10_${scala.version}</artifactId><version>${spark.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>
<!-- kafka_learn\pom.xml -->

3、在 kafka_learn 工程中，创建 生产者 ProducerFastStart.java 类

/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ProducerFastStart.java**  2024-6-21 创建 生产者 ProducerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import io.protostuff.StringSerializer;
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;public class ProducerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//2)设置重试次数properties.put(ProducerConfig.RETRIES_CONFIG, 10);//3)设置值序列化器properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//4)设置集群地址properties.put("bootstrap.servers", brokerList);//properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);ProducerRecord<String,String> record = new ProducerRecord<>(topic, "kafka-demo", "hello-kafka-test!");try{producer.send(record);}catch (Exception e){e.printStackTrace();}producer.close();}
}

在这里插入图片描述

七、kafka java 第一个程序 002

1、在 kafka_learn 工程中，创建 消费者 ConsumerFastStart.java 类


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ConsumerFastStart.java**  2024-6-21 创建 消费者 ConsumerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ConsumerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";private static final String groupId = "group.demo";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//2)设置值序列化器properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//3)设置集群地址properties.put("bootstrap.servers", brokerList);properties.put("group.id", groupId);KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);consumer.subscribe(Collections.singletonList(topic));while (true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){System.out.println(record.value());}}}
}

2、在 kafka_learn 工程中，启动 消费者 ConsumerFastStart.java 类 和 生产者 ProducerFastStart.java 类，进行测试。会发现，消费者这边接收到，生产者发送的消费。

消费者接收消息.png

八、kafka 优化与总结

1、在 kafka_learn 工程中，修改 生产者 ProducerFastStart.java 类 ，进行优化代码。


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ProducerFastStart.java**  2024-6-21 创建 生产者 ProducerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
//注意导包，一定要导成 kafka 的序列化包
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;public class ProducerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器 -- 优化代码//properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//2)设置重试次数 -- 优化代码properties.put(ProducerConfig.RETRIES_CONFIG, 10);//3)设置值序列化器 -- 优化代码//properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//4)设置集群地址 -- 优化代码//properties.put("bootstrap.servers", brokerList);properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);ProducerRecord<String,String> record = new ProducerRecord<>(topic, "kafka-demo", "优化-2024-6-21-kafka-test!");try{producer.send(record);}catch (Exception e){e.printStackTrace();}producer.close();}
}

2、在 kafka_learn 工程中，修改 消费者 ConsumerFastStart.java 类，进行优化代码。


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ConsumerFastStart.java**  2024-6-21 创建 消费者 ConsumerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
//注意导包，一定要导成 kafka 的序列化包
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ConsumerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";private static final String groupId = "group.demo";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器 -- 优化代码//properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//2)设置值序列化器 -- 优化代码//properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//3)设置集群地址 -- 优化代码//properties.put("bootstrap.servers", brokerList);properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);//properties.put("group.id", groupId);properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);consumer.subscribe(Collections.singletonList(topic));while (true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){System.out.println(record.value());}}}
}

在这里插入图片描述

3、初识 kafka 总结：

kafka 启动前，在进行配置 kafka 参数时，即修改 kafka/kafka_2.12-2.8.0/config/server.properties 配置文件时，需要注意以下几点：

  • 1） zookeeper.connect

指明 Zookeeper 主机地址，如果 zookeeper 是集群则以逗号隔开，如:172.6.14.61:2181,172.6.14.62:2181,172.6.14.63:21B1

  • 2） listeners 监听列表 配置：

broker 对外提供服务时绑定的 IP 和端口。多个以逗号隔开，如果监听器名称不是一个安全的协议,listener.security.protocol.map 也必须设置。主机名称设置 0.0.0.0 绑定所有的接口，主机名称为空则绑定默认的接口。如:PLAINTEXT://myhost:9092,SSL:/1:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

  • 3） broker.id

broker 的唯一标识符，如果不配置则自动生成，建议配置且一定要保证集群中必须唯一，默认-11.4.4

  • 4） log.dirs

日志数据存放的目录，如果没有配置则使用 log.dir，建议此项配置。

  • 5） message.max.bytes

服务器接受单个消息的最大大小，默认 1000012 约等于 976.6KB。


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 修改 kafka 配置参数：
vim config/server.properties# server.properties 配置中需要关注以下几个参数:# 1）Kafka 所需 Zookeeper 集群地址
zookeeper.connect=lolalhost:2181 # 2）listeners 监听列表 配置：brokder 对外提供的服务入口地址（ kafka 监听地址：你的虚拟机 IP 地址）
# listeners=PLAINTEXT://127.0.0.1:9092 
listeners=PLAINTEXT://172.18.30.110:9092 # 3）表示 broker 的编号，如果集群中有多个 broker，则每个 broker 的编号需要设置的不同
broker.id=0 # 4）log.dirs 设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/32480.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

如何使用kimi智能助手：您的智能生活小助手

如何使用kimi智能助手：您的智能生活小助手

Kimi智能助手是一款功能强大的AI工具&#xff0c;旨在帮助用户提高工作效率和生活品质。下面小编将详细介绍如何使用Kimi智能助手&#xff0c;涵盖其主要功能以及一些实用技巧。 一、Kimi智能助手的主要功能 多语言对话能力&#xff1a;Kimi擅长中文和英文的对话&#xff0c;可…
阅读更多...
探索计算机视觉（人工智能重要分支）的发展与应用

探索计算机视觉（人工智能重要分支）的发展与应用

引言 在当今快速发展的科技时代&#xff0c;计算机视觉作为人工智能领域的重要分支&#xff0c;正日益成为各行各业不可或缺的关键技术。从简单的图像处理到复杂的智能系统&#xff0c;计算机视觉的发展不仅改变了我们看待世界的方式&#xff0c;也深刻影响着工业、医疗、交通等…
阅读更多...
Windows安装配置jdk和maven

Windows安装配置jdk和maven

他妈的远程连接不上公司电脑&#xff0c;只能在家重新配置一遍&#xff0c;在此记录一下后端环境全部配置 Windows安装配置JDK 1.8一、下载 JDK 1.8二、配置环境变量三、验证安装 Windows安装配置Maven 3.8.8一、下载安装 Maven并配置环境变量二、设置仓库镜像及本地仓库三、测…
阅读更多...
2024最新版Python 3.12.4安装使用指南

2024最新版Python 3.12.4安装使用指南

2024最新版Python 3.12.4安装使用指南 2024最新版Python 3.12.4安装使用指南0. Python的受欢迎程度1. 安装最新版Python 3.12.42. 验证Python 3.12.4版本3. 验证Python功能4. 使用IDLE交互式开发模式5. 安装Python扩展库相关阅读&#xff1a; By Jackson 2024最新版Python 3.12…
阅读更多...
java基于ssm+jsp 汽车在线销售系统

java基于ssm+jsp 汽车在线销售系统

1 前台功能模块 网站首页 网页首页汽车在线销售系统模块如下&#xff1a;首页、汽车信息、新闻资讯、留言反馈、我的收藏管理等功能图1 图1网页首页 网页前台车辆信息效果图如图2所示 图2 车辆信息界面图 2 管理员功能模块 管理员输入个人的账号、密码登录系统&#xff0c…
阅读更多...
压力测试

压力测试

1.什么是压力测试 压力测试考察当前软硬件环境下系统所能承受的最大负荷并帮助找出系统瓶颈所在。压测都是为了系统在线上的处理能力和稳定性维持在一个标准范围内&#xff0c;做到心中有数 使用压力测试&#xff0c;我们有希望找到很多种用其他测试方法更难发现的错误&#…
阅读更多...
基于matlab的K-means聚类图像分割

基于matlab的K-means聚类图像分割

1 原理 K-means聚类算法在图像分割中的应用是基于一种无监督的学习方法&#xff0c;它将图像中的像素点或特征区域划分为K个不同的簇或类别。以下是K-means聚类算法用于图像分割的原理&#xff0c;包括步骤和公式&#xff1a; 1.1 原理概述 选择簇的数量(K)&#xff1a; 首先…
阅读更多...
YOLOv9基础 | 实时目标检测新SOTA，手把手带你深度解析yolov9论文！

YOLOv9基础 | 实时目标检测新SOTA，手把手带你深度解析yolov9论文！

前言：Hello大家好，我是小哥谈。YOLOv9是Chien-Yao Wang等人提出的YOLO系列的最新版本之一（截止到目前，YOLOv10已发布），于2024年2月21日发布。它是 YOLOv7的改进版本，两者均由Chien-Yao Wang及其同事开发。本节课就以YOLOv9论文为基础带大家深入解析YOLOv9算法。🌈 …
阅读更多...
浏览器-服务器架构 (BS架构) 详解

浏览器-服务器架构 (BS架构) 详解

目录 前言1. BS架构概述1.1 BS架构的定义1.2 BS架构的基本原理 2. BS架构的优势2.1 客户端简化2.2 易于更新和维护2.3 跨平台性强2.4 扩展性高 3. BS架构的劣势3.1 网络依赖性强3.2 安全性问题3.3 用户体验局限 4. BS架构的典型应用场景4.1 企业内部应用4.2 电子商务平台4.3 在…
阅读更多...
java小代码（1）

java小代码（1）

代码 &#xff1a; 今日总结到此结束&#xff0c;拜拜&#xff01;
阅读更多...
1999-2022年 297个地级市-医院卫生院数量及床位数量（数据收集）

1999-2022年 297个地级市-医院卫生院数量及床位数量（数据收集）

全国297个地级市的医院卫生院数量的稳步增长是医疗事业发展的一个重要标志。政府的持续投入和对医疗设施的改善&#xff0c;不仅提升了医疗服务的硬件水平&#xff0c;也通过引进和培养医疗人才、优化服务流程&#xff0c;提高了医疗服务的整体质量。这些举措极大地增强了人民群…
阅读更多...
C语言之详解预处理

C语言之详解预处理

前言&#xff1a; 预处理也叫预编译&#xff0c;是编译代码时的第一步&#xff0c;经过预处理后生成一个.i文件&#xff0c;如果不明白编译与链接作用的小伙伴可以先看看博主的上一篇博客—— &#xff0c;不然知识连贯性可能会显得很差哦。 正文目录&#xff1a; 预定义符号#…
阅读更多...
font-spider按需生成字体文件

font-spider按需生成字体文件

font-spider可以全局安装,也可以单个项目内安装,使用npm run xxxx的形式 npm i font-spider "dev": "font-spider ./*.html" <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name&…
阅读更多...
Android测量

Android测量

最大模式&#xff08;MeasureSpec.AT_MOST&#xff09; 这个也就是父组件&#xff0c;能够给出的最大的空间&#xff0c;当前组件的长或宽最大只能为这么大&#xff0c;当然也可以比这个小。 最高两位是11的时候表示”最大模式”。即MeasureSpec.AT_MOST未指定模式&#xff08;…
阅读更多...
1996年-2023年 全国298个地级市-外商直接投资FDI（数据收集）

1996年-2023年 全国298个地级市-外商直接投资FDI（数据收集）

外商直接投资&#xff08;FDI&#xff09;是一种跨国界的经济活动&#xff0c;它涉及外国投资者在中国境内进行的直接投资行为。这种投资行为不仅包括以货币、实物、技术等形式的资本投入&#xff0c;还可能包括开办独资企业、合资企业、合作企业&#xff0c;以及参与资源开发等…
阅读更多...
微型操作系统内核源码详解系列五（四）：cm3下svc启动任务

微型操作系统内核源码详解系列五（四）：cm3下svc启动任务

系列一&#xff1a;微型操作系统内核源码详解系列一&#xff1a;rtos内核源码概论篇&#xff08;以freertos为例&#xff09;-CSDN博客 系列二&#xff1a;微型操作系统内核源码详解系列二&#xff1a;数据结构和对象篇&#xff08;以freertos为例&#xff09;-CSDN博客 系列…
阅读更多...
开发者配置项、开发者选项自定义

开发者配置项、开发者选项自定义

devOptions.vue源码 <!-- 开发者选项 &#xff08;CtrlAltShiftD&#xff09;--> <template><div :class"$options.name" v-if"visible"><el-dialog:custom-class"sg-el-dialog":append-to-body"true":close-on…
阅读更多...
Flutter 如何发布安卓应用？

Flutter 如何发布安卓应用？

android:hardwareAccelerated“true” android:windowSoftInputMode“adjustResize”> <meta-data android:name“flutterEmbedding” android:value“2” /> Flutter生成的文件建议是大部分内容可以保留不动&#xff0c;但是可以根据需要进行修改。 具体可能要修…
阅读更多...
STM32读写备份寄存器和实时时钟

STM32读写备份寄存器和实时时钟

文章目录 1. 硬件电路 2. RTC操作注意事项 操作步骤 3. 代码实现 3.1 读写备份寄存器 3.1.1 main.c 3.2 实时时钟 3.2.1 MyRTC.c 3.2.2 MyRTC.h 3.2.3 main.c 1. 硬件电路 对于BKP备份寄存器和RTC实时时钟的详细解析可以看下面这篇文章&#xff1a; STM32单片机BKP备…
阅读更多...
Linux查看公网IP的解决方案

Linux查看公网IP的解决方案

大家好，我是爱编程的喵喵。双985硕士毕业，现担任全栈工程师一职，热衷于将数据思维应用到工作与生活中。从事机器学习以及相关的前后端开发工作。曾在阿里云、科大讯飞、CCF等比赛获得多次Top名次。现为CSDN博客专家、人工智能领域优质创作者。喜欢通过博客创作的方式对所学的…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023