Class NotificationOutboxDispatcher

java.lang.Object

io.leandev.appfuse.notification.outbox.NotificationOutboxDispatcher

public class NotificationOutboxDispatcher
extends Object

Outbox 遞送器:fast-path 非同步遞送 + 輪詢重試/補送

兩條路徑共用同一條 dispatchOne(UUID):

• fast-path(dispatchAsync(UUID)):NotificationService 寫入 Outbox 後對每列即刻非同步觸發,把慢速 / 易失敗的實際遞送移出請求執行緒。
• safety net(dispatchPending()):由排程(Quartz / @Scheduled)定期撈取到期的 PENDING/FAILED(及逾時滯留的 SENDING)列重送,保證程序重啟、SMTP 暫斷後補送(at-least-once)。

叢集安全(競爭消費):fast-path 與任意數量的輪詢節點可能同時觸及同一列,故 dispatchOne(UUID) 先原子認領(NotificationOutboxRepositoryBase.claim(UUID, Instant, Instant))才遞送——認領把列翻為 SENDING 並推遲 nextAttemptAt,至多一個執行緒認領成功,其餘拿到 0 即略過,故不會重複遞送。認領本身不加分散式鎖, 多節點可並行 drain 佇列(不退化成單節點)。遞送節點崩潰致列滯留 SENDING 時,逾 claimTimeout 由輪詢重新認領(重用既有到期查詢)。

依 ADR-017：本類 依賴 NotificationOutboxRepositoryBase（應用帶入具體 entity 型參），對租戶的唯一接觸點是 TenantAware 探詢（決策三）——每列遞送前若該列帶非 root 的租戶,以 TenantContext.runAs(String, Runnable) 重建租戶 context(非同步 / 輪詢執行緒不帶 context),讓通道層(如 EmailNotificationChannel → 應用層 MailDelivery) 能解析當前租戶的郵件設定;結束自動清除。ownership 專案的 entity 無租戶欄位,直接遞送。 失敗依指數退避排定 nextAttemptAt,逾 maxAttempts 轉 DEAD(dead-letter)。

Since:

4.0.0

Field Summary

Fields

Modifier and Type	Field	Description
static final long	DEFAULT_CLAIM_TIMEOUT_SECONDS	認領逾時預設秒數(遞送逾此未完成即視為滯留、可被重新認領)

Constructor Summary

Constructors

Constructor	Description
NotificationOutboxDispatcher(NotificationOutboxRepositoryBase<?> repository, List<NotificationChannel> channelList, List<NotificationDeliveryListener> deliveryListeners, int batchSize, long claimTimeoutSeconds, RetryPolicyResolver retryPolicyResolver, Clock clock)
NotificationOutboxDispatcher(NotificationOutboxRepositoryBase<?> repository, List<NotificationChannel> channelList, List<NotificationDeliveryListener> deliveryListeners, int batchSize, long claimTimeoutSeconds, RetryPolicyResolver retryPolicyResolver, Clock clock, TextCipher payloadCipher)	建立具持久化 payload 解密能力的遞送器。

Method Summary

Modifier and Type	Method	Description
void	dispatchAsync(UUID outboxId)	fast-path:非同步遞送單列(寫入 Outbox 後即刻觸發)
int	dispatchPending()	safety net:撈取到期的 PENDING/FAILED 列並逐一遞送(由排程呼叫)

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

DEFAULT_CLAIM_TIMEOUT_SECONDS

public static final long DEFAULT_CLAIM_TIMEOUT_SECONDS

認領逾時預設秒數(遞送逾此未完成即視為滯留、可被重新認領)

See Also:

• Constant Field Values

Constructor Details

NotificationOutboxDispatcher

public NotificationOutboxDispatcher(NotificationOutboxRepositoryBase<?> repository,
 List<NotificationChannel> channelList,
 List<NotificationDeliveryListener> deliveryListeners,
 int batchSize,
 long claimTimeoutSeconds,
 RetryPolicyResolver retryPolicyResolver,
 Clock clock)

Parameters:

repository - Outbox repository（應用宣告的具體型別）

channelList - 已註冊的通道(依 NotificationChannel.type() 建索引)

deliveryListeners - 遞送嘗試回呼(選用稽核 hook;可為空清單,見 NotificationDeliveryListener)

batchSize - 輪詢每輪最多處理列數

claimTimeoutSeconds - 認領逾時秒數;遞送逾此未完成的滯留 SENDING 列可被輪詢重新認領。 應大於單列最慢遞送耗時(含通道逾時),避免遞送中被誤重認領而重複送出

retryPolicyResolver - 依型別解析重試退避;nextAttemptAt 取該型別的 base / max backoff (無 per-type 差異者用 RetryPolicyResolver.constant(RetryPolicy))

clock - 認領、遞送與重試時間點的應用時鐘

NotificationOutboxDispatcher

public NotificationOutboxDispatcher(NotificationOutboxRepositoryBase<?> repository,
 List<NotificationChannel> channelList,
 List<NotificationDeliveryListener> deliveryListeners,
 int batchSize,
 long claimTimeoutSeconds,
 RetryPolicyResolver retryPolicyResolver,
 Clock clock,
 TextCipher payloadCipher)

建立具持久化 payload 解密能力的遞送器。

Parameters:

payloadCipher - 與寫入端相同 purpose 的 cipher

Method Details

dispatchAsync

@Async
public void dispatchAsync(UUID outboxId)

fast-path:非同步遞送單列(寫入 Outbox 後即刻觸發)

Parameters:

outboxId - Outbox 列 ID

dispatchPending

public int dispatchPending()

safety net:撈取到期的 PENDING/FAILED 列並逐一遞送(由排程呼叫)

須在無 TenantContext 的執行緒呼叫(排程執行緒天然無 context),以跨租戶撈取。

Returns:

本輪處理的列數