storm通過(guò)保證數(shù)據(jù)至少被處理一次來(lái)保證數(shù)據(jù)的完整性,由于元祖可以重發(fā),對(duì)于一些需要數(shù)據(jù)精確的場(chǎng)景,可以考慮用storm trident實(shí)現(xiàn)窍仰。
傳統(tǒng)的事物型拓?fù)渲写嬖趲追Nbolt：
1.1 BasicBolt
這是最基本的Bolt,BasicBolt每次只能處理一個(gè)tuple,而且必須等前一個(gè)tuple成功處理后下一個(gè)tuple才能繼續(xù)處理,顯然效率不高。
1.2 BatchBolt
storm的一個(gè)優(yōu)勢(shì)就是能夠批量處理tuple,BatchBolt支持批量處理tuple,每一個(gè)batch中的tuple都會(huì)調(diào)用execute(),處理完成后調(diào)用finishBatch方法她奥。 1.3 Committer BatchBolt
標(biāo)記為Committer的BatchBolt和基本的BasicBolt的區(qū)別在于二者調(diào)用finishBatch()的時(shí)機(jī)不同,標(biāo)記為Committer的BatchBolt在提交階段就會(huì)調(diào)用finishBatch()讯泣。
二阵漏、storm trident的使用
storm目前的版本已經(jīng)將事物拓?fù)涞膶?shí)現(xiàn)封裝trident,trident目前支持3種不同的事物接口预侯，一種是非事物型的(不介紹,因?yàn)榛静挥?,一種是事務(wù)性的TransactionalTridentKafkaSpout,而我們比較常用的是透明型事物OpaqueTridentKafkaSpout(事務(wù)型應(yīng)用最重要的一點(diǎn)是要判斷一批消息是新的還是已來(lái)過(guò)的)。
2.1 TransactionalTridentKafkaSpout
原理是每次在數(shù)據(jù)庫(kù)中存了txid,IPartitionedTransactionalSpout的每一個(gè)tuple都會(huì)綁定在固定的批次(batch)中甲雅。 一個(gè)批次無(wú)論重發(fā)多少次解孙，它也只有一個(gè)唯一且相同的事務(wù)ID坑填，它所包含的內(nèi)容都是完全一致的,而一個(gè)tuple無(wú)論被重發(fā)多少次只會(huì)在同一個(gè)批次里。 但貌似目前TransactionalTridentKafkaSpout有個(gè)bug,啟動(dòng)會(huì)報(bào)：classCastException(非代碼問(wèn)題)
具體可參考：
Java代碼
1. issue:https://issues.apache.org/jira/browse/STORM-1728
然而我們可以想到的是,IPartitionedTransactionalSpout會(huì)有一個(gè)問(wèn)題,假設(shè)一批消息在被bolt消費(fèi)過(guò)程中失敗了弛姜，需要spout重發(fā)脐瑰，此時(shí)如果正巧遇到消息發(fā)送中間件故障，例如某一個(gè)分區(qū)不可讀廷臼，spout為了保證重發(fā)時(shí)每一批次包含的tuple一致苍在，它只能等待消息中間件恢復(fù)，也就是卡在那里無(wú)法再繼續(xù)發(fā)送給bolt消息了荠商，直至消息中間件恢復(fù)(因?yàn)樗仨毎l(fā)送一樣的Batch)寂恬。 2.2 OpaqueTridentKafkaSpout IOpaquePartitionedTransactionalSpout不保證每次重發(fā)一個(gè)批次的消息所包含的tuple完全一致。也就是說(shuō)某個(gè)tuple可能第一次在txid=1的批次中出現(xiàn)莱没，后面有可能在txid=3的批次中出現(xiàn)初肉。這種情況只出現(xiàn)在當(dāng)某一批次消息消費(fèi)失敗需要重發(fā)且恰巧消息中間件故障時(shí)。這時(shí)饰躲，IOpaquePartitionedTransactionalSpout不是等待消息中間件故障恢復(fù),而是先讀取可讀的partition牙咏。例如txid=1的批次在消費(fèi)過(guò)程中失敗了，需要重發(fā)属铁，恰巧消息中間件的16個(gè)分區(qū)有1個(gè)分區(qū)(partition=3)因?yàn)楣收喜豢勺x了眠寿。這時(shí)候IOpaquePartitionedTransactionalSpout會(huì)先讀另外的15個(gè)分區(qū)，完成txid=1這個(gè)批次的發(fā)送焦蘑，這時(shí)候同樣的批次其實(shí)包含的tuple已經(jīng)少了。假設(shè)在txid=3時(shí)消息中間件的故障恢復(fù)了盒发，那之前在txid=1且在分區(qū)partition=3的還沒(méi)有被發(fā)送的tuple會(huì)被重新發(fā)送例嘱， 包含在txid=3的批次中,所以其不保證每批次的batch包含的tuple是一樣的。
