Kafka：分布式部署，实践、API连接操作_kafka分布式部署

---

Kafka 的资源下载链接：https://pan.baidu.com/s/1WlWf__Y8c0jXCOVAFixFRw
提取码：v0lx

---

Kafka 基础知识理论版

Kafka Streams 小实践 ——开放词频统计应用

Kafka集群分布式部署与测试

Flume：kafka+flume 对接数据

---

目的：

1. 学会配置Kafka ；
2. 学会使用 Kafka 生产者和消费者实例；
3. 基于java API 方式使用 Kafka。

---

一、Kafka 安装与启动

1.1 创建文件

mkdir -p /usr/kafka

1.2 解压文件

tar -zxvf kafka_2.11-2.4.0.tgz -C /usr/kafka/

进入配置目录：
cd /usr/kafka/kafka_2.11-2.4.0/config

修改文件 vi server.properties：

# broker的全局唯一编号，不能重复，依次 0、1、2。
broker.id=0
# 用来监听链接的端口，producer或consumer将在此端口建立连接
port=9092
# 处理网络请求的线程数量
num.network.threads=3
# 用来处理磁盘IO的线程数量
num.io.threads=8
# 发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
# 接受套接字的缓冲区大小
socket.receive.buffer.bytes=102400
# 请求套接字的缓冲区大小
socket.request.max.bytes=104857600
# kafka消息存放的路径
log.dirs=/usr/kafka/kafka_2.11-2.4.0/tmp/kafka-logs
# topic在当前broker上的分片个数
num.partitions=2
# 用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# segment文件保留的最长时间，超时将被删除
log.retention.hours=168
# 滚动生成新的segment文件的最大时间
log.roll.hours=1
# 日志文件中每个segment的大小，默认为1G
log.segment.bytes=1073741824
# 周期性检查文件大小的时间
log.retention.check.interval.ms=300000
# 日志清理是否打开
log.cleaner.enable=true
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# broker需要使用zookeeper保存meta数据
zookeeper.connect=master:2181,slave1:2181,slave2:2181

# zookeeper链接超时时间
zookeeper.connection.timeout.ms=6000
# partion buffer中，消息的条数达到阈值，将触发flush到磁盘
log.flush.interval.messages=10000
# 消息buffer的时间，达到阈值，将触发flush到磁盘
log.flush.interval.ms=3000
# 删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除
delete.topic.enable=true
# host.name为本机IP,不改则会Producerconnection to localhost:9092 unsuccessful 错误!
# host.name=slave1（slave2）
host.name=master
1
2
3
4
5
6
7
8
9
10
11
12
13
14

修改完毕，保存退出。

---

1.3 配置环境变量（三个节点）

vi /etc/profile
# kafka
export KAFKA_HOME=/usr/kafka/kafka_2.11-2.4.0
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
1
2
3
4
5

---

1.4 分发文件，且修改配置文件server.properties(两个从节点)

同理，slave2也是这样。

---

启动集群，zookeeper、hadoop、kafka。

# 启动 zookeeper（三个节点）
/usr/zookeeper/zookeeper-3.4.10/bin/zkServer.sh start
/usr/zookeeper/zookeeper-3.4.10/bin/zkServer.sh status
# 启动hadoop(master)
/usr/hadoop/hadoop-2.7.3/sbin/start-all.sh

# 后台启动kafka，(三个节点)
cd $KAFKA_HOME
./bin/kafka-server-start.sh config/server.properties &
# 查看后台运行
jobs -l

1
2
3
4
5
6
7
8
9
10
11
12

---

1.5 创建多个topic

在 Kafka 的 bin 目录下有 kafka-topics.sh 文件，通过该文件就可以操作与主题相关的功能，下面我们创建一个 badou_topic1 主题，分区数为10，副本数3，命令如下：

cd $KAFKA_HOME
# 在三个节点创建主题 badou_topic1
./bin/kafka-topics.sh --create --zookeeper master:2181,slave1:2181,slave2:2181 --replication-factor 3 --partitions 10 --topic badou_topic1
1
2
3

# 查看详细信息
./bin/kafka-topics.sh --describe --zookeeper master:2181,slave1:2181,slave2:2181 --topic badou_topic1

# 查看主题
./bin/kafka-topics.sh --list --zookeeper master:2181,slave1:2181,slave2:2181

# --delete删除主题
kafka-topics.sh --delete --zookeeper master:2181,slave1:2181,slave2:2181 --topic badou_topic1
1
2
3
4
5
6
7
8

---

1.6 创建生产者producer、消费者consumer

主题创建成功后，在master 创建生产者生产消息，通过bin目录下的 kafka-console-producer.sh 文件操作 —— 向主题发送消息数据，命令如下：

# 创建生产者producer
./bin/kafka-console-producer.sh --broker-list master:9092,slave1:9092,slave2:9092  --topic badou_topic1

# 创建消费者consumer
./bin/kafka-console-consumer.sh --from-beginning  --topic badou_topic1 --bootstrap-server master:9092,slave1:9092,slave2:9092
1
2
3
4
5

在生产者输入数据，会在消费者进行数据拉取

现在kill 掉slave2 的 kafka进程

# 查看详细信息
./bin/kafka-topics.sh --describe --zookeeper master:2181,slave1:2181,slave2:2181 --topic badou_topic1
1
2

我们可以发现，leader 变成0,1；Isr变成0,1。

---

三、基于java API 方式使用 Kafka

用户不仅能够通过命令行形式操作 Kafka 服务，还能用多种语言操作 Kafka。用户在开发独立项目时，通过调用Kafka API来操作Kafka集群，其核心API主要有5种分别是：Producer API、Consumer API 、Streams API、Connect API 、AdminClient API。

KafkaProducer API 常用API表

KafkaComsumer API 常用API表

---

3.1 添加依赖

<!--kafka-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.4.0</version>
        </dependency>
1
2
3
4
5
6

3.2 编写生产者客户端

for循环 输入50个数字到消费者客户端

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerTest {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 1、指定Kafka集群的主机名和端口号
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2: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、设置key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 8、设置value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 9、生产数据
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 50; i++) {
            producer.send(new ProducerRecord<String, String>("theme", Integer.toString(i), "hello world-" + i));
        }
        producer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3.3 编写消费者客户端

接收生产者客户端信息，并读取内容显示。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
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 java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerTest {
    public static void main(String[] args) {
        // 1、准备配置文件
        Properties props = new Properties();
        // 2、指定Kafka集群主机名和端口号
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave1:9092");
        // 3、指定消费者组ID，在同一时刻同一消费组中只有一个线程可以去消费一个分区数据，不同的消费组可以去消费同一个分区的数据。
        props.put("group.id", "theme");
        // 4、自动提交偏移量
        props.put("enable.auto.commit", "true");
        // 5、自动提交时间间隔，每秒提交一次
        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");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
        // 6、订阅数据，这里的topic可以是多个
        kafkaConsumer.subscribe(Arrays.asList("theme"));
        // 7、获取数据
        while (true) {
            //每隔100ms就拉去一次
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("topic = %s,offset = %d, key = %s, value = %s%n", record.topic(), record.offset(), record.key(), record.value());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38