kafka操作5

集成springboot

springboot是一个在javaee开发中非常常用的组件。可以用于kafka的生产者，也可以用于kafka的消费者

1. 在idea中安装lombok插件，在plugins下搜索lombok然后在线安装即可，安装后注意重启
2. springboot环境准备，创建一个spring initializr

springboot生产者

1. 修改springboot的核心配置文件application.propeties，添加生产者相关信息

# 指定kafka的地址
spring.kafka.bootstrap-
servers=hadoop101:9092,hadoop102:9092,hadoop103:9092

# 指定key和value的序列化器
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
1
2
3
4
5
6
7

2. 创建controller从浏览器接收数据，并写入指定的topic
   application.properties

# 连接kafka集群
spring.kafka.bootstrap-servers=hadoop101:9092,hadoop102:9092

# key value的序列化
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

1
2
3
4
5
6
7

controller

package com.thenema.springbootkafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author 29467
 * @date 2022/10/19 16:45
 */
@RestController
public class ProducerController {

    @Autowired
    KafkaTemplate<String,String> kafka;

    @RequestMapping("/thenema")
    public String data(String msg){
        kafka.send("first", msg);
        return "ok";
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

在kafka启动消费者

bin/kafka-console-consumer.sh --bootstrap-server hadoop101:9092 --topic first
1

即可消费到数据

springboot消费者

consumer

package com.thenema.springbootkafka.controller;

import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

/**
 * @author 29467
 * @date 2022/10/19 16:57
 */
@Configuration
public class KafkaConsumer {
    @KafkaListener(topics = "first")
    public void consumerTopic(String msg){
        System.out.println("收到消息：" + msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

application.properties

# 连接kafka集群

# key value的反序列化
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# 消费者组id
spring.kafka.consumer.group-id=I am group id
1
2
3
4
5
6
7
8

kafka硬件配置选择

场景说明

100万日活，每人每天100条日志，每天总共的日志条数是100万 * 100条=1亿条。
1亿/24小时/60分/60秒=1150条/每秒钟。每条日志大小:0.5k - 2k（取1k)。
1150条/每秒钟 * 1k 1m/s
高峰期每秒钟:1150条 * 20倍=23000条
每秒多少数据量:20MB/s

服务器台数选择

服务器台数 = 2 * （生产者峰值生产速率【m/s】 * 副本数 / 100）+ 1

磁盘选择

kafka按照顺序读写，机械硬盘和固态硬盘顺序读写差不多
1亿条 * 1k = 100g
100g * 2个副本 * 3天 / 0.7（30%损耗） = 1t
建议三台服务器总的磁盘大小 大于1t

内存选择

kafka 内存 = 堆内存 (kafka 内部配置) + 页缓存（服务器内存）
堆内存：10-15g
设置堆内存：export KAFKA HEAP_ OPTS="-Xmx10G"
查看kafka进程：jps查看进程号
查看kafka的GC情况：使用jstat -gc xxx ls 10，查看YGC
根据kafka进程号，查看kafka的堆内存：jmpa -heap xxx
页缓存是linux服务器是内存，我们只需要保证一个segment（1g）中25%的数据在内存中就好，分区数 * 1g * 25% / 3 = xg
那么一台服务器等于10g + xg

CPU选择

num.io.threads = 8 负责写磁盘的线程数，整个参数值要占总核数的50%
num.replica.fetchers = 1 副本拉取线程数， 这个参数占总核数的50%的1/3
num.netwrok.thread = 3 数据传输线程数，这个参数占总核数的50%的2/3
建议32个 cpu core

网络选择

网络带宽 = 峰值吞吐量 约等于 20mb/s 选择千兆网卡即可
100Mbps 的单位是比特 10M/s的单位是字节 1byte = 8bit 100Mbps = 12.5M/s
一般的百兆网卡(100Mbps) 千兆网卡(1000Mbps) 万兆网卡(10000Mbps)

kafka服务器挂了

在生产环境中，如果某个kafka节点挂掉，处理方法如下

1. 先尝试重新启动一下，如果能启动正常，那直接解决
2. 如果重启不行，考虑增加内存，增加CPU，网络带宽
3. 如果将kafka整个节点误删除，如果副本数大于等于2，可以按照服役新节点的方式重新服役一个新节点，并执行负载均衡

集群压力测试

1. kafka压测

用kafka官方自带的脚本，对kafka进行压测。
- 生产者压测：kafka-producer-perf-test.sh
- 消费者压测：kafka-consumer-perf-test.sh
1
2
3

kafka producer压力测试

1. 创建一个test topic，设置为3个分区3个副本

bin/kafka-topics.sh --bootstrap-server hadoop101:9092 --create --replication-factor 3 --partitions 3 --topic test
1

2. 在/opt/kafka/bin目录下面有这两个文件，可以测试一下

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=0
1

3. 调整batch.size大小，batch.size默认值是16k，本次实验将batch.size设置为32k

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=32648 linger.ms=0
1

4. 调整linger.ns时间，linger.ms默认是0ms，本次实验linger.ms设置为50ms

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50
1

5. 调整压缩方式，默认的压缩方式是none，本次实验compression.type设置为snappy

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50 conpression.type=snappy
1

压缩方式设置为zstd

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50 conpression.type=zstd
1

gzip

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50 conpression.type=gzip
1

lz4

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50 conpression.type=lz4
1

6. 调整缓存大小，默认生产者端缓存大小32m，本次实验buffer.memory设置为64m

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-records 1000000 --throughput 10000 bootstrap.servers=hadoop101:9092, hadoop102:9092, hadoop103:9092 batch.size=16384 linger.ms=50 buffer.memory=67108864
1

kafka consumer压力测试

1. 修改/opt/kafka/config/consumer.properties文件中的一次拉取条数为500

max.poll.records=500
1

2. 消费100万条日志进行压测

bin/kafka-consumer-perf-test.sh --bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092 --topic test --messages1000000 --consumer.config config/consumer.propertiese
1

3. 一次拉取条数为2000

max.poll.records=500
1

4. 调整fetch.max.bytes大小为100m，修改/opt/kafka/config/consumer.properties文件中的拉取一批数据大小为100m

fetch.max.bytes=104857600
1