一、Kafka环境搭建和Topic管理

1 单机环境搭建

1.1 环境准备

安装JDK1.8+，配置JAVA_HOME(CentOS 6.10+ 64bit)
配置主机名和IP映射
关闭防火墙&防火墙开机自启动
集群环境下需要配置每台机器，同步时钟 ntpdate cn.pool.ntp.org | ntp[1-7].aliyun.com
安装&启动zookeeper
安装&启动|关闭Kafka

1.1.1 JDK 安装

安装

~ rpm -qa | grep jdk # 如果没安装过，需要安装

~ rpm -e `rpm -qa | grep jdk` # 卸载老的版本

~ rpm -ivh jdk-8u191-linux-x64.rpm

默认安装在/usr/java/下

配置环境变量

虽然rpm包安装，默认能识别我们的java和javac命令，但不能证明JAVA_HOME就配置成功了，可以通过jps来验证。

1、 vi ~/.bashrc



# 底部增加

JAVA_HOME=/usr/java/latest

PATH=$PATH:$JAVA_HOME/bin

CLASSPATH=.

export JAVA_HOME

export PATH

export CLASSPATH

2、source ~/.bashrc

3、echo $JAVA_HOME

1.1.2 配置主机名和ip

1、改主机名cat /etc/sysconfig/network,把HOSTNAME更改掉后reboot。这里假设更改为CentOS

2、更改ip

ipconfig看看eth0的网卡信息
vi /etc/hosts/ 末尾添加 192.168.52.129 CentOS。这里192.168.52.129假设是eth0网卡的ip

3、ping CentOS 来测试是否生效

1.1.3 关闭防火墙和防火墙开机自启动

为了测试方便，内网环境下，我们把防火墙关闭

1、关闭防火墙

终端命令：service iptables status看下防火墙服务是否启动状态
如果没关闭，终端执行service iptables stop
如果关闭了，要启动防火墙：service iptables start

2、关闭防火墙开启自启动

终端执行：chkconfig iptables off，其中iptables是服务名
chkconfig --list查看所有的开机自启动的服务

1.1.4 zookeeper下载安装

zookeeper官网，选择历史发行版本下载，选择我们要下载的版本，scp到我们的安装机器上

1、解压

tar -zxf zookeeper-3.4.6.tar.gz -C /usr/解压我们的zk

2、进入解压目录

cd /usr/zookeeper-3.4.6/

3、修改zk配置文件,先拷贝一份模板配置文件再修改

cp conf/zoo_sample.cfg conf/zoo.cfg
vi conf/zoo.cfg
修改zookeeper的数据目录为：dataDir=/root/zkdata，创建我们刚在指定的数据目录：mkdir /root/zkdata

4、启动我们的zk服务，先进入zk目录：cd /usr/zookeeper-3.4.6/，再启动服务：./bin/zkServer.sh start zoo.cfg

5、用jps命令查看zookeeper的服务：jps后zk的进程名叫：QuoeumPeerMain，用zk自带命令查看zk的状态，./bin/zkServer.sh status zoo.cfg。看到zk的Mode为standalone单机部署版，且状态ok

1.1.5 kafka下载安装

kafka官网，左下角download按钮，选择我们合适的版本，根据scala版本做选择，下载之后scp到安装机器

1、解压安装包：tar -zxf kafka_2.11-2.2.0.tag -C /usr/,进入安装目录cd /usr/kafka_2.11-2.2.0 ,常规命令目录ls bin/，常规配置目录：ls config/

2、配置server.peoperties: vi config/server.properties

配置解释：

broler.id=0 # kafka的节点唯一标识

listeners=PLAINTEXT://CentOS:9092 #CentOS为当前节点的主机名

log.dirs=/usr/kafka-logs  # kafka的broker节点存储数据的位置

num.partitions=1 # 设置topic的默认分区数

log.retention.hours=168 # 日志默认保存的策略，默认7天

zookeeper.connect=CentOS:2181 # 配置kafka依赖的zookeeper的信息

保存退出

3、kafka的启动，首先要保证zookeeper是正常的。./bin/kafka-server-start.sh -daemon config/server.properties,其中-daemon指的是以后台方式启动

4、查看kafka的启动状态，jps命令，可以看到Kafka进程

5、关闭kafka服务：./bin/kafka-server-stop.sh等一秒左右

6、创建topic：./bin/kafka-topic.sh --help可以得知为:

./bin/kafka-topic.sh --bootstrap-server CentOS:9092 --create --topic topic01 --partitions 2 --replication-factor 1

--topic 指定分区名称

--partitions 指定分区数目

--replication-factor 指定副本因子，副本因子个数不能大于我们broker的个数

7、订阅topic：



./bin/kafka-console-consumer.sh --bootstrap-server CentOS:9092 --topic topic01 --group group1

--topic 指定订阅哪个topic

--group 消费者要以组的形式，不指定会默认分配

8、生产者生产消息,打开另外终端充当生产者。



./bin/kafka-console-produser.sh --broker-list CentOS:9092 --topic topic01 # 回车后指定发送的消息

> hello~

第一章可以知道，由于相同组是均分topic信息，不同组相互负载均衡。

当我们启动多个消费者时，如果是在一个组里，每次我们发送消息会被其中一个消费者消费到。

当我们启动多个消费者，不在一个组里，每个组都能收到我们发布的信息，达到广播的效果。

2 集群环境搭建

准备三台服务器的虚拟机,假设三台主机的ip分别为，192.168.52.130,192.168.52.131,192.168.52.132

2.1 安装JDK

在三台机器上分别用rpm包，安装jdk：rpm -ivh jdk-8u191-linux-x64.rpm

2.2 配置每台机器的配置

先在a机器上配置，之后拷贝配置到b和c机器即可。跨机器拷贝，做一个机器间sh免密码验证。否则scp需要带上机器密码信息

1、在A机器上，vi /etc/hosts，增加配置：

192.168.52.130 CentOSA

192.168.52.131 CentOSB

192.168.52.132 CentOSC

接着，scp我们的配置到其他两台B和C机器：

scp /etc/hosts CentOSB:/etc/

scp /etc/hosts CentOSC:/etc/

2、在A机器上配置JDK的环境变量，scp到其他两台B和C机器：

第一步：vi ~/.bashrc



# 底部增加

JAVA_HOME=/usr/java/latest

PATH=$PATH:$JAVA_HOME/bin

CLASSPATH=.

export JAVA_HOME

export PATH

export CLASSPATH

第二步：source ~/.bashrc
第三步：echo $JAVA_HOME
第三步：验证没问题，拷贝到B和C机器：

scp ~/.bashrc CentOSB:~/

scp ~/.bashrc CentOSC:~/

第四步，在b和c机器上，分别source，让配置文件生效

2.3 关闭三台机器的防火墙

参考上文单机的关闭方法

2.4 配置集群网络时钟同步

每台机器安装ntp工具：yum install ntp -y

每台机器上都设置时钟：ntpdate ntp1.aliyun.com并执行clock -w命令同步时钟

2.5 zookeeper集群安装和搭建

1、安装及拷贝配置文件

tar -zxf zookeeper-3.4.6.tar.gz -C /usr/

cd /usr/zookeeper-3.4.6/

cp conf/zoo_sample.cfg conf/zoo.cfg

2、接着更改配置文件：vi zoo.cfg

# 修改原dataDir目录为

dataDir=/root/zkdata

# 新增集群其他节点信息，并配置服务端口和主从选举端口

server.1=CentOSA:2888:3888

server.2=CentOSB:2888:3888

server.3=CentOSC:2888:3888

3、创建我们配置文件中指定的数据目录

mkdir /root/zkdata

4、把我们在A上做的操作拷贝到其他节点B和C

scp -r /usr/zookeeper-3.4.6 CentOSB:/usr/

scp -r /usr/zookeeper-3.4.6 CentOSC:/usr/

5、在其他集群节点B和C上同样都创建数据目录

mkdir /root/zkdata

6、在每台机器的数据目录中，写入当前机器的id号。id号参考我们zoo.cfg下的节点配置

A机器： echo 1 > /root/zkdata/myid

B机器： echo 2 > /root/zkdata/myid

C机器： echo 3 > /root/zkdata/myid

7、启动我们每台节点的zookeeper

启动：/usr/zookeeper-3.4.6/bin/zkServer.sh start zoo.cfg

查看状态：/usr/zookeeper-3.4.6/bin/zkServer.sh status zoo.cfg

3.6 集群安装Kafka

1、安装及修改配置文件



tar -zxf kafka_2.11-2.2.0.tgz -C /usr/

cd /usr/kafka_2.11-2.2.0/

vi config/server.properties

2、修改配置文件

A节点配置，其他节点scp后单独改自己的配置：

listeners = PLAINTEXT://CentOSA:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181

3、拷贝到其他节点并且更改配置

scp -r kafka_2.11-2.2.0 CentOSB:/usr/

scp -r kafka_2.11-2.2.0 CentOSC:/usr/

B节点：

# A上，默认是0，这里要更改其他节点的brokerID

broker.id=1

listeners = PLAINTEXT://CentOSB:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181

C节点：

# A上，默认是0，这里要更改其他节点的brokerID

broker.id=2

listeners = PLAINTEXT://CentOSC:9092

log.dirs=/usr/kafka-logs

zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181

3 集群环境下topic管理概览

1、创建topic

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                    --create

                    --topic topic01

                    --partitions 3

                    --replication-factor 3

在我们配置的kafka日志目录中，可以看到我们topic的分区分配情况：ls /usr/kafka-logs/

2、topic列表

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

			        --list

3、topic详情

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                    --describe

                    --topic topic01

Topic:topic01	PartitionCount:3	ReplicationFactor:3	Configs:segment.bytes=1073741824

	Topic: topic01	Partition: 0	Leader: 0	Replicas: 0,2,3	Isr: 0,2,3

	Topic: topic01	Partition: 1	Leader: 2	Replicas: 2,3,0	Isr: 2,3,0

	Topic: topic01	Partition: 2	Leader: 0	Replicas: 3,0,2	Isr: 0,2,3

4、修改

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                    --create

                    --topic topic03

                    --partitions 1

                    --replication-factor 1

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                    --alter

                    --topic topic03

                    --partitions 2

注意：kafka的分区数只能由小修改到大，不能由大修改到小

5、删除

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh

                    --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                    --delete

                    --topic topic03

删除topic之后，我们相应的topic的数据，在我们的日志目录下标识为删除：ls /usr/kafka-logs/

6、订阅

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-console-consumer.sh

                  --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                  --topic topic01

                  --group g1

                  --property print.key=true

                  --property print.value=true

                  --property key.separator=,

--topic topic01 需要订阅的topic

--group g1 指定消费者组

--property print.key=true 是否打印key，可选

--property print.value=true 是否打印值，可选

--property key.separator=, key和值用什么分割，可选

注意：订阅，用的是./bin/kafka-console-consumer.sh 而不是./bin/kafka-topics.sh

7、生产

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-console-producer.sh

                  --broker-list CentOSA:9092,CentOSB:9092,CentOSC:9092

                  --topic topic01

注意：这时用的是--broker-list而不是--bootstrap-server

注意：发布，用的是./bin/kafka-console-producer.sh 而不是./bin/kafka-topics.sh

8、消费组

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh

                  --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                  --list

                  g1

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh

                  --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092

                  --describe

                  --group g1

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID    HOST            CLIENT-ID

topic01 1                      0                    0                           0     consumer-1-**    /192.168.52.130 consumer-1

topic01 0                      0                    0                          0      consumer-1-**   /192.168.52.130 consumer-1

topic01 2                      1                     1                          0      consumer-1-**   /192.168.52.130 consumer-1

TOPIC ：分区栏

PARTITION：分区栏

CURRENT-OFFSET：消费的偏移量

LOG-END-OFFSET：写入的偏移量

LAG：间隙，有多少数据没有被消费。LAG=LOG-END-OFFSET-CURRENT-OFFSET

二、 Kafka基础API

1 Topic基本操作 DML管理

1.1 新建maven工程

1.2 pom引入依赖

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->

<dependency>

    <groupId>org.apache.kafka</groupId>

    <artifactId>kafka-clients</artifactId>

    <version>2.2.0</version>

</dependency>

<!-- https://mvnrepository.com/artifact/log4j/log4j -->

<dependency>

    <groupId>log4j</groupId>

    <artifactId>log4j</artifactId>

    <version>1.2.17</version>

</dependency>

<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->

<dependency>

    <groupId>org.slf4j</groupId>

    <artifactId>slf4j-api</artifactId>

    <version>1.7.25</version>

</dependency>

<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->

<dependency>

    <groupId>org.slf4j</groupId>

    <artifactId>slf4j-log4j12</artifactId>

    <version>1.7.25</version>

</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->

<dependency>

    <groupId>org.apache.commons</groupId>

    <artifactId>commons-lang3</artifactId>

    <version>3.9</version>

</dependency>

1.3 配置文件增加日志配置项

1、resources目录下，新建log4j.properies

2、添加配置信息

log4j.rootLogger = info,console

log4j.appender.console = org.apache.log4j.ConsoleAppender

log4j.appender.console.Target = System.out

log4j.appender.console.layout = org.apache.log4j.PatternLayout

log4j.appender.console.layout.ConversionPattern =  %p %d{yyyy-MM-dd HH:mm:ss} %c - %m%n

1.4 增加案例demo-DML

1、链接Kafka

Properties props = new Properties();

props.put(AdminClientConfig.BOOTSTRAP_SERVER_CONFIG, "CentOSA:9092,CentOSB:9092,CentOSC:9092");

KafkaAdminClient adminClient = (KafkaAdminClient)KafkaAdminClient.create(props);

// 查看Topic列表

ListTopicResult topicResult = adminClient.listTopics();

Set<String> names = topicResult.names().get;

names.foreach(name - > Printf:out(name));

// 创建topic信息。默认是异步的

CreateTopicResult createTopicsResult = adminClient.createTopic(Arrays.asList(new NewTopic("topic02",3,(short)3);

// 改为同步创建需加上下面这句

createTopicsResult.all().get();

// 删除topic

DeleteTopicsResult deleteTopics = adminClient.deleteTopics(Arrays.asList("topic02","topic03"));

// 改为同步删除，加上下面代码

deleteTopics.all().get();

// 查看topic详情信息

DescribeTopicsResult dtr = adminClient.describeTopics(Array.asList("topic01"));

Map<String, TopicDescribe> topicDescriptionMap = dtr.all().get();

for(Map.Entry<String, TopicDescribption> entry : topicDescriptionMap.entrySet()) {

    System.out.prientln(entry.getKey() + "\t" + entry.getValue());

}

// 关闭AdminClient

adminClient.close();

1.2 生产者-demo

链接信息的客户端要换成ProducerConfig，且增加key和value的反序列化配置项SERIALIZER, StringSerizlizer.class

// 链接信息配置

Properties props = new Properties();

// 这里由原来的AdminClientConfig更改为ProducerConfig

props.put(ProducerConfig.BOOTSTRAP_SERVER_CONFIG, "CentOSA:9092,CentOSB:9092,CentOSC:9092");

// 配置Key的序列化

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerizlizer.class.getName());

// 配置值的序列化

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerizlizer.class.getName());

// 生产者

KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

// 发送10条消息给broker

for(int i=0; i<10; i++) {

    // 构造消息，消息有多种构造，包括带不带分区，时间戳信息等。

    // 这里例子用一个比较简单的topic+key+value的形式

    ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic01","key"+i, "value"+i);

    // 发送消息给服务器

    producer.send(record);

}

// 关闭

producer.close();

1.3 消费者-demo sub/assign

链接信息的客户端要换成ConsumerConfig，且根据生产者的序列化配置SERIALIZER，改为反序列化DESERIALIZER, StringDeserizlizer.class，切要配置消费者组信息，这里配置"g1"



// 链接信息配置

Properties props = new Properties();

// 这里由原来的AdminClientConfig更改为ConsumerConfig

props.put(ConsumerConfig.BOOTSTRAP_SERVER_CONFIG, "CentOSA:9092,CentOSB:9092,CentOSC:9092");

// 配置Key的反序列化DESERIALIZER

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class.getName());

// 配置值的反序列化DESERIALIZER

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class.getName());

// 消费者必须属于某一个消费者组

props.put(ConsumerConfig.GROUP_ID_CONFIG, "g1");

// 通过配置创建消费者

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

// 订阅相关的topics.支持正则。这里订阅所有以“topic”开头的topic

consumer.subscribe(Pattern.compile("^topic.*""));

// 遍历消息队列

while(true) {

    // 每一秒抓取一次

    ConsumeRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));

    // 从队列中取到数据

    if(!consumerRecords.isEmpty()) {

        Iterator<ConsumerRecord<String,String>>  recordIterator = consumerRecords.iterator();

        while(recordIterator.hasNext()) {

            // 获取一个消费消息

            ConsumerRecord<String, String> record = recordIterator.next();

            // 该消息所属topic

            String topic = record.topic();

            // 该消息所属分区

            int partition = record.partition();

            // 该消息在队列中的偏移量

            long offset = record.offset();

            // 该消息的key

            String key = record.key();

            // 该消息的value

            String value = record.value();

            // 该消息的时间戳

            long timestamp = record.timestamp();

            // 打印

            System.out.println(topic);

            System.out.println(partition);

            System.out.println(offset);

            System.out.println(key);

            System.out.println(value);

            System.out.println(timestamp);

        }

    }

}

不指定消费者组的消费，这个时候我们可以启动n个消费者，消费者之前不会互相均分。因为没有了组管理信息。消费实例之间不会互相影响

// 链接信息配置

Properties props = new Properties();

// 这里由原来的AdminClientConfig更改为ConsumerConfig

props.put(ConsumerConfig.BOOTSTRAP_SERVER_CONFIG, "CentOSA:9092,CentOSB:9092,CentOSC:9092");

// 配置Key的反序列化DESERIALIZER

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class.getName());

// 配置值的反序列化DESERIALIZER

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class.getName());

// 通过配置创建消费者

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

// 订阅相关的topics,手动指定消费分区，失去组管理特性

// 消费分区信息

List<TopicPartion> partions = Arrays.asList(new TopicPartion("topic01",0);

// 指定消费者的消费分区

consumer.assign(partions);

// 指定消费分区的位置,根据上文，可以知道我们从topic01的0分区从头消费

// consumer.seekToBeginning(partitions);

// 从某Topic的某分区的，某位置消费。例如从topic01的0分区的1位置开始消费

consumer.seek(new TopicPartition("topic01",0), 1);

// 遍历消息队列

while(true) {

    // 每一秒抓取一次

    ConsumeRecords<String,String> consumerRecords = consumer.poll(Duration.ofSeconds(1));

    // 从队列中取到数据

    if(!consumerRecords.isEmpty()) {

        Iterator<ConsumerRecord<String,String>>  recordIterator = consumerRecords.iterator();

        while(recordIterator.hasNext()) {

            // 获取一个消费消息

            ConsumerRecord<String, String> record = recordIterator.next();

            // 该消息所属topic

            String topic = record.topic();

            // 该消息所属分区

            int partition = record.partition();

            // 该消息在队列中的偏移量

            long offset = record.offset();

            // 该消息的key

            String key = record.key();

            // 该消息的value

            String value = record.value();

            // 该消息的时间戳

            long timestamp = record.timestamp();

            // 打印

            System.out.println(topic);

            System.out.println(partition);

            System.out.println(offset);

            System.out.println(key);

            System.out.println(value);

            System.out.println(timestamp);

        }

    }

}

1.4 自定义分区

当我们生产者指定key，我们的负载时根据key的hash来分配，当我们不指定key直接发送消息，我们通过轮询来保证负载均衡

如果指定了key，我们要自定义分区策略，不采用默认的key的hash。那么：

1、指定分区



for(int i=0; i<10; i++) {

    // "topic01", 1, "key"+i, "value"+i,其中第二个参数表示，我们都只发送到1分区，不在使用默认的轮询

    ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic01", 1, "key"+i, "value"+i);

    producer.send(record);

}

2、重写分区策略

Kafka的默认分区名称为DefaultPartitioner，策略实现了Partitioner。

看DefaultPartitioner的源码，可以看到，如果key为空则轮询，不为空则用kay的hash



public class UserDefinePartitioner implements Partitioner {

    // 重写策略，返回分区号

    // keyBytes：key的字节数组

    // valueBytes：value的字节数组

    // cluster ： 集群信息

    @Override

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        if(keyBytes == null) {

            // 参考默认的分区策略，也用轮询 TODO

            return 0;

        } else {

            // 我们自己定义我们想要的分区策略，大参考DefaultPartitioner

            // TODO

            return 0;

        }

        return 0;

    }

    @Override

    public void close() {

        Systen.out.println("close");

    }

    @Override

    public void configure(Map<String, ?> configs) {

        Systen.out.println("configure");

    }

}

1.5 序列化

在我们简单的生产者中，需要制定序列化配置。



// 配置Key的序列化

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerizlizer.class.getName());

// 配置值的序列化

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerizlizer.class.getName());

默认的StringSerizlizer 序列化，实现了Serizlizer

巴特西

《Kafka笔记》2、环境搭建、Topic管理