測試工具：最新版JMeter 3.1

RocketMQ：1m單點或2m-2s-async叢集      

   版本apache-rocketmq 4.0.0-incubating（參考 點選開啟連結）

JDK：Java7+

純粹生產者TPS壓測

public class Producter extends AbstractJavaSamplerClient{

    public SampleResult runTest(JavaSamplerContext javaSamplerContext) {

        SampleResult sr = new SampleResult();
        sr.setSampleLabel("RocketMQ測試");
        try {
            sr.sampleStart();
            DefaultMQProducer producer = new DefaultMQProducer(MixUtil.createGroupName("ProducerGroupName_"));
            producer.setNamesrvAddr(MixUtil.NAME_SRV_ADDR);
            producer.setInstanceName("Producer");
            producer.setVipChannelEnabled(false);

            producer.start();// once
            int times = 1000;
            for(int i=0;i<times;++i){
                try {
                    {
                        Message msg = new Message("WangXiaoRuiTopic", "TagA", "WangXiaoRui", ("王小瑞，花名誓嘉，阿里巴巴中介軟體訊息團隊負責人，具有豐富的高可用，高可靠分散式系統構建經驗，主導了阿里巴巴多次雙十一訊息引擎的改進優化專案，擁有多項分散式領域的專利。Apache RocketMQ聯合創始人").getBytes());
                        SendResult sendResult = producer.send(msg);
                        if(sendResult ==null || sendResult.getSendStatus() != SendStatus.SEND_OK){
                            System.err.println(sendResult);
                        }
                    }

                    {
                        Message msg = new Message("FengJiaTopic", "TagB", "FengJia", ("馮嘉，花名鼬神，阿里巴巴中介軟體架構師，具有豐富的分散式軟體架構、高併發網站設計、效能調優經驗，擁有多項分散式領域的專利。開源愛好者，專注分散式、大資料領域，關注Hbase/Hadoop/Spark/Flink等大資料技術棧。目前負責阿里訊息中介軟體生態輸出、雲上商業化，Apache RocketMQ聯合創始人。聯絡方式: 
 
[email protected]").getBytes());
                        SendResult sendResult = producer.send(msg);
                        if(sendResult ==null || sendResult.getSendStatus() != SendStatus.SEND_OK){
                            System.err.println(sendResult);
                        }
                    }

                    {
                        Message msg = new Message("RocketMQHostTopic", "TagC", "RocketMQHist", ("第三代，以拉模式為主，兼有推模式的高效能、低延遲訊息引擎RocketMQ，在二代功能特性的基礎上，為電商金融領域添加了可靠重試、基於檔案儲存的分散式事務等特性，並做了大量優化。從2012年開始，經歷了歷次雙11核心交易鏈路檢驗。目前已經捐贈給Apache基金會。時至今日，RocketMQ很好的服務了阿里集團大大小小上千個應用，在雙11當天，更有不可思議的萬億級訊息流轉，為集團大中臺的穩定發揮了舉足輕重的作用").getBytes());
                        SendResult sendResult = producer.send(msg);
                        if(sendResult ==null || sendResult.getSendStatus() != SendStatus.SEND_OK){
                            System.err.println(sendResult);
                        }
                    }

                }catch (Exception e){
                    e.printStackTrace();;
                }
            }

            producer.shutdown();

            sr.setResponseData("success","utf-8");
            sr.setDataType(SampleResult.TEXT);
            sr.setSuccessful(true);
        }catch(Exception e){
            sr.setSuccessful(false);
            sr.setResponseData(MixUtil.getStackTrace(e),"utf-8");
            e.printStackTrace();
        }
        finally {

            sr.sampleEnd();
        }


        return sr;
    }
...
 

沒有純粹的消費者TPS壓測例子。。生產和消費混合TPS壓測

package cn.treebear.rocketmq.tester;

import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient;
import org.apache.jmeter.protocol.java.sampler.JavaSamplerContext;
import org.apache.jmeter.samplers.SampleResult;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by zhujl on 2017/4/13.
 */
public class Consumer extends AbstractJavaSamplerClient {

    public SampleResult runTest(JavaSamplerContext javaSamplerContext) {

        SampleResult sr = new SampleResult();
        sr.setSampleLabel("RocketMQ消費者測試");
        try {
            sr.sampleStart();
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MixUtil.createGroupName("ConsumerGroupname_"));
            consumer.setNamesrvAddr(MixUtil.NAME_SRV_ADDR);
            consumer.setInstanceName("Consumer");

            consumer.subscribe("WangXiaoRuiTopic", "TagA || TagC || TagD");

            consumer.subscribe("FengJiaTopic", "*");

            consumer.registerMessageListener(new OutMessageListener());


            consumer.start();
            System.out.println("Consumer Started.");

            TimeUnit.MILLISECONDS.sleep(10 * 1000);

            consumer.shutdown();

            sr.setResponseData("success","utf-8");
            sr.setDataType(SampleResult.TEXT);
            sr.setSuccessful(true);
        }catch (Exception e){
            sr.setSuccessful(false);
            sr.setResponseData(MixUtil.getStackTrace(e),"utf-8");
            e.printStackTrace();
        }finally {
            sr.sampleEnd();
        }

        return sr;
    }
}
 

package cn.treebear.rocketmq.tester;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.io.Writer;
import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.Enumeration;

/**
 * Created by zhujl on 2017/4/13.
 */
public class MixUtil {

    public final static String NAME_SRV_ADDR = "namesrv1:9876;namesrv2:9876";
    public static String getStackTrace(Throwable aThrowable) {
        final Writer result = new StringWriter();
        final PrintWriter printWriter = new PrintWriter(result);
        aThrowable.printStackTrace(printWriter);
        return result.toString();
    }

    public  static String getIpAddr(){
        String ip="";
        try {
            for (Enumeration<NetworkInterface> en = NetworkInterface.getNetworkInterfaces(); en.hasMoreElements(); ) {
                NetworkInterface intf = en.nextElement();
                String name = intf.getName();
                if (!name.contains("docker") && !name.contains("lo")) {
                    for (Enumeration<InetAddress> enumIpAddr = intf.getInetAddresses(); enumIpAddr.hasMoreElements(); ) {
                        //獲得IP
                        InetAddress inetAddress = enumIpAddr.nextElement();
                        if (!inetAddress.isLoopbackAddress()) {
                            String ipaddress = inetAddress.getHostAddress().toString();
                            if (!ipaddress.contains("::") && !ipaddress.contains("0:0:") && !ipaddress.contains("fe80")) {

                                System.out.println(ipaddress);
                                if (!"127.0.0.1".equals(ip)) {
                                    ip = ipaddress;
                                }
                            }
                        }
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return ip;
    }

    public static final int getProcessID() {
        RuntimeMXBean runtimeMXBean = ManagementFactory.getRuntimeMXBean();
        System.out.println(runtimeMXBean.getName());
        return Integer.valueOf(runtimeMXBean.getName().split("@")[0])
                .intValue();
    }

    public static String createGroupName(String groupPrefix){
        return groupPrefix +"_" + getIpAddr().replaceAll("\\.","_") + "_"+getProcessID()+"_"+Thread.currentThread().getId();
    }


    public static void main(String[] args){
        System.out.println(MixUtil.createGroupName("ProducerGroupName_"));
    }
}

package cn.treebear.rocketmq.tester;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * Created by zhujl on 2017/4/17.
 *
 */
public class OutMessageListener implements MessageListenerConcurrently {
    /**
     * 預設msgs裡只有一條訊息，可以通過設定consumeMessageBatchMaxSize引數來批量接收訊息
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        System.out.println(Thread.currentThread().getName() + " Receive new Message:" + msgs.size());
        MessageExt msg = msgs.get(0);
        if (msg.getTopic().equals("WangXiaoRuiTopic")) {
            if (msg.getTags() != null && msg.getTags().equals("TagA")) {
                System.out.println("do with TagA,msg:" + msg);
            } else if (msg.getTags() != null && msg.getTags().equals("TagC")) {
                System.out.println("do with TagC,msg:" + msg);
            } else if (msg.getTags() != null && msg.getTags().equals("TagC")) {
                System.out.println("do with TagC,msg:" + msg);
            }
        } else if (msg.getTopic().equals("TopicTest2")) {
            System.out.println(new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

將原始碼打為jar包與其它依賴的包一起拷貝到jmeter擴充套件目錄

新建執行計劃，新增java請求節點，選擇生產者或者消費者