性能工具之Jmeter扩展函数及压测ActiveMQ实践

一、简介

JMeter作为Apache的开源性能测试工具允许使用者对其进行二次开发扩展，比如用户可以扩展自定义的函数（函数是可以在测试脚本中插入到任何Sampler或者测试元素中，可以封装一些功能，比如对用户名加解密函数或者得到一个自定义功能等）

首先本文将以Java扩展一个返回两个数值之和函数的例子来简单演示整个过程。
总体来说，二次开发扩展JMeter的函数可以分成下面几个步骤：

1. 新建Maven项目，引入扩展JMeter Function所需的依赖包
2. 编写实现自定义函数的代码，并对其编译打包
3. 将编译好的包拷贝至JMeter的扩展目录，编辑测试脚本，调用自定义函数
4. 使用并查看自定义的函数是否正确

二、参考

参考JMeter官方API的AbstractFunction，它将指导我们创建自己的函数 &__7DDemoPlus函数，以便我们了解它是怎么生成的。

三、创建Java Maven项目

新建一个Maven项目，打开pom.xml，加入ApacheJMeter_functions依赖库

 <dependencies>
        <dependency>
            <groupId>org.apache.jmeter</groupId>
            <artifactId>ApacheJMeter_functions</artifactId>
            <version>4.0</version>
        </dependency>
    </dependencies>

3.1、扩展自定义函数

要实现二次扩展函数，主要有两点：

• 定义function的类的package声明必须包含.functions
• 需要继承实现AbstractFunction四个方法

定义包名

继承并实现AbstractFunction四个抽象方法：

• execute方法用于接收Jmeter传入的参数值并执行工作
• setParameters方法用于传递执行过程中的实际参数值
• getReferenceKey方法用于定义函数名字
• getArgumentDesc方法用于描述函数参数

public class Plus extends AbstractFunction {

    //定义一个obect对象去接受传入变量值
    private Object[] values;
    //存储第一个和第二个参数
    private CompoundVariable first,second;

    /**
     * 执行方法
     * @param sampleResult
     * @param sampler
     * @return
     * @throws InvalidVariableException
     */
    public String execute(SampleResult sampleResult, Sampler sampler) throws InvalidVariableException {
        //接住第一个参数值
        first = (CompoundVariable) values[0];
        //接住第二个参数值
        second = (CompoundVariable) values[1];

        //计算两个参数的和
        int count = new Integer(first.execute().trim()) + new Integer(second.execute().trim());
        System.out.println("两个参数的和是："+count);

        //返回一个String类型的和
        return String.valueOf(count);
    }

    /**
     * 设置参数，接受用户传递的参数
     * @param collection
     * @throws InvalidVariableException
     */
    public void setParameters(Collection<CompoundVariable> collection) throws InvalidVariableException {
        //检查参数是否合法
        checkParameterCount(collection,2);
        //转换成数组
        values = collection.toArray();
    }

    /**
     * 函数名称
     * @return
     */
    public String getReferenceKey() {
        return "__7DDemoPlus";
    }

    /**
     * 函数描述，获取参数描述
     * @return
     */
    public List<String> getArgumentDesc() {
        List desc = new ArrayList();
        //界面上显示两行参数描述
        desc.add("第一个数字");
        desc.add("第二个数字");

        return desc;
    }
}

在控制台使用 mvn cleanpackage打包

拷贝自定义函数包文件到jmeter/lib/ext目录下

重启Jmeter后打开函数助手，并生成并复制自定义函数表达式

并使用BeanShell调用自定义函数，并检查结果

我们可以看到控制台已经正确输出函数结果了~

至此，我们的Jmeter扩展函数已经完成了，同学们是不是so easy...^_^

四、实践小例子

目前能实现压测ActiveMQ有以下方法：

• JMS Sampler
• 自定义Java请求
• 扩展Sampler
• JSR223 Sampler
• BeanShell Sampler
• 扩展Function（今天介绍的）

接下我们在以上示例的基础实践压测ActiveMQ消息服务

想要了解ActiveMQ是啥，首先得知道JMS，所以先对MOM和JMS做一个介绍

4.1、MOM

企业消息系统，即面向消息的中间件，提供了以松散耦合的灵活方式集成应用程序的一种机制。它们提供了基于存储和转发的应用程序之间的异步数据发送，即应用程序彼此不直接通信，而是与作为中介的MOM通信。

JMS简介

Java Message Service：是Java平台上有关面向消息中间件的技术规范。

有一个比较通俗的解释，JMS类似于JDBC，JDBC是可以用来访问许多不同关系数据库的API，而JMS则提供同样与厂商无关的访问的API，以访问消息收发服务。比如IBM 的MQSeries、BEA的Weblogic JMS service；而ActiveMQ也是其中的一种，所以：activeMQ就是支持jms规范的一个server；它对于JDBC和数据库的关系来说，它就是个mysql（MQSeries就是DB2，Weblogic JMS service就是Oracle）。

在没有JDBC之前，程序员需要访问数据库的时候，需要根据不同的数据库进行不同的编码；在有了JDBC之后，开发过程中，对于不同数据库的访问方法被规范化，只需要根据不同的数据库使用不同的数据库驱动，就可以用通用的方法访问数据库。

在没有JMS之前，程序员开发过程中，如果需要和MOM进行消息发送或接受的时候，需要根据不同的MOM进行不同的编码；相同的，有了JMS之后，代码被规范使用。

4.2、ActiveMQ简介

ActiveMQ是目前最流行的消息中间件之一，是一种在分布式系统中应用程序借以传递消息的媒介，常见的消息中间有ActiveMQ，RabbitMQ，Kafka。ActiveMQ是Apache下的开源项目，完全支持JMS1.1和JSE1.4规范的JMS Provider实现

特点：

• 支持多种语言编写客户端
• 对Spring的支持，很容易和Spring整合
• 支持多种传送协议：TCP，SSL，NIO，UDP等
• 支持AJAX

消息形式：

• 点对点（Queue）
• 一对多（Topic）

4.3、 新建一个消费者maven工程

加入依赖库

 <dependencies>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.4</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>2.11.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.11.1</version>
        </dependency>
    </dependencies>

实现消费者代码类，这里我们使用Queue的消息形式

/**
 * 消费者
 */
public class Receiver {
   
   
    private static final Logger logger = LogManager.getLogger(Receiver.class);

    public static void main(String[] args) {
   
   
        //连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory;

        //JMS client到JMS Provider的连接
        Connection connection = null;

        //一个发送或接受消息的线程
        Session session;

        //消息的目的地，消息发送谁
        Destination destination;

        //消费者，消息接受者
        MessageConsumer consumer;
        connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.5:61616");
        try {
   
   
            //构造从工厂得到连接对象
            connection = connectionFactory.createConnection();

            //启动
            connection.start();

            //获取操作连接
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //获取session的注意参数值，7DQueue是一个消息服务器的队列
            destination = session.createQueue("7DQueue");
            consumer = session.createConsumer(destination);

            while (true) {
   
   
                //设置接收者接收消息的时间
                TextMessage message = (TextMessage) consumer.receive(500000);
                if (null != message) {
   
   
                    //System.out.println("收到消息：" + message.getText());
                    logger.info("收到消息：" + message.getText());

                } else {
   
   
                    break;
                }
            }
        } catch (Exception e) {
   
   
            e.printStackTrace();
        } finally {
   
   
            try {
   
   
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
   
   
            }
        }
    }
}

4.4、新建一个扩展ActiveMQByJmeter maven工程

导入ApacheJMeter_functions依赖库及ActiveMQ相关依赖库

 <dependencies>
        <dependency>
            <groupId>org.apache.jmeter</groupId>
            <artifactId>ApacheJMeter_functions</artifactId>
            <version>4.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.4</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>2.11.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.11.1</version>
        </dependency>
    </dependencies>

实现生产者代码类

/**
 * 生产者
 */
public class Producer {
   
   
    private static final int SEND_NUMBER = 3;
    private static final Logger logger = LogManager.getLogger(Producer.class);

    public static void main(String[] args) {
   
   
        //连接工厂，JMS使用它创建连接
        ConnectionFactory connectionFactory;
        //Provider的连接
        Connection connection = null; //JMS客户端到JMS
        Session session; //一个发送和接受消息的线程
        Destination destination; //消息的目的地，消息发给谁
        MessageProducer producer; //消息发送者
        //构造连接工厂实例对象
        connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.5:61616");
        try {
   
   
            //构造从工厂得到连接对象
            connection = connectionFactory.createConnection();
            //启动
            connection.start();
            //获取操作连接
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            //获取session，注意指定消息队列
            destination = session.createQueue("FirstQueue");
            //得到消息生产者（发送者）
            producer = session.createProducer(destination);
            //设置消息持久化方式
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            //构造消息，此处写死，项目就是参数，或者方法获取
            sendMessage(session, producer);
            session.commit();
        } catch (Exception e) {
   
   
            e.printStackTrace();
        } finally {
   
   
            try {
   
   
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
   
   
            }
        }
    }

    public static void sendMessage(Session session, MessageProducer producer) throws Exception {
   
   
        for (int i = 1; i <= SEND_NUMBER; i++) {
   
   
            //发送消息到目的地方
            TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i);
            //System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
            logger.info("发送消息：" + "ActiveMq 发送的消息" + i);
            producer.send(message);
        }
    }
}

编写一个RunMian测试类

/**
 * 测试类
 */
public class RunMain {
   
   
    public static void main(String[] args) {
   
   
        new Producer().producer
                ("tcp://192.168.1.5:61616","7DQueue","7DTest...");
    }
}

运行RunMian发送消息，查看消费端接受情况

我们看到消费端已经收到消息，测试成功

接下来扩展实现JmeterActiveMQFunction

/**
 * ActiveMQ扩展函数
 */
public class ActiveMQFunction extends AbstractFunction{
   
   
    //定义一个obect对象去接受传入变量值
    private Object[] values;
    //存储三个参数
    private CompoundVariable brokerURL,Queue,message;

    /**
     * 执行方法
     * @param sampleResult
     * @param sampler
     * @return
     * @throws InvalidVariableException
     */
    public String execute(SampleResult sampleResult, Sampler sampler) throws InvalidVariableException {
   
   
        //接住服务地址URL参数值
        brokerURL = (CompoundVariable) values[0];
        //接住消息队列参数值
        Queue = (CompoundVariable) values[1];
        //接住消息内容参数值
        message = (CompoundVariable) values[2];

        //把参数类型转成字符串
        String strBrokerURL = brokerURL.execute().toString();
        String strQueue = Queue.execute().toString();
        String strMessage = message.execute().toString();

        //指定目的地，发送消息
        new Producer().producer(strBrokerURL,strQueue,strMessage);

        //返回值
        return null;
    }

    /**
     * 设置参数，接受用户传递的参数
     * @param collection
     * @throws InvalidVariableException
     */
    public void setParameters(Collection<CompoundVariable> collection) throws InvalidVariableException {
   
   
        //检查参数是否合法
        checkParameterCount(collection,3);
        //转换成数组
        values = collection.toArray();
    }

    /**
     * 函数名称
     * @return
     */
    public String getReferenceKey() {
   
   
        return "__7DGroupActiveMQ";
    }

    /**
     * 函数描述，获取参数描述
     * @return
     */
    public List<String> getArgumentDesc() {
   
   
        List desc = new ArrayList();
        //界面上显示两行参数描述
        desc.add("服务地址");
        desc.add("消息队列");
        desc.add("消息内容");

        return desc;
    }
}

在控制台使用 mvn clean package打包测试类

使用 mvn dependency:copy-dependencies -DoutputDirectory=lib复制所依赖的jar包都会到项目下的lib目录下

复制测试代码Jar包到jmeter\lib\ext目录下，复制依赖包到jmeter\lib目录下

重启Jmeter后打开函数助手，并生成并复制自定义函数表达式，使用 __Random函数对消息内容简单参数化

下面我们将进行性能压测，设置线程组，设置5个并发线程。定义并使用BeanShell调用自定义函数

我们可以看到消费端已经接收到消息

至此，压测ActiveMQ成功了~

本文源码：
https://github.com/zuozewei/blog-example/tree/master/Performance-testing/01-test-tool/jmeter/activemq