问题描述:
在消费端能够正常消费到Kafka数据并成功生产到producer topic 中,当将kafka的一台机器关机之后,正常情况下应该是 消费端是不受影响的。因为有还有两台的负载机器。问题就是一台机器停止运行之后,消费端酒 shutdown 而无法重新starting
解决方式 :
在如下代码中。
1 public void run(){ 2 try{ 3 System. out.println( "Consumer...."); 4 Map<String,Integer> topicCountMap = new HashMap<String,Integer>(); 5 topicCountMap.put( topic, partitionNum); 6 Map<String,List<KafkaStream< byte[], byte[]>>> consumerMap = consumer.createMessageStreams( topicCountMap); 7 List<KafkaStream< byte[], byte[]>> partitions = consumerMap.get( topic); 8 9 threadPool = Executors. newCachedThreadPool(); 10 for(KafkaStream< byte[], byte[]> partition : partitions){ 11 threadPool.execute( new MessageFetcher(partition,producer )); 12 } 13 } catch(Exception ex){ 14 logger.info( "KafkaConsumer-> Run -> ErrInfo : " +ex.getMessage()); 15 close(); 16 } 17 }
有一个 partitionNum,在代码中的可配置值为 private int partitionNum = 3;
把partitionNum 改为 1 即可解决此问题。
问题跟踪源码分析:
partitionNUm 改为 1 , 此处的Num 为ThreadNum ,因为kafka内部实现中,都为多线程, partition为1时,此时有一个backingQueue1,三个fetch thread 线程,该topic分布在几个node上就有几个 fetch thread 每个fetch thread 会于kafka broker建立一个连接,3个fetch thread线程去拉去消息数据,最终防盗blockingQueue中,等到consumer thread来消费。