ActiveMQ高并發(fā)處理方案

1、高并發(fā)發(fā)送消息異常解決方法：現(xiàn)象：使用10個線程每100ms發(fā)送一條消息，大約3000多條后，出現(xiàn)異常，所有線程停止： javax.jms.JMSException: Could not connect to broker URL: tcp:/localhost:61616. Reaso.BindException: Address already in use: connect; nested exception is .BindException: Address already in use: connect原因：創(chuàng)建了太多jms連接沒有來得及回收解決方法：使用jms連接池原來

2、的配置：<bean><property name="environment"><props><prop key="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</prop><prop key="vider.url">tcp:/huzq-linux:61616</prop></props>&l

3、t;/property></bean><bean><property name="jndiName"><value>ConnectionFactory</value></property><property name="jndiTemplate"><ref local="jndiTemplate"></ref></property></bean>修改為：<bean><propert

4、y name="connectionFactory"><bean><property name="brokerURL" value="tcp:/huzq-linux:61616" /></bean></property></bean>解決activemq多消費者并發(fā)處理遇到一個現(xiàn)象，如果activemq隊列積壓了數(shù)據(jù)的話，如果在spring中啟動listner，只有一個consumer執(zhí)行，查閱了很多資料，無果，后來偶爾通過activemq的監(jiān)控網(wǎng)頁看到消費者列表中，只

5、有一個消費者有等待處理的數(shù)據(jù)，其他都沒有，如下圖： 由此得知，activemq有一定機制將隊列中的數(shù)據(jù)交給consumer處理，這個機制就是數(shù)據(jù)的數(shù)量分配，查資料得知，默認是1000，因此，把這個值調(diào)小就可以了。在客戶端的連接url中，修改為tcp:/ipaddr:61616?jms.prefetchPolicy.all=2這樣基本消費者就分配公平了，不會出現(xiàn)一個消費者忙死，另外的消費者閑死了。為高并發(fā)程序部署ActiveMQ使用ActiveMQ來擴展你的應(yīng)用程序需要一些時間并要花一些精力.本節(jié)中我們將介紹三種技術(shù)用于擴展應(yīng)用程序.我們將從垂直擴展開始,這種擴展方式中,單個代理需要處理成千上萬

6、的連接和消息隊列.接下來我們將介紹水平擴展,這種擴展方式需要處理比前一種方式更多的網(wǎng)絡(luò)連接.最后,我們介紹的傳輸負載分流,可以在擴展和性能間得到平衡,但是會增加ActiveMQ程序的復(fù)雜性.1. 垂直擴展:垂直擴展是一種用于增加單個ActiveMQ代理連接數(shù)(因而也增加了負載能力)的技術(shù).默認情況下,ActiveMQ的被設(shè)計成盡可高效的傳輸消息以確保低延遲和良好的性能.但是,你也可以進行一些配置使的ActiveMQ代理可以同時處理大量并發(fā)的連接以及大量的消息隊列.默認情況下,ActiveMQ使用阻塞IO來處理傳輸連接,這種方式為每一個連接分配一個線程.你可以為ActiveMQ代理使用非阻塞IO

7、(同時客戶端可以使用默認的傳輸)以減少線程的使用.可以在ActiveMQ的配置文件中通過傳輸連接器配置非阻塞IO.下面的是配置非阻塞IO的示例代碼：配置NIO傳輸連接器<broker><transportConnectors><transportConnector name="nio" uri="nio:/localhost:61616"/></<transportConnectors></broker>除了為每個連接使用一個線程的阻塞IO,ActiveMQ還可以為每一個客戶端連接使用一個

8、消息分發(fā)線程.你可以通過將系統(tǒng)參數(shù)eDedicatedTaskRunner設(shè)置為false來設(shè)置ActiveMQ使用一個搞線程池.下面是一個示例:ACTIVEMQ_OPTS="-eDedicatedTaskRunner=false"確保ActiveMQ代理用于足夠的內(nèi)存來處理大量的并發(fā)連接,需要分兩步進行:首先,你需要確保運行ActiveMQ的JVM在啟動之前已經(jīng)配置了足夠的內(nèi)存.可以使用JVM的-Xmx選項來配置,如下所示:ACTIVEMQ_OPTS="-Xmx1024eDedicatedTaskRunner=false"其次,需要確保JVM配置了適量

9、的專門供ActiveMQ代理使用的內(nèi)存.這個配置可用通過<system-Usage> 元素的limit屬性來配置.一個不錯的根據(jù)經(jīng)驗得到的規(guī)則時,在連接數(shù)為幾百個時配置512MB為最小內(nèi)存.如果測試發(fā)現(xiàn)內(nèi)存不夠用,可以增加內(nèi)存配置.你可以按照下面代碼示例來配置ActiveMQ使用的內(nèi)存限制:代碼:為ActiveMQ代理設(shè)置內(nèi)存使用限制<systemUsage><systemUsage><memoryUsage><memoryUsage limit="512 mb"/></memoryUsage>

10、; <storeUsage><storeUsage limit="10 gb" name="foo"/></storeUsage> <tempUsage><tempUsage limit="1 gb"/></tempUsage> </systemUsage></systemUsage>同樣,簡易減少每個連接的CPU負載.如果你正使用Open-Wire格式的消息,關(guān)閉tight encoding選項,開啟該選項會

11、導(dǎo)致CPU占有過多.Tight encoding選項可以通過客戶端連接的URI中的參數(shù)設(shè)置以便關(guān)閉該選項.下面是示例代碼:String uri = "failover:/(tcp:/localhost:61616?" + wireFormat.tightEncodingEnabled=false)"ConnectionFactory cf = new ActiveMQConnectionFactory(uri);了解了一些擴展ActiveMQ代理處理大量連接的調(diào)優(yōu)選項之后,我們在了解一些讓ActiveMQ處理大量消息隊列的調(diào)優(yōu)選項.默認的消息隊列配置中使用一個獨立

12、的線程負責將消息存儲中的消息提取到消息隊列中而后再被分發(fā)到對其感興趣的消息消費者.如果有大量的消息隊列,建議通過啟用optimizeDispatch這個屬性改善這個特性,示例代碼如下所示:<destinationPolicy><policyMap><policyEntries><policyEntry queue=">" optimizedDispatch="true"/></policyEntries></policyMap></destinationPolicy>

13、;注意,代碼清單中使用通配符>表示該配置會遞歸的應(yīng)用到所有的消息隊列中.為確保擴展配置既可以處理大量連接也可以處理海量消息隊列,請使用JDBC或更新更快的KahaDB消息存儲.默認情況下ActiveMQ使用KahaDB消息存儲.到目前位置,我們關(guān)注了連接數(shù)擴展,減少線程使用以及選擇正確的消息存儲.下面的示例配置代碼展示了ActiveMQ配置中為擴展進行了調(diào)優(yōu):代碼：為擴展進行調(diào)優(yōu)的配置示例代碼<broker xmlns="" brokerName="amq-broker" dataDirectory="$activemq.base/

14、data"> <persistenceAdapter><kahaDB directory="$activemq.base/data" journalMax"32mb"/></persistenceAdapter> <destinationPolicy><policyMap><policyEntries><policyEntry queue="&gt;" optimizedDispatch="true"/>&l

15、t;/policyEntries></policyMap></destinationPolicy> <systemUsage><systemUsage> <memoryUsage><memoryUsage limit="512 mb"/></memoryUsage> <storeUsage><storeUsage limit="10 gb" name="foo"/></storeUsage> <tempU

16、sage><tempUsage limit="1 gb"/></tempUsage> </systemUsage></systemUsage> <transportConnectors><transportConnector name="openwire" uri="nio:/localhost:61616"/> </transportConnectors></broker>注意示例代碼中所有為調(diào)優(yōu)而建議的配置條目,這些調(diào)優(yōu)條目在默認

17、的配置文件中并沒有配置,所以請確保給予充分重視.了解過如何擴展ActiveMQ后,現(xiàn)在是時候了解使用代理網(wǎng)絡(luò)來進行橫向擴展了.2. 橫向擴展除了擴展單獨的代理,你還可以使用代理網(wǎng)絡(luò)來增加應(yīng)用程序可用的代理數(shù)量.因為網(wǎng)絡(luò)會自動傳遞消息給所有互聯(lián)的具有對消息感興趣的消息消費者的代理,所以你可以配置客戶端連接到一個代理集群,隨機的選擇集群中的一個代理來連接.可以通過URI中的參數(shù)來配置,如下所示:failover:/(tcp:/broker1:61616,tcp:/broker2:61616)?randomize=true 為了確保隊列或持久化主題中的消息不會卡在某個代理上而不能進行轉(zhuǎn)發(fā),

18、需要在配置網(wǎng)絡(luò)連接時,將dynamicOnly配置成true并使用小一點的prefetchSize.下面是一個示例:<networkConnector uri="static:/(tcp:/remotehost:61617)" name="bridge" dynamicOnly="true" prefetchSize="1"</networkConnector> 示例代理網(wǎng)絡(luò)來橫向擴展并不會的代理更多的延遲,因為消息在傳送到消息消費者在之前會經(jīng)過多個代理.另外一種可選的部署方案可以提供更多的擴展性和更好的性能,但是需要在應(yīng)用程序中做更多的計劃.這種混合的解決方案被稱為傳輸負載分流(traffic partitioning),這種方案通過在應(yīng)用程序中分割消息目的地到不同的代理上以完成垂直擴展.3. 傳輸負載分流客戶端的傳輸負載分流是一個垂直和水平混合的負載分流方案.通常不使用代理網(wǎng)絡(luò),因為客戶端程序會決定將哪個負載發(fā)送到哪個(些)代理上.客戶端程序需要維護多個JMS連接并且決定