1.mq原則
數據不能多，也不能少，不能多是說消息不能重復消費，這個我們上一節已解決；不能少，就是說不能丟失數據。如果mq傳遞的是非常核心的消息，支撐核心的業務，那么這種場景是一定不能丟失數據的。
2.丟失數據場景
丟數據一般分為兩種，一種是mq把消息丟了，一種就是消費時將消息丟了。下面從rabbitmq和kafka分別說一下，丟失數據的場景，
（1）rabbitmq
A:生產者弄丟了數據
生產者將數據發送到rabbitmq的時候，可能在傳輸過程中因為網絡等問題而將數據弄丟了。
B:rabbitmq自己丟了數據
如果沒有開啟rabbitmq的持久化，那么rabbitmq一旦重啟，那么數據就丟了。所依必須開啟持久化將消息持久化到磁盤，這樣就算rabbitmq掛了，恢復之后會自動讀取之前存儲的數據，一般數據不會丟失。除非極其罕見的情況，rabbitmq還沒來得及持久化自己就掛了，這樣可能導致一部分數據丟失。
C：消費端弄丟了數據
主要是因為消費者消費時，剛消費到，還沒有處理，結果消費者就掛了，這樣你重啟之后，rabbitmq就認為你已經消費過了，然后就丟了數據。
（2）kafka
A:生產者弄丟了數據
生產者沒有設置相應的策略，發送過程中丟失數據。
B:kafka弄丟了數據
比較常見的一個場景，就是kafka的某個broker宕機了，然后重新選舉partition的leader時。如果此時follower還沒來得及同步數據，leader就掛了，然后某個follower成為了leader，他就少了一部分數據。
C:消費者弄丟了數據
消費者消費到了這個數據，然后消費之自動提交了offset，讓kafka知道你已經消費了這個消息，當你准備處理這個消息時，自己掛掉了，那么這條消息就丟了。
3.如何防止消息丟失
（1）rabbitmq
A:生產者丟失消息
①：可以選擇使用rabbitmq提供是事物功能，就是生產者在發送數據之前開啟事物，然后發送消息，如果消息沒有成功被rabbitmq接收到，那么生產者會受到異常報錯，這時就可以回滾事物，然后嘗試重新發送；如果收到了消息，那么就可以提交事物。
channel.txSelect();//開啟事物
try{
//發送消息
}catch(Exection e){
channel.txRollback()；//回滾事物
//重新提交
}
缺點：rabbitmq事物已開啟，就會變為同步阻塞操作，生產者會阻塞等待是否發送成功，太耗性能會造成吞吐量的下降。
②：可以開啟confirm模式。在生產者哪里設置開啟了confirm模式之后，每次寫的消息都會分配一個唯一的id，然后如何寫入了rabbitmq之中，rabbitmq會給你回傳一個ack消息，告訴你這個消息發送OK了；如果rabbitmq沒能處理這個消息，會回調你一個nack接口，告訴你這個消息失敗了，你可以進行重試。而且你可以結合這個機制知道自己在內存里維護每個消息的id，如果超過一定時間還沒接收到這個消息的回調，那么你可以進行重發。
//開啟confirm
channel.confirm();
//發送成功回調
public void ack(String messageId){
}
// 發送失敗回調
public void nack(String messageId){
//重發該消息
}
二者不同
事務機制是同步的，你提交了一個事物之后會阻塞住，但是confirm機制是異步的，發送消息之后可以接着發送下一個消息，然后rabbitmq會回調告知成功與否。
一般在生產者這塊避免丟失，都是用confirm機制。
B:rabbitmq自己弄丟了數據
設置消息持久化到磁盤。設置持久化有兩個步驟：
①創建queue的時候將其設置為持久化的，這樣就可以保證rabbitmq持久化queue的元數據，但是不會持久化queue里面的數據。
②發送消息的時候講消息的deliveryMode設置為2，這樣消息就會被設為持久化方式，此時rabbitmq就會將消息持久化到磁盤上。
必須要同時開啟這兩個才可以。
而且持久化可以跟生產的confirm機制配合起來，只有消息持久化到了磁盤之后，才會通知生產者ack，這樣就算是在持久化之前rabbitmq掛了，數據丟了，生產者收不到ack回調也會進行消息重發。
C:消費者弄丟了數據
使用rabbitmq提供的ack機制，首先關閉rabbitmq的自動ack，然后每次在確保處理完這個消息之后，在代碼里手動調用ack。這樣就可以避免消息還沒有處理完就ack。
（2）kafka
A:消費端弄丟了數據
關閉自動提交offset，在自己處理完畢之后手動提交offset，這樣就不會丟失數據。
B:kafka弄丟了數據
一般要求設置4個參數來保證消息不丟失：
①給topic設置 replication.factor參數：這個值必須大於1，表示要求每個partition必須至少有2個副本。
②在kafka服務端設置min.isync.replicas參數：這個值必須大於1，表示 要求一個leader至少感知到有至少一個follower在跟自己保持聯系正常同步數據，這樣才能保證leader掛了之后還有一個follower。
③在生產者端設置acks=all：表示 要求每條每條數據，必須是寫入所有replica副本之后，才能認為是寫入成功了
④在生產者端設置retries=MAX(很大的一個值，表示無限重試)：表示 這個是要求一旦寫入事變，就無限重試
C：生產者弄丟了數據
如果按照上面設置了ack=all，則一定不會丟失數據，要求是，你的leader接收到消息，所有的follower都同步到了消息之后，才認為本次寫成功了。如果沒滿足這個條件，生產者會自動不斷的重試，重試無限次。
上一篇《如何保證消息不重復消費》
下一篇《如何保證消息按順序執行》