kafka 集群使用笔录

缘起:

今天业务说推送功能延时很严重.
遇到问题不要慌, 先来一波日志分析, 并发现了问题所在:
Kafka消息消费能力有限, 导致队列里有N多消息未来得及即时处理,
也就引发了延时处理数据的外在表现.

解决方案:

既然单节点Kafka 只能接一个消费端, 那么我们就来一组Kafka吧.

Round 1

我们搭建了三节点的Kafka集群, 也起用了三个消费端项目.
本以为这个任务就这么轻易的被解决了.

但是, 直接被打脸了.

为毛只有一个端能收到数据, 其他端 都不动呢...

Round 2

首先想到的是, 听说过partition , 但是没去配置, 莫非是这里的原因了呗.

broker.id=2
advertised.listeners=PLAINTEXT://10.10.1.107:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/app/kafka/kafka-logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.bytes=10737418240
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.10.1.107:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

先来一波配置, 加重启.

恩, 看起来是没问题, 三消费端重启一波.

咦 ?
为啥还不行, 明明每个消费端, 已经指定了消息的partition了.

Round 3

难道说, 生产者不是提交后, 自动向三个partition里写数据的么?
果然是我太天真了,

生产者, 居然有这样的API存在 :

    @Override
    public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
        ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
        return doSend(producerRecord);
    }

不管怎样, 先试一下吧, 生产端, int partitionId = (int) (new Date().getTime() % 3); 通过这样随机选定一个partition 号. 然后再向Kafka上传.

各种重启后, 三个消费端, 终于可以读到数据了~~~~~~

监测一段时间后, 发现吞吐能力十分稳定.
本次升级任务基本完成 .

后记

一个topic分parition上传, 消费端再分partition进行消费.
这么做, 跟分成多个topic上传, 有啥区别呢......