kafkaspot在ack機制下如何保證記憶體不溢

storm框架中的kafkaspout類實現的是baserichspout，它裡面已經重寫了fail和ack方法，所以我們的bolt必須實現ack機制，就可以保證訊息的重新傳送；如果不實現ack機制，那麼kafkaspout就無法得到訊息的處理響應，就會在超時以後再次傳送訊息，導致訊息的重**送。

但是回想一下我們自己寫乙個spout類實現baserichspout並讓他具備訊息重發，那麼我們是會在我們的spout類裡面定義乙個map集合，並以msgid作為key。

public class myspout extends baserichspout 

public void open(map conf, topologycontext context, spoutoutputcollector collector) 
public void nexttuple() 
@override
public void ack(object msgid) 
@override
public void fail(object msgid) 
}

其實並沒有，回想kafka的原理，kafka會為每乙個consumergroup保留一些metadata資訊–當前消費的訊息的position，也即offset。這個offset由consumer控制。正常情況下consumer會在消費完一條訊息後線性增加這個offset。當然，consumer也可將offset設成乙個較小的值，重新消費一些訊息。也就是說，kafkaspot在消費kafka的資料是，通過offset讀取到訊息並傳送給bolt後，kafkaspot只是儲存者當前的offset值。

當失敗或成功根據msgid查詢offset值，然後再去kafka消費該資料來確保訊息的重新傳送。

那麼雖然offset資料小，但是當offset的資料量上去了還是會記憶體溢位的？

其實並沒有，kafkaspout發現快取的資料超過限制了，會把某端的資料清理掉的。

kafkaspot中傳送資料的**

collector.emit(tup, new kafkamessageid(_partition, toemit.offset));

它不快取已經傳送出去的資料資訊。

當他接收到來至bolt的響應後，會從接收到的msgid中得到offset。以下是從原始碼中折取的關鍵**：

public void ack(object msgid) 

} m.ack(id.offset);
public void ack(long offset) public void fail(object msgid) 
}  m.fail(id.offset);
public void fail(long offset) 
sortedset_pending = new treeset();
sortedsetfailed = new treeset();

kafkaspot在ack機制下如何保證記憶體不溢

storm框架中的kafkaspout類實現的是baserichspout，它裡面已經重寫了fail和ack方法，所以我們的bolt必須實現ack機制，就可以保證訊息的重新傳送 如果不實現ack機制，那麼kafkaspout就無法得到訊息的處理響應，就會在超時以後再次傳送訊息，導致訊息的重 送。但是...

kafkaspot在ack機制下如何保證記憶體不溢

storm框架中的kafkaspout類實現的是baserichspout，它裡面已經重寫了fail和ack方法，所以我們的bolt必須實現ack機制，就可以保證訊息的重新傳送 如果不實現ack機制，那麼kafkaspout就無法得到訊息的處理響應，就會在超時以後再次傳送訊息，導致訊息的重 送。但是...

kafkaspot在ack機制下如何保證記憶體不溢

storm框架中的kafkaspout類實現的是baserichspout，它裡面已經重寫了fail和ack方法，所以我們的bolt必須實現ack機制，就可以保證訊息的重新傳送 如果不實現ack機制，那麼kafkaspout就無法得到訊息的處理響應，就會在超時以後再次傳送訊息，導致訊息的重 送。但是...