Springboot非同步訊息處理的方法

在工作中經常會碰到需要進行非同步訊息處理的業務場景，根據訊息性質的不同有完全不同的處理方式。

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中文網其他相關文章！