接這篇

在上文中，主要實現了可靠模式的consumer。而可靠模式的sender實現的相對簡略，主要通過rabbitTemplate來完成。
本以為這樣的實現基本是沒有問題的。但是前段時間做了一個性能壓力測試，但是發現在使用rabbitTemplate時，會有一定的丟資料問題。

當時的場景是用30個執行緒，無間隔的向rabbitmq傳送資料，但是當執行一段時間後發現，會出現一些connection closed錯誤，rabbitTemplate雖然進行了自動重連，但是在重連的過程中，丟失了一部分資料。當時傳送了300萬條資料，丟失在2000條左右。
這種丟失率，對於一些對一致性要求很高的應用(比如扣款，轉賬)來說，是不可接受的。

在google了很久之後，在stackoverflow上找到rabbitTemplate作者對於這種問題的解決方案，他給的方案很簡單，單純的增加connection數：

connectionFactory.setChannelCacheSize(100);

修改之後，確實不再出現connection closed這種錯誤了，在傳送了3000萬條資料後，一條都沒有丟失。
似乎問題已經完美的解決了，但是我又想到一個問題：當我們的網路在發生抖動時，這種方式還是不是安全的？
換句話說，如果我強制切斷客戶端和rabbitmq服務端的連線，資料還會丟失嗎？

為了驗證這種場景，我重新發送300萬條資料，在傳送過程中，在rabbitmq的管理介面上點選強制關閉連線：

然後發現，仍然存在丟失資料的問題。

看來這個問題，沒有想象中的那麼簡單了。

在閱讀了部分rabbitTemplate的程式碼之後發現：
1 rabbitTemplate的ack確認機制是非同步的
2 這種確認機制是一種事後發現機制，並不能同步的發現問題
也就是說，即便打開了

connectionFactory.setPublisherConfirms(true);
rabbitTemplate.setMandatory(true);

並且實現了：

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if
 
 (!ack) {
                log.info("send message failed: " + cause + correlationData.toString());
            } 
        });

依舊是不安全的。
rabbitTemplate的傳送流程是這樣的：
1 傳送資料並返回(不確認rabbitmq伺服器已成功接收)
2 非同步的接收從rabbitmq返回的ack確認資訊
3 收到ack後呼叫confirmCallback函式
注意：在confirmCallback中是沒有原message的，所以無法在這個函式中呼叫重發，confirmCallback只有一個通知的作用

在這種情況下，如果在2，3步中任何時候切斷連線，我們都無法確認資料是否真的已經成功傳送出去，從而造成資料丟失的問題。

最完美的解決方案只有1種：
使用rabbitmq的事務機制。
但是在這種情況下，rabbitmq的效率極低，每秒鐘處理的message在幾百條左右。實在不可取。
第二種解決方式，使用同步的傳送機制，也就是說，客戶端傳送資料，rabbitmq收到後返回ack，再收到ack後，send函式才返回。程式碼類似這樣：

建立channel
send message
wait for ack(or 超時)
close channel
返回成功or失敗

同樣的，由於每次傳送message都要重新建立連線，效率很低。

基於上面的分析，我們使用一種新的方式來做到資料的不丟失。
在rabbitTemplate非同步確認的基礎上
1 在本地快取已傳送的message
2 通過confirmCallback或者被確認的ack，將被確認的message從本地刪除
3 定時掃描本地的message，如果大於一定時間未被確認，則重發

當然了，這種解決方式也有一定的問題：
想象這種場景，rabbitmq接收到了訊息，在傳送ack確認時，網路斷了，造成客戶端沒有收到ack，重發訊息。（相比於丟失訊息，重發訊息要好解決的多，我們可以在consumer端做到冪等）。
自動重試的程式碼如下：

public class RetryCache {
    private MessageSender sender;
    private boolean stop = false;
    private Map<String, MessageWithTime> map = new ConcurrentHashMap<>();
    private AtomicLong id = new AtomicLong();

    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    private static class MessageWithTime {
        long time;
        Object message;
    }

    public void setSender(MessageSender sender) {
        this.sender = sender;
        startRetry();
    }

    public String generateId() {
        return "" + id.incrementAndGet();
    }

    public void add(String id, Object message) {
        map.put(id, new MessageWithTime(System.currentTimeMillis(), message));
    }

    public void del(String id) {
        map.remove(id);
    }

    private void startRetry() {
        new Thread(() ->{
            while (!stop) {
                try {
                    Thread.sleep(Constants.RETRY_TIME_INTERVAL);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                long now = System.currentTimeMillis();

                for (String key : map.keySet()) {
                    MessageWithTime messageWithTime = map.get(key);

                    if (null != messageWithTime) {
                        if (messageWithTime.getTime() + 3 * Constants.VALID_TIME < now) {
                            log.info("send message failed after 3 min " + messageWithTime);
                            del(key);
                        } else if (messageWithTime.getTime() + Constants.VALID_TIME < now) {
                            DetailRes detailRes = sender.send(messageWithTime.getMessage());

                            if (detailRes.isSuccess()) {
                                del(key);
                            }
                        }
                    }
                }
            }
        }).start();
    }
}

在client端傳送之前，先在本地快取message，程式碼如下：

@Override
public DetailRes send(Object message) {
    try {
        String id = retryCache.generateId();
        retryCache.add(id, message);
        rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
    } catch (Exception e) {
        return new DetailRes(false, "");
    }

    return new DetailRes(true, "");
}

在收到ack時刪除本地快取，程式碼如下：

rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        log.info("send message failed: " + cause + correlationData.toString());
    } else {
        retryCache.del(correlationData.getId());
    }
});

再次驗證剛才的場景，傳送300w條資料，在傳送的過程中過一段時間close一次connection，傳送結束後，實際傳送資料301.2w條，有一些重複，但是沒有丟失資料。
同時需要驗證本地快取的記憶體洩露問題，程式連續傳送1.5億條資料，記憶體佔用穩定在900M，並沒有明顯的波動。

最後貼一下rabbitmq的效能測試資料：
1 300w條1k的資料，單機部署rabbitmq(8核，32G)
在ack確認模式下平均傳送效率為1.1w條/秒
非ack確認模式下平均傳送效率為1.6w條/秒

2 300w條1k的資料，cluster模式部署3臺(8核*3， 32G*3）
在ack確認模式下平均傳送效率為1.3w條/秒
非ack確認模型下平均傳送效率為1.7w條/秒

3 300w條1k的資料，單機部署rabbitmq(8核，32G)
在ack確認模式下平均消費效率為9000條/秒

4 300w條1k的資料，cluster模式部署3臺(8核*3， 32G*3）
在ack確認模式下平均消費效率為1w條/秒

程式碼地址：