摘要：年月日上午阿里云消息服，隊列消息發送以及消費的并發測試解析配置文件二者等價線程數并發數程序入口準備工作發送消息線程池一個計數信號量。但是，不使用實際的許可對象，只對可用許可的號碼進行計數，并采取相應的行動。

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)，隊列消息發送以及消費的并發測試 
 *                                   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");// 二者等價
        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: 線程數" + threadNum);
        clientNum = Integer.parseInt(properties.getProperty("ClientNum",
                String.valueOf(clientNum)));
        LOG.info("ClientNum: 并發數" + 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 {
        // 準備工作
        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("發送消息");
        // 線程池
        ExecutorService exec = Executors.newFixedThreadPool(500);
        /**
         * Semaphore 一個計數信號量。從概念上講，信號量維護了一個許可集。如有必要，在許可可用前會阻塞每一個
         * acquire()，然后再獲取該許可。每個 release()
         * 添加一個許可，從而可能釋放一個正在阻塞的獲取者。但是，不使用實際的許可對象，Semaphore 只對可用許可的號碼進行計數
         * ，并采取相應的行動。拿到信號量的線程可以進入代碼，否則就等待。通過acquire()和release()獲取和釋放訪問許可。
         */
        final Semaphore semp = new Semaphore(threadNum);// ["sem?f??]
        final Semaphore semaphore = new Semaphore(10, true);
                                                        // 拿到信號量的線程可以進入代碼，否則就等待
        // Semaphore（信號量）是用來控制同時訪問特定資源的線程數量，它通過協調各個線程，以保證合理的使用公共資源
        // 輔助理解 ：很多年以來，我都覺得從字面上很難理解Semaphore所表達的含義，只能把它比作是控制流量的紅綠燈，
        // 比如XX馬路要限制流量，只允許同時有一百輛車在這條路上行使，其他的都必須在路口等待，所以前一百輛車會看到綠燈，
        // 可以開進這條馬路，后面的車會看到紅燈，不能駛入XX馬路，但是如果前一百輛中有五輛車已經離開了XX馬路，
        // 那么后面就允許有5輛車駛入馬路，這個例子里說的車就是線程，駛入馬路就表示線程在執行，離開馬路就表示線程執行完成，看見紅燈就表示線程被阻塞，不能執行。
        long startTime = System.currentTimeMillis(); // 開啟時間
        /**
         * 原理:
         * 更進一步，信號量的特性如下：信號量是一個非負整數（車位數），所有通過它的線程（車輛）都會將該整數減一（通過它當然是為了使用資源），
         * 當該整數值為零時，所有試圖通過它的線程都將處于等待狀態。在信號量上我們定義兩種操作： Wait（等待） 和 Release（釋放）。
         * 當一個線程調用Wait
         * （等待）操作時，它要么通過然后將信號量減一，要么一直等下去，直到信號量大于一或超時。Release（釋放）實際上是在信號量上執行加操作
         * ，對應于車輛離開停車場，該操作之所以叫做“釋放”是因為加操作實際上是釋放了由信號量守護的資源。
         */
        // 開始
        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");
                            // 發送消息
                            queue.putMessage(message);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        semp.release();// 歸還許可
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            exec.submit(task);
        }
        long endTime = System.currentTimeMillis(); // 開啟時間
        exec.shutdown();
        LOG.info(clientNum + " 的并發發送消息總耗時：>>>" + (endTime - startTime) + " ms");
        LOG.info(clientNum + " 的并發發送消息 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(); // 開啟時間
        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(); // 開啟時間
        exec2.shutdown();
        // 忽略線程切換的耗時 精確的做法？
        LOG.info(clientNum + " 的并發接收消息總耗時：>>>" + (endTime2 - startTime2)

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

}