摘要：中間件的分類(lèi)基于遠(yuǎn)程過(guò)程調(diào)用的中間件?；趯?duì)象請(qǐng)求代理的中間件。消息傳遞指的是程序之間通過(guò)在消息中發(fā)送數(shù)據(jù)進(jìn)行通信，而不是通過(guò)直接調(diào)用彼此來(lái)通信，直接調(diào)用通常是用于諸如遠(yuǎn)程過(guò)程調(diào)用的技術(shù)。

1.1 什么是中間件?

由于業(yè)務(wù)、機(jī)構(gòu)和技術(shù)是不斷變化的，因此為其服務(wù)的軟件系統(tǒng)必須適應(yīng)這樣的變化。在合并、添加服務(wù)或擴(kuò)展可用服務(wù)之后，公司可能無(wú)力負(fù)擔(dān)重新創(chuàng)建信息系統(tǒng)所需的成本。正是在這個(gè)關(guān)鍵時(shí)刻，才需要集成新組件或者盡可能高效地?cái)U(kuò)展現(xiàn)有組件。要集成異類(lèi)組件，最方便的方法不是將它們重新創(chuàng)建為同類(lèi)元素，而是提供一個(gè)允許它們進(jìn)行通信（不考慮它們之間的差異）的層。該層被稱(chēng)作中間件。

1.2 中間件的分類(lèi)

基于遠(yuǎn)程過(guò)程調(diào)用 (Remote Procedure Call, RPC)的中間件。

基于對(duì)象請(qǐng)求代理 (Object Request Broker, ORB) 的中間件。

面向消息的中間件或基于 MOM 的中間件。

消息隊(duì)列中間件是分布式系統(tǒng)中重要的組件，主要解決應(yīng)用耦合、異步消息、流量削鋒等問(wèn)題。實(shí)現(xiàn)高性能、高可用、可伸縮和最終一致性架構(gòu)。是大型分布式系統(tǒng)不可缺少的中間件。

3.1 什么是jms?

JMS即Java消息服務(wù)（Java Message Service）應(yīng)用程序接口，是一個(gè)Java平臺(tái)中關(guān)于面向消息中間件（MOM）的API，用于在兩個(gè)應(yīng)用程序之間，或分布式系統(tǒng)中發(fā)送消息，進(jìn)行異步通信。Java消息服務(wù)是一個(gè)與具體平臺(tái)無(wú)關(guān)的API，絕大多數(shù)MOM提供商都對(duì)JMS提供支持。

3.2 JMS 消息傳送模式

客戶(hù)端 A、C 和 D之間的消息傳送說(shuō)明了點(diǎn)對(duì)點(diǎn)模式(P2P)?？蛻?hù)端使用此模式向隊(duì)列目的地發(fā)送一條消息，只有一個(gè)接收者能夠從該目的地獲得該消息。訪問(wèn)該目的地的其他任何接收者都不能獲得該消息。

客戶(hù)端 B、E 和 F之間的消息傳送說(shuō)明了發(fā)布/訂閱模式(publish-subscribe)?？蛻?hù)端使用此廣播模式向主題目的地發(fā)送一條消息，任意數(shù)量的使用方訂戶(hù)都可以從該目的地檢索此消息。每個(gè)訂戶(hù)都獲得此消息的一個(gè)副本。

3.3 JMS 消息傳送對(duì)象

JMS 消息傳送的對(duì)象在編程域中基本保持不變：連接工廠、連接、會(huì)話(huà)、生成方、使用方、消息和目的地。

MQ全稱(chēng)為Message Queue,消息隊(duì)列(MQ)是正確而又完整的 JMS 實(shí)現(xiàn),消息隊(duì)列（MQ）是一種應(yīng)用程序?qū)?yīng)用程序的通信方法。應(yīng)用程序通過(guò)寫(xiě)和檢索出入列隊(duì)的針對(duì)應(yīng)用程序的數(shù)據(jù)（消息）來(lái)通信，而無(wú)需專(zhuān)用連接來(lái)鏈接它們。消息傳遞指的是程序之間通過(guò)在消息中發(fā)送數(shù)據(jù)進(jìn)行通信，而不是通過(guò)直接調(diào)用彼此來(lái)通信，直接調(diào)用通常是用于諸如遠(yuǎn)程過(guò)程調(diào)用的技術(shù)。

   場(chǎng)景說(shuō)明:新用戶(hù)注冊(cè)發(fā)放100積分,180元新手大禮包,激活會(huì)員卡，傳統(tǒng)的做法有兩種:串行方式,并行方式。    

串行方式

使用消息隊(duì)列

    以上兩種方式，很容易發(fā)現(xiàn)同步處理的情況下都會(huì)涉及到非主業(yè)務(wù)的其他操作，其實(shí)注冊(cè)的的主流程不應(yīng)該受其他事件影響,通過(guò)消息隊(duì)列的方式,可以把后續(xù)的處理流程進(jìn)行異步處理可以大大提高響應(yīng)速度。

場(chǎng)景說(shuō)明:企業(yè)中經(jīng)常出現(xiàn)企業(yè)合作如:本公司的驢粉卡與電信合作，新開(kāi)卡的用戶(hù)從電信端推送到我方,除了相對(duì)應(yīng)的福利外，首先判斷是否注冊(cè)本公司賬戶(hù)，
沒(méi)有給予注冊(cè)，但是新用戶(hù)的相對(duì)應(yīng)權(quán)益需要對(duì)等的發(fā)放。

傳統(tǒng)方式

缺點(diǎn):

1.與其他系統(tǒng)過(guò)度耦合
2.短信發(fā)放或優(yōu)惠券發(fā)放失敗，影響主業(yè)務(wù)

使用消息隊(duì)列

優(yōu)點(diǎn):

1.注冊(cè)完成然后將消息寫(xiě)入隊(duì)列返回成功。
2.發(fā)放權(quán)益業(yè)務(wù)不影響主業(yè)務(wù)，實(shí)現(xiàn)解耦。

場(chǎng)景說(shuō)明:秒殺活動(dòng)對(duì)稀缺或者特價(jià)的商品進(jìn)行定時(shí)定量售賣(mài)，吸引成大量的消費(fèi)者進(jìn)行搶購(gòu)，但又只有少部分消費(fèi)者可以下單成功。
因此，秒殺活動(dòng)將在較短時(shí)間內(nèi)產(chǎn)生比平時(shí)大數(shù)十倍，上百倍的頁(yè)面訪問(wèn)流量和下單請(qǐng)求流量。

秒殺前：用戶(hù)不斷刷新商品詳情頁(yè)，頁(yè)面請(qǐng)求達(dá)到瞬時(shí)峰值。

秒殺開(kāi)始：用戶(hù)點(diǎn)擊秒殺按鈕，下單請(qǐng)求達(dá)到瞬時(shí)峰值。

秒殺后：一部分成功下單的用戶(hù)不斷刷新訂單或者產(chǎn)生退單操作，大部分用戶(hù)繼續(xù)刷新商品詳情頁(yè)等待退單機(jī)會(huì)。

秒殺前，用戶(hù)不斷刷新商品詳情頁(yè)，造成大量的頁(yè)面請(qǐng)求。所以，我們需要把秒殺商品詳情頁(yè)與普通的商品詳情頁(yè)分開(kāi)。對(duì)于秒殺商品詳情頁(yè)盡量將能靜態(tài)化的元素靜態(tài)化處理，除了秒殺按鈕需要服務(wù)端進(jìn)行動(dòng)態(tài)判斷，其他的靜態(tài)數(shù)據(jù)可以緩存在瀏覽器和CDN 上。這樣，秒殺前刷新頁(yè)面導(dǎo)致的流量進(jìn)入服務(wù)端的流量只有很小的一部分。

利用讀寫(xiě)分離 Redis 緩存攔截流量(活動(dòng)未開(kāi)始時(shí)攔截大部分動(dòng)態(tài)數(shù)據(jù)請(qǐng)求)

成功參與下單后，進(jìn)入下層服務(wù)，開(kāi)始進(jìn)行訂單信息校驗(yàn)，庫(kù)存扣量。為了避免直接訪問(wèn)數(shù)據(jù)庫(kù)，我們使用主從版 Redis 來(lái)進(jìn)行庫(kù)存扣量

如果還有大量并發(fā)的請(qǐng)求則利用消息隊(duì)列組件，當(dāng)秒殺服務(wù)將訂單信息寫(xiě)入消息隊(duì)列后，即可認(rèn)為下單完成，避免直接操作數(shù)據(jù)庫(kù)。

ActiveMQ是Apache軟件基金下的一個(gè)開(kāi)源軟件，它遵循JMS1.1規(guī)范（Java Message Service），是消息驅(qū)動(dòng)中間件軟件（MOM）。它為企業(yè)消息傳遞提供高可用，出色性能，可擴(kuò)展，穩(wěn)定和安全保障。

P2P （點(diǎn)對(duì)點(diǎn)）消息域使用 queue 作為 Destination，消息可以被同步或異步的發(fā)送和接收，每個(gè)消息只會(huì)給一個(gè) Consumer 傳送一次。
Pub/Sub（發(fā)布/訂閱，Publish/Subscribe）消息域使用 topic 作為 Destination，發(fā)布者向 topic 發(fā)送消息，訂閱者注冊(cè)接收來(lái)自 topic 的消息。發(fā)送到 topic 的任何消息都將自動(dòng)傳遞給所有訂閱者。接收方式（同步和異步）與 P2P 域相同。

消息生產(chǎn)者

       //創(chuàng)建session會(huì)話(huà)
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.187.13:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

        //創(chuàng)建一個(gè)消息隊(duì)列 session.createQueue("jms.test.topic")--P2P模式
        Destination destination = session.createTopic("jms.test.topic");

        //創(chuàng)建消息生產(chǎn)者
        MessageProducer producer = session.createProducer(destination);

        //消息持久化
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);

        for (int i = 0; i < messageNum; i++) {
            producer.send(session.createTextMessage("Message Producer:" + i));
        }

        //提交會(huì)話(huà)
        session.commit();

消息消費(fèi)者

       //創(chuàng)建session會(huì)話(huà)
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.187.13:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

        //創(chuàng)建一個(gè)消息隊(duì)列 session.createQueue("jms.test.topic")--P2P模式
        Destination destination = session.createTopic("jms.test.topic");

        //創(chuàng)建消息消費(fèi)者
        MessageConsumer consumer = session.createConsumer(destination);

        while (true) {
            TextMessage message = (TextMessage) consumer.receive();
            if (message != null){
                System.out.println("Message Consumer:"+message.getText());
            }else {
                break;
            }
        }
        session.commit();

KahaDB

ActiveMQ 5.3 版本起的默認(rèn)存儲(chǔ)方式。KahaDB存儲(chǔ)是一個(gè)基于文件的快速存儲(chǔ)消息，設(shè)計(jì)目標(biāo)是易于使用且盡可能快。它使用基于文件的消息數(shù)據(jù)庫(kù)意味著沒(méi)有第三方數(shù)據(jù)庫(kù)的先決條件。

AMQ

MQ存儲(chǔ)使用戶(hù)能夠快速啟動(dòng)和運(yùn)行，因?yàn)樗灰蕾?lài)于第三方數(shù)據(jù)庫(kù)。AMQ 消息存儲(chǔ)庫(kù)是可靠持久性和高性能索引的事務(wù)日志組合，當(dāng)消息吞吐量是應(yīng)用程序的主要需求時(shí)，該存儲(chǔ)是最佳選擇。但因?yàn)樗鼮槊總€(gè)索引使用兩個(gè)分開(kāi)的文件，并且每個(gè) Destination 都有一個(gè)索引，所以當(dāng)你打算在代理中使用數(shù)千個(gè)隊(duì)列的時(shí)候，不應(yīng)該使用它。

JDBC

選擇關(guān)系型數(shù)據(jù)庫(kù)，通常的原因是企業(yè)已經(jīng)具備了管理關(guān)系型數(shù)據(jù)的專(zhuān)長(zhǎng)，但是它在性能上絕對(duì)不優(yōu)于上述消息存儲(chǔ)實(shí)現(xiàn)。事實(shí)是，許多企業(yè)使用關(guān)系數(shù)據(jù)庫(kù)作為存儲(chǔ)，是因?yàn)樗麄兏敢獬浞掷眠@些數(shù)據(jù)庫(kù)資源。

內(nèi)存存儲(chǔ)

內(nèi)存消息存儲(chǔ)器將所有持久消息保存在內(nèi)存中。在僅存儲(chǔ)有限數(shù)量 Message 的情況下，內(nèi)存消息存儲(chǔ)會(huì)很有用，因?yàn)?Message 通常會(huì)被快速消耗。在 activema.xml 中將 broker 元素上的 persistent 屬性設(shè)置為 false 即可。

    -rw-rw-r--. 1 lvmama01 lvmama01 32M 5月  18 09:47 db-1.log
    -rw-rw-r--. 1 lvmama01 lvmama01 32K 5月  18 09:47 db.data
    -rw-rw-r--. 1 lvmama01 lvmama01 33K 5月  18 09:47 db.redo
    -rw-rw-r--. 1 lvmama01 lvmama01 0   5月  16 19:31 lock

可以看出，上面directory一共有四個(gè)文件：

①db.data

它是消息的索引文件。本質(zhì)上是B-Tree的實(shí)現(xiàn)，使用B-Tree作為索引指向db-*.log里面存儲(chǔ)的消息。

②db.redo

主要用來(lái)進(jìn)行消息恢復(fù)。

③db-*.log 存儲(chǔ)消息的內(nèi)容。對(duì)于一個(gè)消息而言，不僅僅有消息本身的數(shù)據(jù)(message data)，而且還有(Destinations、訂閱關(guān)系、事務(wù)...),data log以日志形式存儲(chǔ)消息，而且新的數(shù)據(jù)總是以APPEND的方式追加到日志文件末尾。因此，消息的存儲(chǔ)是很快的。比如，對(duì)于持久化消息，Producer把消息發(fā)送給Broker，Broker先把消息存儲(chǔ)到磁盤(pán)中(enableJournalDiskSyncs配置選項(xiàng))，然后再向Producer返回Acknowledge。Append方式在一定程度上減少了Broker向Producer返回Acknowledge的時(shí)間。

④lock文件

KahaDB內(nèi)部分為：data logs, 按照Message ID高度優(yōu)化的索引，memory message cache。

①在內(nèi)存(cache)中的那部分B-Tree是Metadata Cache

通過(guò)將索引緩存到內(nèi)存中，可以加快查詢(xún)的速度(quick retrival of message data)。但是需要定時(shí)將 Metadata Cache 與 Metadata Store同步。這個(gè)同步過(guò)程就稱(chēng)為：check point。由checkpointInterval選項(xiàng) 決定每隔多久時(shí)間進(jìn)行一次checkpoint操作。

②BTree Indexes則是保存在磁盤(pán)上的，稱(chēng)為Metadata Store，它對(duì)應(yīng)于文件db.data，它就是對(duì)Data Logs以B樹(shù)的形式索引。

有了它，Broker（消息服務(wù)器）可以快速地重啟恢復(fù)，因?yàn)樗窍⒌乃饕鶕?jù)它就能恢復(fù)出每條消息的location。如果Metadata Store被損壞，則只能掃描整個(gè)Data Logs來(lái)重建B樹(shù)了。

③Data Logs則對(duì)應(yīng)于文件 db-*.log，默認(rèn)是32MB

Data Logs以日志形式存儲(chǔ)消息，它是生產(chǎn)者生產(chǎn)的數(shù)據(jù)的真正載體。

④Redo Log則對(duì)應(yīng)于文件 db.redo,redo log的原理用到了“Double Write”。

簡(jiǎn)要記錄下自己的理解：因?yàn)榇疟P(pán)的頁(yè)大小與操作系統(tǒng)的頁(yè)大小不一樣，磁盤(pán)的頁(yè)大小一般是16KB，而OS的頁(yè)大小是4KB。而數(shù)據(jù)寫(xiě)入磁盤(pán)是以磁盤(pán)頁(yè)大小為單位進(jìn)行的，即一次寫(xiě)一個(gè)磁盤(pán)頁(yè)大小，這就需要4個(gè)OS的頁(yè)大?。?*4=16）。如果在寫(xiě)入過(guò)程中出現(xiàn)故障(突然斷電)就會(huì)導(dǎo)致只寫(xiě)入了一部分?jǐn)?shù)據(jù)(partial page write)
而采用了“Double Write”之后，將數(shù)據(jù)寫(xiě)入磁盤(pán)時(shí)，先寫(xiě)到一個(gè)Recovery Buffer中，然后再寫(xiě)到真正的目的文件中。在ActiveMQ的源碼PageFile.java中有相應(yīng)的實(shí)現(xiàn)。

public void unload() throws IOException {
     //load時(shí)創(chuàng)建writeFile(db.data)和 recoveryFile(db.redo)
    writeFile = new RecoverableRandomAccessFile(file, "rw", false);

    ........

    if (enableRecoveryFile) {
         recoveryFile = new RecoverableRandomAccessFile(getRecoveryFile(), "rw");
    }
}
private void writeBatch() throws IOException {
    .......
    //將數(shù)據(jù)寫(xiě)入磁盤(pán)時(shí)，先寫(xiě)到一個(gè)Recovery Buffer中(db.data)
    for (PageWrite w : batch) {
        try {
            checksum.update(w.getDiskBound(), 0, pageSize);
        } catch (Throwable t) {
            throw IOExceptionSupport.create("Cannot create recovery file. Reason: " + t, t);
        }
        recoveryFile.writeLong(w.page.getPageId());
        recoveryFile.write(w.getDiskBound(), 0, pageSize);
    }
    .......
    //寫(xiě)入真正的目的文件中(db.redo)
    for (PageWrite w : batch) {
        writeFile.seek(toOffset(w.page.getPageId()));
        writeFile.write(w.getDiskBound(), 0, pageSize);
        w.done();
    }
}

1.默認(rèn)的單機(jī)部署（kahadb）

略......

2.共享存儲(chǔ)主從模式(基于數(shù)據(jù)庫(kù))

3.共享存儲(chǔ)主從模式(基于文件系統(tǒng))

4.基于zookeeper的主從（levelDB Master/Slave詳細(xì)說(shuō)明）

第一步:zookeeper集群搭建

server.1=lvmama01:2888:3888
server.2=lvmama02:2888:3888
server.3=lvmama03:3888:3888

第二步:activemq集群搭建修改activemq.xml文件:

第三步:分別啟動(dòng)三臺(tái)activemq(仔細(xì)查看日志):

1.啟動(dòng)第一臺(tái)機(jī)器(lvmama01:192.168.187.11)

2.啟動(dòng)第二臺(tái)機(jī)器(lvmama02:192.168.187.12)

3.第三臺(tái)啟動(dòng)同第二臺(tái)

第三步:查看是否啟動(dòng)成功(沒(méi)成功可以查看activemq.log日志)

啟動(dòng)成功后通過(guò)zkCli.sh可以看到已創(chuàng)建leveldb-stores如下:

第四步：通過(guò)流量器訪問(wèn)web管理頁(yè)面(注意只有master機(jī)器可以訪問(wèn))

第五步:測(cè)試

        String userName = ActiveMQConnectionFactory.DEFAULT_USER;
        String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
        String brokerURL = "failover:(tcp://192.168.187.11:61616,tcp://192.168.187.12:61616,tcp://192.168.187.13:61616)?Randomize=false";

        //2. 通過(guò)ConnectionFactory建立一個(gè)Connection連接，并且調(diào)用start方法開(kāi)啟
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, password, brokerURL);
        Connection connection = connectionFactory.createConnection();
        connection.start();

        //3. 通過(guò)Connection創(chuàng)建Session，用于接收消息[第一個(gè)參數(shù)：是否啟用事務(wù)；第二個(gè)參數(shù)：設(shè)置簽收模式]
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

        //4. 通過(guò)Session創(chuàng)建Destination對(duì)象
        Destination destination = session.createQueue("cluster-queue");

        //5. 通過(guò)Session創(chuàng)建發(fā)送或接受對(duì)象
        MessageProducer messageProducer = session.createProducer(null);
        

運(yùn)行結(jié)果(此時(shí)發(fā)送的目標(biāo)為192.168.187.11):

Connected to the target VM, address: "127.0.0.1:12266", transport: "socket"
 INFO | Successfully connected to tcp://192.168.187.11:61616
生產(chǎn)者：Hello MQ:1
生產(chǎn)者：Hello MQ:2
生產(chǎn)者：Hello MQ:3
生產(chǎn)者：Hello MQ:4
生產(chǎn)者：Hello MQ:5
生產(chǎn)者：Hello MQ:6
生產(chǎn)者：Hello MQ:7
生產(chǎn)者：Hello MQ:8
生產(chǎn)者：Hello MQ:9

此時(shí)將activemq master服務(wù)停止,集群自動(dòng)重新選舉 lvmama02(192.168.187.12)成為Master

我們?cè)僭囘\(yùn)行測(cè)試用例發(fā)現(xiàn)消息任然可以發(fā)送，只不過(guò)發(fā)送的目標(biāo)變?yōu)?92.168.187.12

Connected to the target VM, address: "127.0.0.1:12400", transport: "socket"
 INFO | Successfully connected to tcp://192.168.187.12:61616
生產(chǎn)者：Hello MQ:1
生產(chǎn)者：Hello MQ:2
生產(chǎn)者：Hello MQ:3
生產(chǎn)者：Hello MQ:4
生產(chǎn)者：Hello MQ:5
生產(chǎn)者：Hello MQ:6
生產(chǎn)者：Hello MQ:7
生產(chǎn)者：Hello MQ:8
生產(chǎn)者：Hello MQ:9

1.安裝Jmeter測(cè)試工具，參考

2.新建jndi.properties到j(luò)meter/bin目錄下

//ActiveMQ jar包中init所需的類(lèi)名
java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory

//ActiveMQ的地址
java.naming.provider.url = tcp://127.0.0.1:61616

//連接工廠名稱(chēng)
connectionFactoryNames = connectionFactory

//p2p 隊(duì)列名稱(chēng)
queue.MyQueue = example.MyQueue
topic.MyTopic = example.MyTopic

3.把配置文件打到ApacheJMeter.jar 中 在jmeter/bin目錄下運(yùn)行

  jar uf ApacheJMeter.jar jndi.properties

4.下載Activemq,并加activemq-all-5.15.3.jar添加到Jmeter/lib下

5.配置Jmeter測(cè)試p2p模式

新建線程組

新建JMS Point-to-Point采樣并配置(參考jndi.properties)

6.進(jìn)行測(cè)試(單線程+60s+10000條消息)

測(cè)試結(jié)果:

Jmeter官網(wǎng)測(cè)試結(jié)果:

可能由于機(jī)器原因，測(cè)試結(jié)果差距蠻大 ^_^