集群下的kafka实现多线程消费

亦凉 2022-05-28 11:51 226阅读 0赞

https://blog.csdn.net/u011622226/article/details/72500713

上一篇文章讲述了如何部署kafka集群，而这篇文章则来探讨一下如何使用多线程消费，提高消费能力，保障数据的时效性。而实现多线程消费其实很简单，只需要三步即可：

## 一：kafka集群配置 ##

多线程消费，说白了就是多区消费，kafka可以给topic设置多个partition，从而实现生产的时候提交到不同的分区，以减少统一区块的压力。而消费则是从不同的分区里拿数据进行消费。

1.首先修改server.properties里：  
num.partitions=3  
这里等于3是因为本人的集群是用了三台机子，也就是3个broker，所以设置成3，具体数值可以根据集群情况设置。  
2.创建topic：  
bin/kafka-topics.sh –create –zookeeper 192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181 –replication-factor 3 –partitions 3 –topic test  
这里的3对应上面的配置里的num.partitions=3  
3.查看topic信息：  
./kafka-topics.sh –zookeeper 192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181 –topic test–describe  
会有如下显示表示创建成功：  
Topic:test PartitionCount:3 ReplicationFactor:3 Configs:  
Topic: test Partition: 0 Leader: 2 Replicas: 2,0,1 Isr: 0,1,2  
Topic: test Partition: 1 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2  
Topic: test Partition: 2 Leader: 1 Replicas: 1,2,0 Isr: 0,1,2

## 二：生产者随机分区提交数据 ##

这也是一个比较关键步骤，只有随机提交到不同的分区才能实现多分区消费；  
这里借用了一片代码，自定义随机分区：

public class MyPartition implements Partitioner{
        
         private static Logger LOG = LoggerFactory.getLogger(MyPartition.class); 
        @Override
        public void configure(Map<String, ?> arg0) {
            // TODO Auto-generated method stub
        }
    
        @Override
        public void close() {
            // TODO Auto-generated method stub
        }
    
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value,
                byte[] valueBytes, Cluster cluster) {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            int partitionNum = 0;
            try {
                partitionNum = Integer.parseInt((String) key);
            } catch (Exception e) {
                partitionNum = key.hashCode() ;
            }
    //        System.out.println("kafkaMessage topic:"+ topic+" |key:"+ key+" |value:"+value);
            return Math.abs(partitionNum  % numPartitions);
        }
    }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22
 *  23
 *  24
 *  25
 *  26
 *  27

然后在初始化kafka生产者配置的时候修改如下配置

props.put("partitioner.class", properties.getProperty("com.mykafka.MyPartition"));

*  1

这样就实现了kafka生产者随机分区提交数据。

## 三：消费者 ##

最后一步就是消费者，修改单线程模式为多线程，这里的多线程实现方式有很多，本例知识用了最简单的固定线程模式：

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
            for (int i = 0; i < 3; i++) {
                fixedThreadPool.execute(new Runnable() {
                    @Override
                    public void run() {
                        kafkaConsumerService.getInstance();
                    }
                });
            }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9

在消费方面得注意，这里得遍历所有分区，否则还是只消费了一个区：

ConsumerRecords<String, String> records = consumer.poll(1000);
            for (TopicPartition partition : records.partitions()) {  
                List<ConsumerRecord<String, String>> partitionRecords = records  
                        .records(partition); 
            for (ConsumerRecord<String, String> record : partitionRecords) {
                System.out.println(
                        "message==>key:" + record.key() + " value:" + record.value() + " offset:" + record.offset()
                        + " 分区:" + record.partition());
                if (record.value() == null || record.key() == null) {
                    consumer.commitSync();
                } else {
                    // dealMessage
                    KafkaServer.dealMessage(record.key(),record.value(),consumer);
    //              consumer.commitSync();
                }
             }
            }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17

注意上面的线程为啥只有3个，这里得跟上面kafka的分区个数相对应起来，否则如果线程超过分区数量，那么只会浪费线程，因为即使使用3个以上的线程也只会消费三个分区，而少了则无法消费完全。所以这里必须更上面的对应起来。  
至此，多线程消费就已经完毕。这也只是本人摸索以及借鉴的经验，如有错误或者问题，请留言交流，互相学习。