kafka high-level consumer 多線程訪

在使用kafka high-level的consumer，使用多線程消費數據時報錯，簡單分析一下原因下載 ，ConsumerIterator取不到消息時會阻塞，并且將內部狀態置為FAILED，當其他線程訪問時就會拋出異常。

Java代碼  

1.  def hasNext(): Boolean = {  

2.     if(state == FAILED)         //處于FAILED狀態時，另外線程訪問會直接異常  

3.       throw new IllegalStateException("Iterator is in failed state")  

4.     state match {  

5.       case DONE => false  

6.       case READY => true  

7.       case _ => maybeComputeNext()  

8.     }  

9.   }  

10.   

11.   

12.   def maybeComputeNext(): Boolean = {  

13.     state = FAILED              //重置了狀態  

14.     nextItem = Some(makeNext())          

15.     if(state == DONE) {  

16.       false  

17.     } else {  

18.       state = READY  

19.       true  

20.     }  

21.   }  

22.   下載

23.   

24. protected def makeNext(): MessageAndMetadata[K, V] = {  

25.     var currentDataChunk: FetchedDataChunk = null  

26.     // if we don't have an iterator, get one  

27.     var localCurrent = current.get()  

28.     if(localCurrent == null || !localCurrent.hasNext) {  

29.       if (consumerTimeoutMs < 0)  

30.         currentDataChunk = channel.take             //channel是BlockingQueue這里會阻塞  

31.   

32.       else {  

33.         currentDataChunk = channel.poll(consumerTimeoutMs, TimeUnit.MILLISECONDS)  

34.         if (currentDataChunk == null) {  

35.           // reset state to make the iterator re-iterable  

36.           resetState()  

37.           throw new ConsumerTimeoutException  

38.         }  

39.       }  

40. //省略部分代碼  

41. }