在工作中經常會碰到需要進行非同步訊息處理的業務場景，根據訊息性質的不同有完全不同的處理方式。
1、訊息不獨立
不獨立的訊息通常是有順序依賴關係，這時訊息處理機制將退化為線性佇列處理模式，只能由一個消費者去單線程處理訊息。
2、訊息完全獨立
完全獨立的訊息，可以由多個消費者（執行緒）並發同時處理，可以達到最大的並發處理能力。
3、訊息不完全獨立
通常這種情況是，同源訊息（來自同一生產者）要求有序，異源訊息順序無關。
這個場景的訊息處理會相對複雜點，為了保證同源訊息有序，很容易想到對同一來源的訊息綁定固定的消費者線程，這樣做很簡單但存在很大問題。
如果生產者數量很大，綁定線程數可能不夠，當然可以復用線程資源，同一線程綁定多個消息來源進行處理，這樣做又會有另一個問題：消息源之間的相互影響。
考慮以下場景：
生產者P1產生大量訊息進入佇列後被分配給消費性執行緒C1處理（C1可能需要處理很長時間），這時生產者P2產生了一個訊息，不幸的是也被分配給了消費線程C1處理
那麼生產者P2的訊息處理將被P1的大量訊息給阻塞住，導致了P1和P2之間的相互影響，而且也不能充分利用其它消費線程導致不均衡。
所以，我們必須考慮避免這樣的問題。做到消費處理的及時性（盡快）、隔離性（避免相互幹擾）、均衡性（最大化並發處理）
在實作中，會有兩種模式，比較容易想到的是線程派發模型（PUSH方式），具體做法通常如下：
1. 有一個全域訊息派遣者，輪詢佇列取出訊息。
2. 根據消息來源，派發給適當的消費執行緒處理。
派發的演算法機制簡單的可以類似像基於消息來源的Hash，複雜的可以根據各個消費線程的當前負載，等待隊列長短、消息的複雜度進行綜合分析選擇派發。
簡單Hash肯定會碰到上述場景描述的問題，但複雜派發計算很明顯實現起來非常麻煩和復雜，效率也不一定好，在均衡性方面也很難做到十分平衡。
第二種模式採用PULL方式，執行緒按需拉取，具體做法如下：
1. 訊息來源直接將產生的訊息放入對應該來源的臨時佇列中（如下所示每個session代表一個不同的消息來源），再將session置入一個阻塞隊列通知線程處理
2. 多個消費線程同時輪詢隊列，爭搶訊息（保證只有一個線程取到
3. 檢查佇列指示器是否正被其他執行緒處理（實作時需要在執行緒層級基於同源訊息的偵測同步）
4. 若未被其他執行緒處理，則在同步區置處理中指示狀態，退出同步區後對臨時佇列中的訊息進行處理
5. 處理完成後，最後再次進入同步區置處理指示狀態為空閒
以下用一段程式碼來描述下消費性執行緒處理流程：
public void run() {
try {
for (AbstractSession s = squeue.take(); s != null; s = squeue.take()) {
// first check any worker is processing this session?
// if any other worker thread is processing this event with same session, just ignore it.
synchronized (s) {
if (!s.isEventProcessing()) {
s.setEventProcessing(true);
} else {
continue;
}
}
// fire events with same session
fire(s);
// last reset processing flag and quit current thread processing
s.setEventProcessing(false);
// if remaining events, so re-insert to session queue
if (s.getEventQueue().size() > 0 && !s.isEventProcessing()) {
squeue.offer(s);
}
}
} catch (InterruptedException e) {
LOG.warn(e.getMessage(), e);
}
}以上是Springboot非同步訊息處理的方法的詳細內容。更多資訊請關注PHP中文網其他相關文章！
