無限期地等待可能永遠不會到達的消息

Question

我有一個Java類型的actor，負責可能暫時不可用的外部資源上的過濾器/重試邏輯。演員的領域，常用的方法有：

public class MyActorImpl implements MyActor { 
    private static final long MINWAIT = 50; 
    private static final long MAXWAIT = 1000; 
    private static final long DEFAULTWAIT = 0; 
    private static final double BACKOFFMULTIPLIER = 1.5; 

    private long updateWait(long currentWait) { 
     return Math.min(Math.max((long) (currentWait * BACKOFFMULTIPLIER), MINWAIT), MAXWAIT); 
    } 

    // mutable 
    private long opWait = DEFAULTWAIT; 
    private final Queue<OpInput> opBuffer = new ArrayDeque<>(); 

    // called from external actor 
    public void operation(OpInput opInput) { 
     operation(opInput, DEFAULTWAIT); 
    } 

    // called internally 
    public void operation(OpInput opInput, long currentWait); 
} 

演員有幾個操作所有或多或少相同的重試/緩衝邏輯;每個操作都有自己的[op]Wait和[op]Buffer字段。

1. 父演員調用void operation(OpInput opInput)
2. void operation(OpInput opInput, long currentWait)使用DEFAULTWAIT用於第二參數前述方法調用
3. 如果currentWait參數不等於opWait然後輸入被存儲在opBuffer，否則該輸入被髮送到外部資源。
4. 如果外部資源返回成功，則opWait設置爲DEFAULTWAIT，並且opBuffer的內容通過operation(opInput)方法發回。如果外部資源（或更可能是網絡）返回錯誤，則我使用opWait毫秒的延遲更新opWait = updateWait(opWait)和調度參與者系統調度器上的operation(opInput, opWait)。

I.e.我使用actor系統調度器來實現指數退避;我正在使用currentWait參數來標識正在重試的消息，並正在緩衝其他消息，直到主消息被外部資源成功處理。

問題是，如果預定的operation(opInput, currentWait)消息丟失，那麼我將永遠緩衝消息，因爲currentWait == opWait防護將對所有其他消息失敗。我可以使用像spring-retry這樣的實現指數回退，但我沒有看到合併操作的重試循環的方式，這意味着我可以在每個重試循環中使用一個線程（而使用actor系統的調度器不會放得太多系統上的應變）。

我正在尋找一種更容錯的方法來實現角色和外部資源之間的接口上的緩衝和指數退避，而不必爲任務分配太多資源。

Answer 1

如果我正確理解你，如果唯一的問題是失去了計劃的消息，你爲什麼不只是使用類似的Reliable Proxy Pattern針對特定信息，然後如果失敗opWait = DEFAULTWAIT;

有關於你的代碼的一些事情，我得到的，當你說public void operation(OpInput opInput)被稱爲外部時，我不明白你的意思。你的意思是說這種方法與網絡相互作用，網絡使用的資源有時不可用？

如果我可以，我建議一個替代方案。根據我的理解，你的主要問題是你有一個不可用的資源，所以你有某種你使用某種等待邏輯實現的que/buffer，這樣消息一旦被再次使用就會被處理，這很不幸涉及一些可能會丟失並導致無限等待的消息。我認爲你可以通過超時使用期貨來實現你想要的。然後重試，如果未來在一段時間內沒有完成，就說3次重複。您甚至可以根據服務器負載和完成消息所需的時間來調整此時間。希望有所幫助。