Kafka集群部署安装及使用

Kafka集群部署安装及使用一、安装部署1.1集群规划1.2 jar包下载1.3 部署1.4 Kafka命令行操作二、Kafka API的使用2.1 Producer API2.1.1 消息发送流程2.1.2 异步发送API2.2 Consumer API2.2.1 自动提交offset2.2.2 手动提交offset一、安装部署1.1集群规划hadoop100hadoop101hadoop102

pit_man

4476人浏览 · 2021-10-05 11:47:38

pit_man · 2021-10-05 11:47:38 发布

Kafka集群部署安装及使用

一、安装部署
二、Kafka API的使用

一、安装部署

1.1集群规划

kafka依赖zookeeper负责broker的注册，所以需要提前安装好zookeeper，安装教程参考。

hadoop100	hadoop101	hadoop102
zookeeper	zookeeper	zookeeper
kafka	kafka	kafka

1.2 jar包下载

http://kafka.apache.org/downloads

1.3 部署

1）解压安装包

[user@hadoop100 software]$ tar -zxvf kafka_2.11-2.4.1.tgz -C /opt/module/

2）修改解压后的文件名称

[user@hadoop100 module]$ mv kafka_2.11-2.4.1/ kafka

3）在/opt/module/kafka目录下创建logs文件夹

[user@hadoop100 kafka]$ mkdir logs

4）修改配置文件

[user@hadoop100 kafka]$ cd config/
[user@hadoop100 config]$ vi server.properties

输入以下内容：

#broker的全局唯一编号，不能重复
broker.id=1
#删除topic功能使能
delete.topic.enable=true
#处理网络请求的线程数量
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=/opt/module/kafka/logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=hadoop100:2181,hadoop101:2181,hadoop102:2181/kafka

$\color{red}{注意：zookeeper和broker id是在原来的内容上修改的，delete.topic.enable是增加的}$
在这里插入图片描述

5）配置环境变量

[user@hadoop100 module]$ sudo vi /etc/profile

添加如下内容：

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

使修改的环境变量生效：

[user@hadoop100 module]$ source /etc/profile

6）分发安装包

[user@hadoop100 module]$ scp -r kafka/ root@hadoop101:/opt/module
[user@hadoop100 module]$ scp -r kafka/ root@hadoop102:/opt/module

注意：分发之后记得配置其他机器的环境变量
分别在hadoop101和hadoop102上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=2、broker.id=3
注：broker.id不得重复

7）启动集群
依次在hadoop100、hadoop101、hadoop102节点上启动kafka

[user@hadoop100 kafka]$ kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
[user@hadoop101 kafka]$ kafka-server-start.sh -daemon  $KAFKA_HOME/config/server.properties
[user@hadoop102 kafka]$ kafka-server-start.sh -daemon  $KAFKA_HOME/config/server.properties

8）关闭集群（停止kafka需要几秒的时间，才会看到kafka进程关闭）

[user@hadoop100 kafka]$ kafka-server-stop.sh
[user@hadoop101 kafka]$ kafka-server-stop.sh
[user@hadoop102 kafka]$ kafka-server-stop.sh

1.4 Kafka命令行操作

1）查看当前服务器中的所有topic

[user@hadoop100 kafka]$ kafka-topics.sh --zookeeper hadoop100:2181/kafka --list

2）创建topic

[user@hadoop100 kafka]$ kafka-topics.sh --zookeeper hadoop100:2181/kafka \
--create --replication-factor 1 --partitions 1 --topic first
选项说明：
--topic 定义topic名
--replication-factor  定义副本数
--partitions  定义分区数

3）删除topic

[user@hadoop100 kafka]$ kafka-topics.sh --zookeeper hadoop100:2181/kafka \
--delete --topic first
需要server.properties中设置delete.topic.enable=true否则只是标记删除。

4）发送消息

[user@hadoop100 kafka]$ kafka-console-producer.sh \
--broker-list hadoop100:9092 --topic first

输入信息：

hello world
user user

–broker-list：指定数据发送到哪个broker上

5）消费消息

[user@hadoop101 kafka]$ kafka-console-consumer.sh \
--bootstrap-server hdp01:9092 --topic first

[user@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
--bootstrap-server hdp01:9092 --from-beginning --topic first

–from-beginning：会把主题中以往所有的数据都读取出来。

二、Kafka API的使用

2.1 Producer API

2.1.1 消息发送流程

Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

在这里插入图片描述

相关参数：
batch.size：只有数据积累到batch.size之后，sender才会发送数据。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。

2.1.2 异步发送API

1）导入依赖

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>2.4.1</version>
</dependency>

2）编写代码
需要用到的类：
KafkaProducer：需要创建一个生产者对象，用来发送数据
ProducerConfig：获取所需的一系列配置参数
ProducerRecord：每条数据都要封装成一个ProducerRecord对象
（1）不带回调函数的API

package com.user.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop100:9092");//kafka集群，broker-list
        props.put("acks", "all");
        props.put("retries", 1);//重试次数
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待时间
        props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));
        }
        producer.close();
    }
}

（2）带回调函数的API
回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是RecordMetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

package com.user.kafka;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustormProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hdp01:9092");//kafka集群，broker-list
        props.put("acks", "all"); // 或者 ("acks", "-1")，不能设为("acks", -1)
        props.put("retries", 1);//重试次数
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待时间
        props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer< String, String>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)), new Callback() {

                //回调函数，该方法会在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.println("success->" + metadata.offset() + ", topic: " + metadata.topic() + ", partition: " + metadata.partition());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

2.2 Consumer API

Consumer消费数据时的可靠性是很容易保证的，因为数据在Kafka中是持久化的，故不用担心数据丢失问题。
由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。
所以offset的维护是Consumer消费数据是必须考虑的问题。

2.2.1 自动提交offset

1）导入依赖
在maven工程pom.xml文件中添加依赖

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>2.4.1</version>
</dependency>

2）编写代码
需要用到的类：
KafkaConsumer：需要创建一个消费者对象，用来消费数据
ConsumerConfig：获取所需的一系列配置参数
ConsuemrRecord：每条数据都要封装成一个ConsumerRecord对象
为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。
自动提交offset的相关参数：
enable.auto.commit：是否开启自动提交offset功能
auto.commit.interval.ms：自动提交offset的时间间隔
以下为自动提交offset的代码：

package com.user.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class CustomConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop100: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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("first"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

consumer.poll: 如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。为了防止这种情况，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

2.2.2 手动提交offset

虽然自动提交offset十分简介便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
提交offset
以下为提交offset的示例。

package com.user.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * @author liubo
 */
public class CustomComsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop100:9092");//Kafka集群
        props.put("group.id", "test");//消费者组，只要group.id相同，就属于同一个消费者组
        props.put("enable.auto.commit", "false");//关闭自动提交offset
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("first"));//消费者订阅主题

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);//消费者拉取数据
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            consumer.commitSync();//提交offset
        }
    }
}