消息中心

"Java中的線程池類其實就是一種生產者和消費者模式的實現方式，可是我以爲其實現方式更加高明。生產者把任務丟給線程池，線程池建立線程並處理任務，若是將要運行的任務數大於線程池的基本線程數就把任務扔到阻塞隊列裏，這種作法比只使用一個阻塞隊列來實現生產者和消費者模式顯然要高明不少，由於消費者可以處理直接就處理掉了，這樣速度更快，而生產者先存，消費者再取這種方式顯然慢一些。"（參考http://ifeve.com/producers-and-consumers-mode/ 清英）

4.生產者-消費者模式

經典的多線程設計模式，爲多線程間的協做提供了良好的解決方案。生產者和消費者之間經過共享內存緩存區進行通訊。共享內存緩存區的主要功能是數據在多線程間的共享，此外經過該緩存區，能夠緩解生產者和消費者間的性能差，做爲生產者和消費者間的通訊橋樑，避免了生產者和消費者的直接通訊，從而將生產者和消費者進行解耦。

具體參考：「語言-高併發-生產者消費者模式」

5.阻塞隊列

阻塞隊列的基本概念以及簡單實現 http://ifeve.com/blocking-queues/

若是隊列是空的，消費者會一直等待，當生產者添加元素時候，消費者是如何知道當前隊列有元素的呢？若是讓你來設計阻塞隊列你會如何設計，讓生產者和消費者可以高效率的進行通信呢？讓咱們先來看看JDK是如何實現的。

使用通知模式實現。所謂通知模式，就是當生產者往滿的隊列裏添加元素時會阻塞住生產者，當消費者消費了一個隊列中的元素後，會通知生產者當前隊列可用。經過查看JDK源碼發現ArrayBlockingQueue使用了Condition來實現。

好比當執行take操做時，若是隊列爲空，則讓當前線程等待在notEmpty上。當新元素進入隊列後，須要通知等待在notEmpty上的線程，讓他們繼續工做。對於put操做也是同樣的，當隊列滿時，須要讓壓入線程等待，當有元素從隊列中被挪走時，隊列中出現空位時，天然也須要通知等待入隊的線程。

http://www.infoq.com/cn/articles/java-blocking-queue/

方案二

1.如何實現線程通訊

傳統的線程通訊：

Object類提供的wait()、notify()、notify()三個方法，這三個方法必須由同步監視器對象來調用（在調用這些方法以前，線程必須得到該對象的對象級別鎖），能夠分爲如下兩種狀況：

對於使用synchronize修飾的同步方法，由於該類的默認實例this就是同步監視器，因此能夠在同步方法中直接調用這三個方法；

對於使用synchronize修飾的同步代碼塊，同步監視器是synchronize後括號裏的對象，因此必須使用該對象調用這三個方法。

wait方法可使調用該方法的線程釋放共享資源的鎖，而後從運行狀態退出，進入等待隊列，直到被再次喚醒

notify方法能夠隨機喚醒等待隊列中等待同一共享資源的一個線程，並使該線程退出等待隊列，進行可運行狀態。只有當前線程放棄對該同步監視器的鎖定後，才能夠執行被喚醒的線程。

notifyAll方法可使全部正在等待隊列中等待同一共享資源的「所有線程」從等待狀態退出，進入可運行狀態。只有當前線程放棄對該同步監視器的鎖定後，才能夠執行被喚醒的線程。

wait使線程中止運行，而notify使中止的線程繼續運行。

2.如何實現線程的掛起（suspend）和繼續執行（resume）？

JAVA多線程機制之暫停、恢復和中止

http://blog.csdn.net/jianggujin/article/details/50458486

Java 線程暫停與繼續

http://blog.csdn.net/ritterliu/article/details/7453264

3.ConcurrentHashMap以及線程安全的JDK併發容器

（1）包裝方式

基本原理：使用委託，將本身全部的map功能交給傳入的HashMap實現，而本身則主要負責保證線程安全。不管是對Map的讀取或者寫入，都須要得到mutex的鎖，這會致使全部對map的操做都進入等待狀態，直到mutex可用，若是併發級別不高，通常也夠用。可是，在高併發環境中，咱們也有必要尋求新的解決方案。

（2）一種更加專業的併發HashMap是ConcurrentHashMap

ConcurrentHashMap原理分析：https://my.oschina.net/hosee/blog/639352

不變模式：參考《Java高併發程序設計》

簡要理解以下：

Hashtable是一個線程安全的類，它使用synchronized來鎖住整張Hash表來實現線程安全，即每次鎖住整張表讓線程獨佔。

ConcurrentHashMap容許多個修改操做併發進行，其關鍵在於使用了分段鎖技術。它使用了多個鎖來控制對hash表的不一樣部分進行的修改。ConcurrentHashMap內部使用段(Segment)來表示這些不一樣的部分，每一個段其實就是一個小的Hashtable，它們有本身的鎖。只要多個修改操做發生在不一樣的段上，它們就能夠併發進行。

如何肯定元素的存取位置？

三次hash：

對於一個key，先進行一次hash操做，獲得hash值h1，也即h1 = hash1(key)；
將獲得的h1的高几位進行第二次hash，獲得hash值h2，也即h2 = hash2(h1高几位)，經過h2可以肯定該元素的放在哪一個Segment；
將獲得的h1進行第三次hash，獲得hash值h3，也即h3 = hash3(h1)，經過h3可以肯定該元素放置在哪一個HashEntry。

get操做不須要加鎖

不變(Immutable)和易變(Volatile)ConcurrentHashMap徹底容許多個讀操做併發進行，讀操做並不須要加鎖。若是使用傳統的技術，如HashMap中的實現，若是容許能夠在hash鏈的中間添加或刪除元素，讀操做不加鎖將獲得不一致的數據。ConcurrentHashMap實現技術是保證HashEntry幾乎是不可變的。HashEntry表明每一個hash鏈中的一個節點，其結構以下所示：

static final class HashEntry<K,V> { final K key; final int hash; volatile V value; volatile HashEntry<K,V> next; }

不變模式（immutable）是多線程安全裏最簡單的一種保障方式。經過volatile和final來確保數據安全。

put操做須要加鎖

size操做三次嘗試（作法比較巧妙）

size操做與put和get操做最大的區別在於，size操做須要遍歷全部的Segment才能算出整個Map的大小，而put和get都只關心一個Segment。假設咱們當前遍歷的Segment爲SA，那麼在遍歷SA過程當中其餘的Segment好比SB可能會被修改，因而這一次運算出來的size值可能並非Map當前的真正大小。因此一個比較簡單的辦法就是計算Map大小的時候全部的Segment都Lock住，不能更新(包含put，remove等等)數據，計算完以後再Unlock。這是普通人可以想到的方案，可是牛逼的做者還有一個更好的Idea：先給3次機會，不lock全部的Segment，遍歷全部Segment，累加各個Segment的大小獲得整個Map的大小，若是某相鄰的兩次計算獲取的全部Segment的更新的次數（每一個Segment都有一個modCount變量，這個變量在Segment中的Entry被修改時會加一，經過這個值能夠獲得每一個Segment的更新操做的次數）是同樣的，說明計算過程當中沒有更新操做，則直接返回這個值。若是這三次不加鎖的計算過程當中Map的更新次數有變化，則以後的計算先對全部的Segment加鎖，再遍歷全部Segment計算Map大小，最後再解鎖全部Segment。

初始化參數

initialCapacity表示新建立的這個ConcurrentHashMap的初始容量，也就是上面的結構圖中的Entry數量。默認值爲static final int DEFAULT_INITIAL_CAPACITY = 16;
loadFactor表示負載因子，就是當ConcurrentHashMap中的元素個數大於loadFactor * 最大容量時就須要rehash，擴容。默認值爲static final float DEFAULT_LOAD_FACTOR = 0.75f;
concurrencyLevel表示併發級別，這個值用來肯定Segment的個數，Segment的個數是大於等於concurrencyLevel的第一個2的n次方的數。好比，若是concurrencyLevel爲12，13，14，15，16這些數，則Segment的數目爲16(2的4次方)。默認值爲static final int DEFAULT_CONCURRENCY_LEVEL = 16;。理想狀況下ConcurrentHashMap的真正的併發訪問量可以達到concurrencyLevel，由於有concurrencyLevel個Segment，假若有concurrencyLevel個線程須要訪問Map，而且須要訪問的數據都剛好分別落在不一樣的Segment中，則這些線程可以無競爭地自由訪問（由於他們不須要競爭同一把鎖），達到同時訪問的效果。這也是爲何這個參數起名爲「併發級別」的緣由。

hash算法：

這裏用到了Wang/Jenkins hash算法的變種，主要的目的是爲了減小哈希衝突，使元素可以均勻的分佈在不一樣的Segment上，從而提升容器的存取效率。假如哈希的質量差到極點，那麼全部的元素都在一個Segment中，不只存取元素緩慢，分段鎖也會失去意義。

4.關於單元測試，mock測試

編寫更好的http://blog.csdn.net/qq_35101189/article/details/53611041

http://blog.csdn.net/chjttony/article/details/14522771

https://www.ibm.com/developerworks/cn/opensource/os-cn-easymock/

兩種方案限流及主要思路

方案一

限流在消息服務器端實現，經過輪詢，控制在INTERVAL（60000毫秒）時間內從服務器端阻塞隊列中取出最多max條消息，建立對應的消息發送器MessageSender(實現Runnable接口)而後提交給線程池（newFixedThreadPool，線程數量爲5）去執行消息的發送（模擬實現）。而消息中心（MessageCenter）主要實現對外提供消息的添加（push）、消息發送器註冊（register）、消息分發（dispatchMessage）、消息中心開啓（start）以及消息中心關閉（shutdown），其中添加即往消息中心阻塞隊列塞各種消息，註冊即往map（需考慮線程安全）中放messagetype-messageServer對，分發即把消息中心阻塞隊列中的消息向各自的服務器阻塞隊列中進行添加，開啓和關閉藉助標誌isRunning（AtomicBoolean類型）實現。各種消息服務器和消息中心均爲單例（使用雙重鎖機制實現，需注意加volatile修飾保證線程安全的延遲初始化）。

方案二

限流在消息中心實現，在消息中心中經過建立專門的發送線程sendThread從消息中心阻塞隊列中取出消息，而後進行發送操做，限流徹底藉助sendThread的sleep（sendSpan）（sendSpan爲消息發送時間間隔，如每分鐘發送的消息數量爲count，則兩者的關係爲：sendSpan = 60 * 1000 / count）來實現。消息中心對外提供：setSendExecutors（從新設置消息發送線程池，用於優化發送速度）、setMsgSendCountPerMinute（設置每分鐘的消息發送數量，從而改變默認的sendSpan值用於不一樣類型消息的限流）、registerOrReplaceSender(在ConcurrentHashMap中註冊或者更換消息發送器)、addMsg(向消息中心傳遞消息，採用非阻塞方式，當消息池滿後拋出異常)、start（啓動/重啓消息發送服務，經過對sendThread的開啓和恢復來實現）、suspend（暫停發送消息，經過對sendThread的掛起實現）、sendMsg（私有方法，根據消息獲取對應的消息發送器，建立一個Runnable對象而後提交給線程池執行）。這裏面sendThread的掛起（suspend）和恢復（resume）需藉助傳統的線程通訊wait和notifyAll來實現可靠的線程掛起和恢復。

相關標籤/搜索