在使用kafka high-level的consumer，使用多線程消費(fèi)數(shù)據(jù)時(shí)報(bào)錯(cuò)，簡單分析一下原因下載 ，ConsumerIterator取不到消息時(shí)會(huì)阻塞，并且將內(nèi)部狀態(tài)置為FAILED，當(dāng)其他線程訪問時(shí)就會(huì)拋出異常。

公司主營業(yè)務(wù)：成都做網(wǎng)站、成都網(wǎng)站設(shè)計(jì)、移動(dòng)網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競爭能力。創(chuàng)新互聯(lián)建站是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊(duì)。公司秉承以“開放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對(duì)我們的高要求，感謝他們從不同領(lǐng)域給我們帶來的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶帶來驚喜。創(chuàng)新互聯(lián)建站推出太子河免費(fèi)做網(wǎng)站回饋大家。

 

 

Java代碼  

1.  def hasNext(): Boolean = {  

2.     if(state == FAILED)         //處于FAILED狀態(tài)時(shí)，另外線程訪問會(huì)直接異常  

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              //重置了狀態(tài)  

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這里會(huì)阻塞  

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. }  

網(wǎng)站標(biāo)題：kafkahigh-levelconsumer多線程訪
分享URL：http://chinadenli.net/article10/gjcggo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供標(biāo)簽優(yōu)化、App開發(fā)、網(wǎng)站排名、建站公司、全網(wǎng)營銷推廣、品牌網(wǎng)站設(shè)計(jì)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)