引言

生產(chǎn)者不必是一個(gè)始終在執(zhí)行的線程，它可以是一個(gè)接口，接受客戶端的請(qǐng)求，向隊(duì)列中插入消息；消費(fèi)者也不必是一個(gè)始終在執(zhí)行的線程，它同樣也可以是一個(gè)接口，接受客戶端的請(qǐng)求，從隊(duì)列中取出屬于自己的消息；看到很多介紹生產(chǎn)者消息者實(shí)現(xiàn)的文章，實(shí)現(xiàn)場(chǎng)景都很簡(jiǎn)單，現(xiàn)實(shí)應(yīng)用往往會(huì)比較復(fù)雜，有一些附加條件，本例中就需要根據(jù)消息中的 familyId 來(lái)判斷消息是不是下發(fā)給自己的。

應(yīng)用場(chǎng)景

本例的應(yīng)用場(chǎng)景是一個(gè)物聯(lián)網(wǎng)智能家居應(yīng)用，系統(tǒng)結(jié)構(gòu)圖如下：

主要實(shí)現(xiàn)的功能是用戶通過(guò)手機(jī)端APP發(fā)出設(shè)備控制的命令（如：開(kāi)燈、關(guān)燈等）后，設(shè)備網(wǎng)關(guān)能夠?qū)崟r(shí)的獲取到控制命令，進(jìn)而控制設(shè)置的狀態(tài)。

為什么要使用長(zhǎng)輪詢功能呢？

其實(shí)可選的方案有很多種：

1、長(zhǎng)輪詢

2、長(zhǎng)鏈接

3、Socket

4、WebSocket

5、MQTT

選擇長(zhǎng)輪詢方案是因?yàn)槠鋵?shí)現(xiàn)的簡(jiǎn)單性，實(shí)現(xiàn)起來(lái)與其它的接口基本沒(méi)有太大的差別。

同步與異步處理模式分析

同步服務(wù)模式：

同步服務(wù)為每個(gè)請(qǐng)求創(chuàng)建單一線程，由此線程完成整個(gè)請(qǐng)求的處理：接收消息，處理消息，返回?cái)?shù)據(jù)；這種情況下服務(wù)器資源對(duì)所有入棧請(qǐng)求開(kāi)放，服務(wù)器資源被所有入棧請(qǐng)求競(jìng)爭(zhēng)使用，如果入棧請(qǐng)求過(guò)多就會(huì)導(dǎo)致服務(wù)器資源耗盡宕機(jī)，或者導(dǎo)致競(jìng)爭(zhēng)加劇，資源調(diào)度頻繁，服務(wù)器資源利用效率降低。

異步服務(wù)則可以分別設(shè)置兩個(gè)線程隊(duì)列，一個(gè)專門負(fù)責(zé)接收消息，另一個(gè)專門負(fù)責(zé)處理消息并返回?cái)?shù)據(jù)，另有一些值守線程負(fù)責(zé)任務(wù)派發(fā)和超時(shí)監(jiān)控等工作。在這種情況下無(wú)論入棧請(qǐng)求有多少，服務(wù)器始終依照自己的能力處理請(qǐng)求，服務(wù)器資源消耗始終在一個(gè)可控的范圍。這種模式的一個(gè)問(wèn)題就是這兩個(gè)線程隊(duì)列的大小如何根據(jù)機(jī)器負(fù)載情況動(dòng)態(tài)調(diào)整。

異步服務(wù)模式：

這種情況下，雖然入棧請(qǐng)求以消息隊(duì)列的方式被異步處理但每個(gè)請(qǐng)求內(nèi)部卻是采用阻塞的方式訪問(wèn)外部資源，如果外部資源訪問(wèn)速度過(guò)慢，可能導(dǎo)致請(qǐng)求處理隊(duì)列中的所有線程均處于阻塞狀態(tài)，此時(shí)CPU使用率雖然很低但是卻因?yàn)殛?duì)列中線程已滿而無(wú)法處理消息隊(duì)列中的新消息，此時(shí)若能調(diào)整線程隊(duì)列最大線程數(shù)將可提高CPU利用率。但另一個(gè)問(wèn)題是如果線程數(shù)被調(diào)高之后所有線程的IO處理突然結(jié)束并且接下來(lái)每個(gè)線程都將進(jìn)行大量計(jì)算的話那么CPU可能出現(xiàn)過(guò)載。

在系統(tǒng)運(yùn)行的每個(gè)時(shí)間點(diǎn)上，當(dāng)時(shí)正在進(jìn)行IO的線程數(shù)量和正在進(jìn)行計(jì)算的線程數(shù)量是不斷變化著的，那么如何才能設(shè)計(jì)出一個(gè)可以根據(jù)系統(tǒng)當(dāng)時(shí)情況自動(dòng)適應(yīng)負(fù)載變化的高度自適應(yīng)的系統(tǒng)呢？

在這方面采用反應(yīng)式計(jì)算模型確實(shí)能設(shè)計(jì)出適應(yīng)負(fù)載能力很強(qiáng)的系統(tǒng)，系統(tǒng)利用率和吞吐量可以大幅提高，但這種系統(tǒng)仍然可能會(huì)出現(xiàn)系統(tǒng)局部負(fù)載過(guò)高的風(fēng)險(xiǎn)。采用反應(yīng)式計(jì)算模型，不僅系統(tǒng)中的入棧請(qǐng)求以消息隊(duì)列的方式得以異步化，而且系統(tǒng)中所有的IO任務(wù)也必需依照此法行之，這些IO任務(wù)的處理需要采用異步模型（如NIO）。另外要考慮的就是如何劃分異步IO消息并為其配置線程隊(duì)列了，比如是要將所有IO任務(wù)放入統(tǒng)一的隊(duì)列還是為某類IO任務(wù)設(shè)置單獨(dú)的隊(duì)列。

服務(wù)器資源雖然由系統(tǒng)分配但大多以線程為持有者被線程持有并使用，如線程堆棧，被線程持有的各類鎖等資源。

實(shí)現(xiàn)步驟

1、定義消息隊(duì)列

我這里是定義的靜態(tài)常量，你找個(gè)類把它放進(jìn)去就可以了。

/**
	 * 存儲(chǔ)客戶端（用戶）提交的設(shè)置控制命令
	 */
	public final static BlockingQueue<Equipment> EQUIPMENT_CONTROL = new LinkedBlockingQueue<Equipment>();

Equipment 是一個(gè)消息實(shí)體類，在本例中它最關(guān)鍵的屬性是 familyId，因?yàn)橐鶕?jù)它來(lái)判斷消息是下發(fā)給哪個(gè)家庭的，你發(fā)了一個(gè)關(guān)燈的命令結(jié)果我家的燈滅了這肯定是不行的。

2、實(shí)現(xiàn)生產(chǎn)者

生產(chǎn)者不必是一個(gè)始終在執(zhí)行的線程，它可以是一個(gè)接口，接受客戶端的請(qǐng)求；

/**
	 * 保存或更新設(shè)備接口
	 **/
	@RequestMapping("/save.do")
	@ResponseBody
	public void save(HttpServletRequest request, HttpServletResponse response) throws Exception {
		Personal personal = SecurityUtils.getPersonal(request);
		Long personalId = personal.getId();
		if (personalId == null) {
			outFailureJson(response, BaseCodeMessage.personal_10001);
			return;
		}
		//做你要做的事情
		//向隊(duì)列中插入消息
		ConstantDict.EQUIPMENT_CONTROL.put(entity);
		//輸出響應(yīng)內(nèi)容
		this.outResultJson(response, "success", "Equipment", entity);
	}

這里你要做的最關(guān)鍵的是：向隊(duì)列中插入消息。

Personal 是一個(gè)用戶信息實(shí)體類，通過(guò) SecurityUtils.getPersonal(request); 方法根據(jù) Session 或 Cookie 來(lái)從緩存或數(shù)據(jù)庫(kù)中獲取當(dāng)前登錄用戶信息。

3、實(shí)現(xiàn)消費(fèi)者

此接口由網(wǎng)關(guān)調(diào)用。

這里要做的循環(huán)從隊(duì)列中取數(shù)據(jù)，然后根據(jù) familyId 判斷消息是不是屬于自己的，是就退出循環(huán)，不是就把剛剛?cè)〕龅南⒃俜呕厝ァ?/p>

/**
	 * 控制設(shè)置狀態(tài)接口_供網(wǎng)關(guān)調(diào)用 這是一個(gè)提供長(zhǎng)輪詢的方法，網(wǎng)關(guān)通過(guò)長(zhǎng)輪詢來(lái)即時(shí)獲得命令信息
	 * 
	 * @author lipw
	 * @date 2017年8月30日下午3:31:59
	 * @param request
	 * @param response
	 * @throws Exception
	 */
	@RequestMapping("/ctrlgw.do")
	@ResponseBody
	public void controlgw(HttpServletRequest request, HttpServletResponse response) throws Exception {
		Personal personal = SecurityUtils.getPersonal(request);
		Long personalId = personal.getId();
		if (personalId == null) {
			outFailureJson(response, BaseCodeMessage.personal_10001);
			return;
		}
		Long familyId = personal.getFamilyId();
		if (familyId == null) {
			outFailureJson(response, "2", "尚未分配家庭編號(hào)！");
			return;
		}
		Equipment equipment = null;
		while (true) {
			equipment = ConstantDict.EQUIPMENT_CONTROL.poll(5000, TimeUnit.MILLISECONDS);
			if (equipment != null) {
				if (familyId.equals(equipment.getFamilyId())) {
					System.out.println("從隊(duì)列取走一個(gè)元素，隊(duì)列剩余" + ConstantDict.EQUIPMENT_CONTROL.size() + "個(gè)元素");
					break;
				} else {
					// 不屬于自己，再放回隊(duì)列
					ConstantDict.EQUIPMENT_CONTROL.put(equipment);
				}
			}
			Thread.sleep(100);
		}
		this.outResultJson(response, "success", "equipment", equipment);
	}

為什么要這樣設(shè)計(jì)呢？

因?yàn)槿绻皇峭ㄟ^(guò) peek 方法來(lái)獲取，而不從 隊(duì)列 中移除，如果隊(duì)列頭部的消息不是屬于自己的，那就要一直循環(huán)下去卻得不到屬于自己的那一條消息。

使用 AJAX 模擬網(wǎng)關(guān)進(jìn)行測(cè)試

<div id="divCommand" style="width:98%; min-height:100px; border:1px solid #888;"></div>
<script src="${ctx}/static/script/jquery-1.10.2.min.js"></script>
<script type="text/javascript">
var successCount = 0;
function loadCommand(){
	$.ajax({  
        url:"${ctx}/xxx/ctrlgw.do?token=xxx&t=" + Date.now(),  
        type:"get",  
        data:{},
        dataType:"json",
        success:function(data)  
        {  
            if(data != null && data!=""){
            	successCount++;
            	$("#divCommand").append(successCount + ", ");
            }
            loadCommand(); //成功后繼續(xù)回調(diào)
        },error:function(data){
        	if(data != null && data!=""){
            	    $("#divCommand").append(data.statusText);
                }
        	if (data.statusText == "timeout"){
        	    loadCommand(); //超時(shí)回調(diào)
        	}
        }
    });
}
$(document).ready(function(){
    loadCommand();
});
</script>

測(cè)試方法通過(guò)一個(gè)回調(diào)函數(shù)，不斷的向服務(wù)器發(fā)出請(qǐng)求； 如果服務(wù)器隊(duì)列中有屬于自己的消息，會(huì)立即返回，沒(méi)有就會(huì)一直等待真到超時(shí)，然后重新發(fā)起請(qǐng)求。

請(qǐng)求成功后會(huì)在 Div 中顯示成功的次數(shù)，失敗了也會(huì)顯示失敗的狀態(tài)文本。

可以通過(guò)瀏覽器的開(kāi)發(fā)者工具中的 Network 來(lái)查看每次請(qǐng)求所用的時(shí)間：

結(jié)束語(yǔ)

本例的實(shí)現(xiàn)方式是同步的，隊(duì)列沒(méi)有設(shè)置大小，生產(chǎn)者被阻塞的可能性很小，除非所有網(wǎng)關(guān)都與平臺(tái)斷開(kāi)了連接不再處理消息；但消費(fèi)者的實(shí)現(xiàn)由于是同步的，會(huì)對(duì)服務(wù)器的性能有所影響，因?yàn)槊總€(gè)消費(fèi)者請(qǐng)求會(huì)占用一個(gè) Servlet 線程導(dǎo)致無(wú)法再去處理其它用戶請(qǐng)求。那么這個(gè)問(wèn)題有沒(méi)有解決方案呢？當(dāng)然有！那就是采用異步處理模式 DeferredResult 。

到此這篇關(guān)于SpringMVC基于阻塞隊(duì)列LinkedBlockingQueue的同步長(zhǎng)輪詢功能實(shí)現(xiàn)詳解的文章就介紹到這了,更多相關(guān)SpringMVC的LinkedBlockingQueue輪詢內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論