【Kafka】kafka-2.13_2.7.0 集群的安装配置 与 最佳实践_kafka error while accepting connection too many op

OS

---

1、IO调优：https://blog.csdn.net/yfkscu/article/details/40082057 （只执行 关闭 swap）

2、socket buffer size（暂时使用默认）

cat  /proc/sys/net/core/wmem_max
cat /proc/sys/net/core/rmem_max

/proc/sys/net/ipv4/tcp_rmem
/proc/sys/net/ipv4/tcp_wmem
1
2
3
4
5

3、增加 ulimit

为避免Kafka报 Too many open files 错误，需要增加最大文件打开数：

cd /etc/security/limits.d/
vim kafka.conf
# 编辑
*  hard   nofile  100000
*  soft   nofile  100000
# 保存
1
2
3
4
5
6

为了提高Kafka的性能，/etc/sysctl.conf文件需增加以下配置（暂时不做配置）：

vm.swappiness=1//物理内存使用率超过99%时，使用swap分区，这样避免使用swap分区，否则影响性能。
vm.dirty_background_ratio=5//脏页达到系统内存5%时，后台刷新页面
vm.dirty_ratio=70//脏页占到系统内存70%时，刷新进磁盘
vm.max_map_count=262144//允许一个进程在虚拟内存区域的最大数量
net.core.wmen_default=131072//socket写缓冲区
net.core.rmem_default=131072//socket读缓冲区
net.core.wmen_max=2097152//socket最大写缓冲区
net.core.rmem_max=2097152//socket最大读缓冲区
1
2
3
4
5
6
7
8

zookeeper

只需要修改 clientPort、dataDir 和 dataLogDir 即可。

tickTime=2000  最小时间单元，单位毫秒
initLimit=10  Follower启动，与Leader同步的时间倍数
syncLimit=5  Leader与Follower之间进行心跳检测的最大延时时间倍数
dataDir=$dir1/zookeeper  快照文件存储路径
dataLogDir=$dir2/zookeeper  事务日志文件存储路径
maxClientCnxns=60 单个IP与zk的最大连接数
clientPort=12999
1
2
3
4
5
6
7

JVM

一般kafka自身的jvm堆内存，分配个10G左右就够了，剩下的内存全部都留给os cache

cd /usr/local/kafka_2.13-2.7.0/bin
vim kafka-server-start.sh

# KAFKA_HEAP_OPTS，默认都是 1G
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx8G -Xms8G"
fi
1
2
3
4
5
6
7

cd /usr/local/kafka_2.13-2.7.0/bin
vim kafka-server-start.sh

# Generic jvm settings you want to add
if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS="-Xmx8g -Xms8g -XX:MetaspaceSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50
 -XX:MaxMetaspaceFreeRatio=85"
fi
1
2
3
4
5
6
7
8

Broker

num.network,threads=9
num.io.threads=32
log.retention.hours=168
delete.topic.enable=true
message.max.bytes=7340032
zookeeper.connection.timeout.ms=18000
auto.create.topics.enable=false
1
2
3
4
5
6
7

一个 consumer group 中的consumer数与Partition 数一样，一个 consumer group 可以订阅一个或多个 topic。

Producer

buffer.memory 67108864
batch.size  10485760
max.request.size=6291456
linger.ms 100 
max.request.size
Compression.type
acks 1
retries 10	
retry.backoff.ms 500 (重试间隔）
1
2
3
4
5
6
7
8
9

batch.size：该值设置越大，吞吐越大，但延迟也会越大；

linger.ms：表示 batch 的超时时间，该值越大，吞吐越大、但延迟也会越大；

Consumer

---

组内所有消费者协调一起订阅一个主题的所有分区。每个分区只能由一个消费者组内的一个消费者实例消费。

理想情况下，消费者实例应该等于该消费者组订阅的主题分区总数。一般不推荐设置大于总分区数的消费者实例。

消费组的重平衡，主要发生于以下情景：

组成员发生变更，如新的消费者实例加入，或者离开

订阅主题数发生变更，如消费者组使用正则方式订阅主题，当新增一个符合条件的主题时。
订阅主题分区发生变更。
消费组的重平衡时会阻塞所有消费者。目前 无法解决这个重平衡机制问题，但是有一些优化途径，尽量避免此类情况发生。

session.timeout.ms = 10s
heartbeat.interval.ms = 3s
max.poll.records 默认 500，一次 poll 的数量
max.poll.interval.ms 默认是 5 分钟。如果5分钟内无法消费完 poll 方法返回的消息，那么消费者会自动离组。
fetch.message.max.bytes  8388608
max.partition.fetch.bytes 8388608
1
2
3
4
5
6

针对大消息的设置策略

参考：https://studygolang.com/articles/24305?fr=sidebar

producer：
max.request.size=6291456（6M）

broker：
message.max.bytes=7340032（7M）

consumer：
fetch.max.bytes=8388608（8M）

max.request.size < message.max.bytes < fetch.max.bytes

batch.size > max.request.size

Producer 每次发送的消息封装成 ProducerRecord，然后利用消息累加器 RecordAccumulator 添加到 ProducerBatch 中，由于每次创建 ProducerBatch 都需要分配一个 batch.size 大小的内存空间，频繁创建和关闭会导致性能极大开销，所以 RecordAccumulator 内部有个 BufferPool，它实现了缓存的复用，只不过只针对 batch.size 大小的 BufferByte 进行复用，如果大于 batch.size 的 ProducerBatch，它并不会加入 BufferPool 中，也就不会复用。

kafka 常见问题

---

Too many open files

日志报错

ERROR Error while accepting connection (kafka.network.Acceptor)
java.io.IOException: Too many open files
1
2

解决办法

netstart -unltpa|grep 9096
ulimit -n
1
2

kafka 常用命令

---

python 操作 kafka

---

pykafka 2.3.1、kafka-python 1.1.1和confluent-kafka-python 0.9.1

以三者的producer为例，confluent-kafka-python >> pykafka(librdkafka backend) >> pykafka > kafka-python；以三者的consumer为例，confluent-kafka-python >> pykafka(librdkafka backend) >> kafka-python > pykafka。

kafka-python

2、consumer

常规消费

from kafka import KafkaConsumer,TopicPartition
consumer = KafkaConsumer(
                          'tbl_stream',
                          group_id="eztest",
                          bootstrap_servers=['192.168.1.1:9092']
                )
print("连接成功...")
res = consumer.poll(100)

consumer.seek_to_beginning()
for msg in consumer:
    print(msg)
1
2
3
4
5
6
7
8
9
10
11
12

指定位置消费

consumer = KafkaConsumer(

                          group_id="eztest",
                          bootstrap_servers=['192.168.1.1:9092']
                )
print("连接成功...")
partition = TopicPartition('tbl_stream_17765',5)
consumer.assign([partition])
consumer.seek(partition, start)
for msg in consumer:
    print(msg)
    print(msg.partition,msg.offset)
1
2
3
4
5
6
7
8
9
10
11
12

根据时间戳查询每个分区的位置

from kafka import KafkaConsumer,TopicPartition
import time

consumer = KafkaConsumer(
                          group_id="eztest",
                          bootstrap_servers=['192.168.1.1:9092']
                )
print("连接成功...")
topic = "tbl_stream_8694"

begin_time = 1623081600000
begin_search = {}
for partition in [0,1,2,3,4,5]:
    tp = TopicPartition(topic, partition)
    begin_search[tp] = begin_time

begin_offset = consumer.offsets_for_times(begin_search)
print(begin_offset)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

pykafka

from pykafka import KafkaClient

client = KafkaClient(hosts='120.133.1.1:9096,120.133.1.2:9096,120.133.1.3:9096')

print(client.topics)

topic = client.topics['tbl_stream_2098']

# 获取消费者

# consumer = topic.get_simple_consumer('abcd', auto_commit_enable=True, auto_commit_interval_ms=1)
consumer = topic.get_simple_consumer()
print(consumer)
suc = 0
for message in consumer:
    suc += 1
    print(suc,message.partition,message.offset)
    print(message.value.decode())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

参考资料

---

知乎 kafka 配置：https://www.zhihu.com/search?type=content&q=kafka%20%E9%85%8D%E7%BD%AE