Gatling JMS 场景不会终止

Question

我正在尝试通过 rabbitmq 代理对一个简单的 request/reply 场景进行负载测试。

import com.rabbitmq.jms.admin.RMQConnectionFactory;
import io.gatling.javaapi.core.ScenarioBuilder;
import io.gatling.javaapi.core.Simulation;
import io.gatling.javaapi.jms.JmsProtocolBuilder;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import java.util.Collections;
import java.util.UUID;

import static io.gatling.javaapi.core.CoreDsl.*;
import static io.gatling.javaapi.jms.JmsDsl.*;

public class ReplySimulation extends Simulation {

    ConnectionFactory connectionFactory = connectionFactory();

    JmsProtocolBuilder jmsProtocolBuilder = jms.connectionFactory( connectionFactory ).usePersistentDeliveryMode().listenerThreadCount( 1 );

    ScenarioBuilder scn = scenario( "Reply Test" ).repeat( 1 ).on(
        exec(
            jms( "request" )
                .requestReply()
                .queue( "requests" ).replyQueue( "loadtest-" + UUID.randomUUID() )
                .textMessage( "Message Body" )
                .jmsType( "textMessage" )
                .check( simpleCheck( message -> true ) ) ) );

    {
        setUp( scn.injectOpen(atOnceUsers( 1 ) )).protocols( jmsProtocolBuilder );
    }

    public ReplySimulation() throws JMSException {

    }

    private ConnectionFactory connectionFactory() throws JMSException {

        RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
        connectionFactory.setUsername( "guest" );
        connectionFactory.setUsername( "guest" );
        connectionFactory.setUris( Collections.singletonList( "amqp://localhost:5672" ) );
        connectionFactory.setVirtualHost( "loadtest" );
        return connectionFactory;
    }
}

此场景向定义的 queue 发送一条简单的文本消息，然后等待定义的 replyQueue 回复。在请求的另一端 queue 是简单的消息侦听器，它读取 JMSReplyTo 字段并将消息发送到该目的地。

在 rabbitmq UI 中，我可以看到有一条消息发布到回复 queue 并且有一个消费者确认。所以这告诉我往返已经完成。

但出于某种原因，加特林机场景一直运行给我这样的输出：



================================================================================
2021-12-15 16:38:01                                          15s elapsed
---- Requests ------------------------------------------------------------------
> Global                                                   (OK=0      KO=0     )


---- Reply Test ----------------------------------------------------------------
[--------------------------------------------------------------------------]  0%
          waiting: 0      / active: 1      / done: 0     
================================================================================

所以在我看来，检查没有触发，否则应该显示为 active: 0 / done: 1。我尝试检查 return false 和 null 甚至抛出 RuntimeException 或添加错误输出。什么都没有改变，所以看起来确实没有执行检查或各种输出以某种方式被吞没。

simulation.log 除了两行 header 之外不包含任何其他内容，并且将 maxDuration 添加到执行中确实会停止该场景，但随后它会失败并出现 ArithmeticException 作为除以零发生。

到目前为止我找到的示例看起来都非常相似，所以我看不出有什么问题，但显然有些地方不起作用。

编辑：这是场景中使用的回显服务

public class DeskclientEcho {

    private enum Parameter {
        CONNECTIONS, EXCHANGE, PASSWORD, QUEUE, SSL( 0 ), USER, VHOST;

        private final int argCount;

        Parameter() {

            this( 1 );
        }

        Parameter( int argCount ) {

            this.argCount = argCount;
        }
    }

    private static final Map<Parameter, String> params = new HashMap<>();

    public static void main( String[] args ) throws JMSException, NoSuchAlgorithmException {

        for ( int i = 0; i < args.length; i++ ) {
            Parameter key = switch ( args[i] ) {
                case "-c", "--connections" -> Parameter.CONNECTIONS;
                case "-e", "--exchange" -> Parameter.EXCHANGE;
                case "-p", "--password" -> Parameter.PASSWORD;
                case "-q", "--queue" -> Parameter.QUEUE;
                case "-s", "--ssl" -> Parameter.SSL;
                case "-u", "--user" -> Parameter.USER;
                case "-v", "--vhost" -> Parameter.VHOST;
                default -> throw new IllegalArgumentException("Unknown parameter :" + args[i]);
            };
            i += key.argCount;
            params.put( key, args[i] );
        }

        RMQConnectionFactory factory = new RMQConnectionFactory();
        factory.setUris( Arrays.stream( params.get( Parameter.CONNECTIONS ).split( "," ) ).filter( Objects::nonNull ).toList() );

        if ( params.containsKey( Parameter.SSL ) ) {
            factory.useSslProtocol();
            factory.setUseDefaultSslContext( true );
            factory.setHostnameVerification( true );
        }

        Optional.ofNullable(params.get(Parameter.USER)).ifPresent( factory::setUsername );
        Optional.ofNullable(params.get(Parameter.PASSWORD)).ifPresent( factory::setPassword );
        Optional.ofNullable(params.get(Parameter.VHOST)).ifPresent( factory::setVirtualHost );

        RMQDestination destination = new RMQDestination();
        destination.setAmqp( false );
        destination.setDestinationName( params.get( Parameter.QUEUE ) );
        destination.setAmqpExchangeName( params.get( Parameter.EXCHANGE ) );
        destination.setAmqpQueueName( params.get( Parameter.QUEUE ) );

        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );

        MessageConsumer consumer = session.createConsumer( destination );

        consumer.setMessageListener( message -> {
            System.out.println("Message received");
            try {
                Destination replyTo = message.getJMSReplyTo();
                for (int i = 0 ; i < 10 ; i++) {
                    session.createProducer( replyTo ).send( new RMQTextMessage() );
                }
                System.out.println("Message sent");
            } catch ( JMSException e ) {
                e.printStackTrace();
            }

        } );

    }
}

有了这个pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>loadtest</groupId>
    <artifactId>deskclient-echo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.rabbitmq.jms</groupId>
            <artifactId>rabbitmq-jms</artifactId>
            <version>2.3.0</version>
        </dependency>
    </dependencies>


</project>

作为经纪人，使用默认的 docker 图像就足够了：

docker run -d rabbitmq

然后创建所需的虚拟主机或更改模拟以使用默认虚拟主机并启动 echo 应用程序。

模拟使用这个pom：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>loadtest</groupId>
    <artifactId>deskclient-requests</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
    </properties>

    <build>
        <plugins>
            <plugin>
                <groupId>io.gatling</groupId>
                <artifactId>gatling-maven-plugin</artifactId>
                <version>4.0.1</version>
                <configuration>
                    <simulationClass>ReplySimulation</simulationClass>
                </configuration>
            </plugin>
        </plugins>

    </build>


    <dependencies>
        <dependency>
            <groupId>io.gatling.highcharts</groupId>
            <artifactId>gatling-charts-highcharts</artifactId>
            <version>3.7.2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.rabbitmq.jms</groupId>
            <artifactId>rabbitmq-jms</artifactId>
            <version>2.3.0</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

有人知道我在这里遗漏了什么吗？

Answer 1

核心问题是您没有在 echo 服务中定义 Gatling 关联出站消息（请求）和入站消息（响应）的任何方式。

相反，您使用 new RMQTextMessage() 进行回复，因此使用完全不同的 JMSMessageID 而没有 JMSCorrelationID。

参见 https://gatling.io/docs/gatling/reference/current/jms/#other-options。

在你的情况下，你可能应该使用 matchByCorrelationId 并相应地传播它：

Message response = new RMQTextMessage();
response.setJMSCorrelationID(message.getJMSCorrelationID());
session.createProducer(replyTo).send(response);

注意：Gatling JMS 支持目前仅支持检查（请求、响应）对，因此无法像使用 for (int i = 0 ; i < 10 ; i++) 循环那样检查一对多响应。

Gatling JMS 场景不会终止

Gatling JMS scenario does not terminate

java

rabbitmq

jms

gatling