這篇文章主要介紹“Springboot異步消息處理的方法”的相關(guān)知識，小編通過實際案例向大家展示操作過程，操作方法簡單快捷，實用性強，希望這篇“Springboot異步消息處理的方法”文章能幫助大家解決問題。

成都創(chuàng)新互聯(lián)是一家專注于成都網(wǎng)站設(shè)計、做網(wǎng)站與策劃設(shè)計,鷹潭網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)做網(wǎng)站,專注于網(wǎng)站建設(shè)10余年,網(wǎng)設(shè)計領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:鷹潭等地區(qū)。鷹潭做網(wǎng)站價格咨詢:13518219792

在工作中經(jīng)常會碰到需要進行異步消息處理的業(yè)務(wù)場景，根據(jù)消息性質(zhì)的不同有完全不同的處理方式。

1、消息不獨立

不獨立的消息通常是有順序依賴關(guān)系，這時消息處理機制將退化為線性隊列處理模式，只能由一個消費者去單線程處理消息。

2、消息完全獨立

完全獨立的消息，可以由多個消費者（線程）并發(fā)同時處理，可以達到較大的并發(fā)處理能力。

3、消息不完全獨立

通常這種情況是，同源消息（來自同一生產(chǎn)者）要求有序，異源消息順序無關(guān)。

這個場景的消息處理會相對復(fù)雜點，為了保證同源消息有序，很容易想到對同一來源的消息綁定固定的消費者線程，這樣做很簡單但存在很大問題。

如果生產(chǎn)者數(shù)量很大，綁定線程數(shù)可能不夠，當(dāng)然可以復(fù)用線程資源，同一線程綁定多個消息來源進行處理，這樣做又會有另一個問題：消息源之間的相互影響。

考慮以下場景：

生產(chǎn)者P1產(chǎn)生大量消息進入隊列后被分配給消費線程C1處理（C1可能需要處理很長時間），這時生產(chǎn)者P2產(chǎn)生了一個消息，不幸的是也被分配給了消費線程C1處理

那么生產(chǎn)者P2的消息處理將被P1的大量消息給阻塞住，導(dǎo)致了P1和P2之間的相互影響，而且也不能充分利用其它消費線程導(dǎo)致不均衡。

所以，我們必須考慮避免這樣的問題。做到消費處理的及時性（盡快）、隔離性（避免相互干擾）、均衡性（較大化并發(fā)處理）

在實現(xiàn)中，會有兩種模式，比較容易想到的是線程派發(fā)模型（PUSH方式），具體做法通常如下：

1. 有一個全局消息派發(fā)者，輪詢隊列取出消息。

2. 根據(jù)消息來源，派發(fā)給合適的消費線程處理。

派發(fā)的算法機制簡單的可以類似像基于消息來源的Hash，復(fù)雜的可以根據(jù)各個消費線程的當(dāng)前負(fù)載，等待隊列長短、消息的復(fù)雜度進行綜合分析選擇派發(fā)。

簡單Hash肯定會碰到上述場景描述的問題，但復(fù)雜派發(fā)計算很明顯實現(xiàn)起來非常麻煩和復(fù)雜，效率也不一定好，在均衡性方面也很難做到十分平衡。

第二種模式采用PULL方式，線程按需拉取，具體做法如下：

1. 消息源直接將產(chǎn)生的消息放入對應(yīng)該源的臨時隊列中（如下所示每個session代表一個不同的消息來源），再將session置入一個阻塞隊列通知線程處理

2.  多個消費線程同時輪詢隊列，爭搶消息（保證只有一個線程取到

3. 檢查隊列指示器是否正被其他線程處理（實現(xiàn)時需要在線程級別基于同源消息的檢測同步）

4. 若未被其他線程處理，則在同步區(qū)置處理中指示狀態(tài)，退出同步區(qū)后對臨時隊列中的消息進行處理

5. 處理完成后，最后再次進入同步區(qū)置處理指示狀態(tài)為空閑

下面用一段代碼來描述下消費線程處理流程：

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);
	}
}

關(guān)于“Springboot異步消息處理的方法”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，小編每天都會為大家更新不同的知識點。

文章題目：Springboot異步消息處理的方法-創(chuàng)新互聯(lián)
本文URL：http://www.rwnh.cn/article40/dhpheo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供微信公眾號、網(wǎng)站排名、服務(wù)器托管、網(wǎng)站設(shè)計、網(wǎng)站維護、品牌網(wǎng)站制作

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)