kafka 多线程消费

1、

   一、Kafka的消费并行度依赖Topic配置的分区数，如分区数为10，那么最多10台机器来并行消费（每台机器只能开启一个线程），或者一台机器消费（10个线程并行消费）。即消费并行度和分区数一致。

   二、（1）若是指定了某个分区,会只讲消息发到这个分区上

       （2）若是同时指定了某个分区和key,则也会将消息发送到指定分区上,key不起做用 

       （3）若是没有指定分区和key,那么将会随机发送到topic的分区中

       （4）若是指定了key,那么将会以hash<key>的方式发送到分区中 

2、多线程消费实例

     paritition 为3，broker为3，节点为3

一、生产者随机分区提交数据

    这也是一个比较关键步骤，只有随机提交到不一样的分区才能实现多分区消费； 
自定义随机分区：

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);
    }
}　　

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

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

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

 二、消费者

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

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

在消费方面得注意，这里得遍历全部分区，不然仍是只消费了一个区：

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();
            }
         }
        }

　　注意上面的线程为啥只有3个，这里得跟上面kafka的分区个数相对应起来，不然若是线程超过度区数量，那么只会浪费线程，由于即便使用3个以上的线程也只会消费三个分区，而少了则没法消费彻底。因此这里必须更上面的对应起来。