摘要：年月日上午阿里云消息服，隊(duì)列消息發(fā)送以及消費(fèi)的并發(fā)測(cè)試解析配置文件二者等價(jià)線程數(shù)并發(fā)數(shù)程序入口準(zhǔn)備工作發(fā)送消息線程池一個(gè)計(jì)數(shù)信號(hào)量。但是，不使用實(shí)際的許可對(duì)象，只對(duì)可用許可的號(hào)碼進(jìn)行計(jì)數(shù)，并采取相應(yīng)的行動(dòng)。

package com.study.mq.aliyunmns;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URL;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

import org.apache.commons.lang3.SystemUtils;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.common.http.ClientConfiguration;
import com.aliyun.mns.model.Message;

/**
 * 
 * @author wangkai
 * @2016年11月22日 上午11:27:14
 * @desc:阿里云消息服(MNS)，隊(duì)列消息發(fā)送以及消費(fèi)的并發(fā)測(cè)試 
 *                                   https://www.aliyun.com/product/mns?spm=5176.8142029
 *                                   .388261.80.fNnCkg
 */
public class MnsQueueAppV2 {

    private static Logger LOG = Logger.getLogger(MnsQueueAppV2.class.getName());

    private static MNSClient client = null;
    // private static AtomicLong totalCount = new AtomicLong(0);
    private static String endpoint = null;
    private static String accessId = null;
    private static String accessKey = null;
    private static String queueName = "articlepricinglog";
    private static int threadNum = 100;
    private static int clientNum = 10000;
    // private static int totalSeconds = 180;
    private static String log4jConfPath = "./log4j.properties";

    static {
        PropertyConfigurator.configureAndWatch(log4jConfPath);
    }

    /**
     * 解析配置文件
     * 
     * @return
     */
    @SuppressWarnings("unused")
    protected static boolean parseConf() {
        // URL resource =
        // MnsQueueAppV2.class.getClassLoader().getResource("name.properties");
        String confFilePath = SystemUtils.getUserDir()
                + SystemUtils.FILE_SEPARATOR
                + "src/main/resources/mns.properties";
        URL resource = MnsQueueAppV2.class.getResource("/mns.properties");
        URL resource2 = MnsQueueAppV2.class.getClassLoader().getResource(
                "mns.properties");// 二者等價(jià)
        BufferedInputStream bis = null;
        try {
            bis = new BufferedInputStream(new FileInputStream(confFilePath));
            if (bis == null) {
                LOG.info("ConfFile not opened: " + confFilePath);
                return false;
            }
        } catch (FileNotFoundException e) {
            LOG.error("ConfFile not found: " + confFilePath, e);
            return false;
        }
        // load file
        Properties properties = new Properties();
        try {
            properties.load(bis);
        } catch (IOException e) {
            LOG.error("Load ConfFile Failed: " + e.getMessage());
            return false;
        } finally {
            try {
                bis.close();
            } catch (Exception e) {
                // do nothing
            }
        }
        // init the member parameters
        endpoint = properties.getProperty("Endpoint");
        LOG.info("Endpoint: " + endpoint);
        accessId = properties.getProperty("AccessId");
        LOG.info("AccessId: " + accessId);
        accessKey = properties.getProperty("AccessKey");
        queueName = properties.getProperty("QueueName", queueName);
        LOG.info("QueueName: " + queueName);
        threadNum = Integer.parseInt(properties.getProperty("ThreadNum",
                String.valueOf(threadNum)));
        LOG.info("ThreadNum: 線程數(shù)" + threadNum);
        clientNum = Integer.parseInt(properties.getProperty("ClientNum",
                String.valueOf(clientNum)));
        LOG.info("ClientNum: 并發(fā)數(shù)" + clientNum);
        // totalSeconds =
        // Integer.parseInt(properties.getProperty("TotalSeconds",
        // String.valueOf(totalSeconds)));
        // LOG.info("TotalSeconds: " + totalSeconds);
        return true;
    }

    /**
     * 程序入口
     * 
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        // 準(zhǔn)備工作
        if (!parseConf()) {
            return;
        }

        ClientConfiguration clientConfiguration = new ClientConfiguration();
        clientConfiguration.setMaxConnections(threadNum);
        clientConfiguration.setMaxConnectionsPerRoute(threadNum);
        CloudAccount cloudAccount = new CloudAccount(accessId, accessKey,
                endpoint, clientConfiguration);
        client = cloudAccount.getMNSClient();
        LOG.info("發(fā)送消息");
        // 線程池
        ExecutorService exec = Executors.newFixedThreadPool(500);
        /**
         * Semaphore 一個(gè)計(jì)數(shù)信號(hào)量。從概念上講，信號(hào)量維護(hù)了一個(gè)許可集。如有必要，在許可可用前會(huì)阻塞每一個(gè)
         * acquire()，然后再獲取該許可。每個(gè) release()
         * 添加一個(gè)許可，從而可能釋放一個(gè)正在阻塞的獲取者。但是，不使用實(shí)際的許可對(duì)象，Semaphore 只對(duì)可用許可的號(hào)碼進(jìn)行計(jì)數(shù)
         * ，并采取相應(yīng)的行動(dòng)。拿到信號(hào)量的線程可以進(jìn)入代碼，否則就等待。通過(guò)acquire()和release()獲取和釋放訪問(wèn)許可。
         */
        final Semaphore semp = new Semaphore(threadNum);// ["sem?f??]
        final Semaphore semaphore = new Semaphore(10, true);
                                                        // 拿到信號(hào)量的線程可以進(jìn)入代碼，否則就等待
        // Semaphore（信號(hào)量）是用來(lái)控制同時(shí)訪問(wèn)特定資源的線程數(shù)量，它通過(guò)協(xié)調(diào)各個(gè)線程，以保證合理的使用公共資源
        // 輔助理解 ：很多年以來(lái)，我都覺(jué)得從字面上很難理解Semaphore所表達(dá)的含義，只能把它比作是控制流量的紅綠燈，
        // 比如XX馬路要限制流量，只允許同時(shí)有一百輛車在這條路上行使，其他的都必須在路口等待，所以前一百輛車會(huì)看到綠燈，
        // 可以開(kāi)進(jìn)這條馬路，后面的車會(huì)看到紅燈，不能駛?cè)隭X馬路，但是如果前一百輛中有五輛車已經(jīng)離開(kāi)了XX馬路，
        // 那么后面就允許有5輛車駛?cè)腭R路，這個(gè)例子里說(shuō)的車就是線程，駛?cè)腭R路就表示線程在執(zhí)行，離開(kāi)馬路就表示線程執(zhí)行完成，看見(jiàn)紅燈就表示線程被阻塞，不能執(zhí)行。
        long startTime = System.currentTimeMillis(); // 開(kāi)啟時(shí)間
        /**
         * 原理:
         * 更進(jìn)一步，信號(hào)量的特性如下：信號(hào)量是一個(gè)非負(fù)整數(shù)（車位數(shù)），所有通過(guò)它的線程（車輛）都會(huì)將該整數(shù)減一（通過(guò)它當(dāng)然是為了使用資源），
         * 當(dāng)該整數(shù)值為零時(shí)，所有試圖通過(guò)它的線程都將處于等待狀態(tài)。在信號(hào)量上我們定義兩種操作： Wait（等待） 和 Release（釋放）。
         * 當(dāng)一個(gè)線程調(diào)用Wait
         * （等待）操作時(shí)，它要么通過(guò)然后將信號(hào)量減一，要么一直等下去，直到信號(hào)量大于一或超時(shí)。Release（釋放）實(shí)際上是在信號(hào)量上執(zhí)行加操作
         * ，對(duì)應(yīng)于車輛離開(kāi)停車場(chǎng)，該操作之所以叫做“釋放”是因?yàn)榧硬僮鲗?shí)際上是釋放了由信號(hào)量守護(hù)的資源。
         */
        // 開(kāi)始
        for (int index = 0; index < clientNum; index++) {

            // final int NO = index;

            Runnable task = new Runnable() {
                public void run() {
                    try {
                        semp.acquire();// 獲取許可
                        try {
                            // 獲取queue
                            CloudQueue queue = client.getQueueRef(queueName);
                            // 組裝消息
                            Message message = new Message();
                            message.setMessageBody("Test");
                            // 發(fā)送消息
                            queue.putMessage(message);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        semp.release();// 歸還許可
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            exec.submit(task);
        }
        long endTime = System.currentTimeMillis(); // 開(kāi)啟時(shí)間
        exec.shutdown();
        LOG.info(clientNum + " 的并發(fā)發(fā)送消息總耗時(shí)：>>>" + (endTime - startTime) + " ms");
        LOG.info(clientNum + " 的并發(fā)發(fā)送消息 QPS為：>>>" + (clientNum * 1000)
                / (endTime - startTime) + " q/s");
        LOG.info("接收消息");
        Thread.sleep(3000);
        ExecutorService exec2 = Executors.newFixedThreadPool(500);
        final Semaphore semp2 = new Semaphore(threadNum);
        long startTime2 = System.currentTimeMillis(); // 開(kāi)啟時(shí)間
        for (int index = 0; index < clientNum; index++) {
            // final int NO = index;
            Runnable task = new Runnable() {
                public void run() {
                    try {
                        semp2.acquire();
                        try {
                            // 獲取queue
                            CloudQueue queue = client.getQueueRef(queueName);
                            // 獲取消息
                            Message message = queue.popMessage();
                            // 刪掉消息
                            if (message != null)
                                queue.deleteMessage(message.getReceiptHandle());
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        semp2.release();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            exec2.submit(task);
        }
        long endTime2 = System.currentTimeMillis(); // 開(kāi)啟時(shí)間
        exec2.shutdown();
        // 忽略線程切換的耗時(shí) 精確的做法？
        LOG.info(clientNum + " 的并發(fā)接收消息總耗時(shí)：>>>" + (endTime2 - startTime2)

            + " ms");
    LOG.info(clientNum + " 的并發(fā)接收消息 QPS為：>>>" + (clientNum * 1000)
            / (endTime2 - startTime2) + " q/s");
}

}