JMeter压测RabbitMQ实践：自定义Java Sampler实现高并发生产者压测

简介：面向RabbitMQ性能测试的JMeter工具包，适用于消息中间件运维、测试开发及架构评估人员，针对性解决RabbitMQ生产与消费链路的高并发压力测试问题。压缩包共2881个文件，大小约53.02MB，以html文档、png图示、jar插件和jmx测试计划为主，辅以js、less、coffee等脚本资源，便于阅读原理并直接复用。目前已有1851人学习。资源包含RabbitMQ JMS采样器相关组件、可运行的JMeter脚本、线程组与监听器配置示例，覆盖生产者和消费者两类压测场景；同时提供图文使用说明，帮助快速搭建环境、设置队列与交换机参数，并借助聚合报告分析吞吐量、响应时间和错误率。无论验证集群容量上限还是定位性能瓶颈，都能提供从环境准备到结果解读的完整支持。

1. 为什么我选择用JMeter压测RabbitMQ

做后端服务的人应该都有这种体会：接口压测工具一抓一大把，但到了消息队列这个环节，网上能直接抄的方案就不多了。我之前接到过一个需求，要评估公司RabbitMQ集群到底能扛多大吞吐量，为后续业务扩容提供依据。一开始想得很简单，直接写个Java生产者脚本，循环发消息就完事了。但真的做下来才发现，这个思路太天真了——消息体大小怎么控制？发送速率怎么调整？多线程发消息怎么设计？结果怎么统计？这些问题全都得自己重新造轮子。

后来我换了个思路，回到我最熟悉的JMeter上。虽然JMeter的强项是HTTP接口压测，但它对AMQP协议并非完全无计可施。标题里这个apache-jmeter-rabbitMQ测试.zip，其实就是我最终整理出的一套完整压测方案：JMeter作为压测框架，通过自定义Java Sampler的方式对接RabbitMQ，实现生产者的高并发消息发送。配合JMeter自带的线程组、聚合报告、结果树这些能力，整个压测过程变得非常直观，也方便给团队其他人复用。

这套方案最大的价值在于，用JMeter做压测的人很多，用RabbitMQ的业务团队也很多，但真正把两者打通并且能直接落地的案例却很少。如果你是测试工程师、中间件运维人员，或者负责系统性能评估的后端开发，这篇内容应该能帮你省掉不少弯路。

1.1 直接写Java脚本和用JMeter压测，差距在哪里

不用JMeter，直接用Java代码写生产者压测RabbitMQ，也不是不行。我之前也这么干过，但踩了几个坑之后发现效率确实低。

首先是脚本管理问题。你每次想调整并发数、消息条数、消息大小，都要去改代码，重新编译打包。压测过程中想动态调整参数？做不到。而JMeter的线程属性和参数化配置，直接在GUI界面上改就行，改完立即生效，不用重新编译。

其次是结果统计。自己写脚本，你通常只能记录发送成功的总数和耗时，然后自己算TPS。但要分析响应时间分布（p50、p95、p99）？要观察吞吐量随时间的变化趋势？就得自己造数据，再去Excel里折腾，非常费劲。JMeter的聚合报告、图表监听器直接给你画好，还能导出CSV做二次分析，省太多工作。

第三是团队协作。你写了个Java类，同事要复用得先看懂你的代码逻辑。但JMeter测试计划是图形化的，任何懂JMeter的人拿过去就能看懂测试逻辑和参数配置，哪怕不熟悉RabbitMQ细节也能快速上手跑起来。

1.2 这套压测方案能解决哪些问题

实际业务里，RabbitMQ的压测需求通常来自这么几个场景：

一是容量评估。新系统上线前，需要知道这个MQ集群最大能承受多高的消息生产速率，以及在高吞吐下消费者是否能及时消费。二是消费能力验证。上游系统每秒生产5000条消息，下游消费者能否跟得上？跟不上就会出现消息积压，影响业务实时性。三是配置调优。比如prefetch count设多少合适？并发消费者线程数开多大？这些参数对性能的影响是实打实的，但凭感觉调很容易跑偏，需要用压测数据说话。

标题里这套方案，主要是解决“生产端压测”的问题，也就是模拟大量消息生产者往RabbitMQ里发消息，评估Broker的接收能力和整体链路的表现。如果你还需要同时验证消费者的能力，也可以在这套方案基础上扩展，思路是相通的。

2. 压测前的准备：部署、环境与核心概念

工欲善其事必先利其器。在写JMeter的Sampler代码之前，我先把环境和基础概念捋清楚了。

2.1 RabbitMQ部署与环境准备

我自己习惯用Docker部署RabbitMQ，方便快速拉起一个测试环境，用完就删，不污染本地开发环境。一条命令就能搞定：

docker run -d --name rabbitmq \ -p 5672:5672 -p 15672:15672 \ -e RABBITMQ_DEFAULT_USER=admin \ -e RABBITMQ_DEFAULT_PASS=admin123 \ rabbitmq:3.13-management

这里有两个端口要特别注意：5672是AMQP协议端口，JMeter发消息走的就是这个；15672是管理控制台端口，用来查看队列状态、消息速率等监控数据。镜像选了带management后缀的版本，因为里面有Web管理界面，调试的时候能直接看到队列里有没有消息进来，非常方便。

RabbitMQ装好之后，我还要确认JMeter的Java环境没问题。JMeter本身是Java应用，JDK版本建议8以上，我用的是11，稳定性没问题。另外，需要用到的RabbitMQ Java客户端依赖包，我在后面编写自定义Sampler时会详细说明。

2.2 AMQP核心概念：Connection、Channel与队列

很多刚接触RabbitMQ的人，在写代码的时候容易把Connection和Channel搞混。这里我用个稍微生活化一点的类比来解释：

Connection可以理解成一根物理光纤，是客户端和服务器之间的TCP长连接。Channel则是这根光纤里划分出来的逻辑通路。在RabbitMQ的客户端实践里，Connection是重量级对象，创建和销毁的代价很高，通常整个进程只需要一个。而Channel是轻量级的，每个线程用独立的Channel去发送消息，这样既保证了并发安全，又不会因为频繁创建TCP连接导致性能损耗。

队列（Queue）就是消息存放的地方。生产者把消息发到指定队列，消费者从队列里拉取消息。如果队列不存在，发送时会直接报错，所以我在Sampler的初始化阶段会执行一次queueDeclare，确保队列存在。

还有一个关键参数是prefetch count，这个主要影响消费者场景，我们做生产者压测时暂时用不到，但如果你的压测场景既要生产又要消费，就需要理解了。

3. 两种JMeter实现方案，我为什么选自定义Java Sampler

JMeter对接RabbitMQ，网上能搜到的大概有两种主流做法。我把这两种都跑过一遍，对比之后选了其中一种，下面说说具体原因。

3.1 方案一：JMS Publisher Sampler的局限

JMeter自带一个叫“JMS Publisher”的Sampler，看名字好像可以直接往MQ里发消息。但我深入看了实现之后发现，它主要适配的是标准JMS（Java Message Service）协议，而RabbitMQ原生走的是AMQP协议，两者虽然有兼容层，但用起来有几个明显的坑。

第一，配置复杂。你需要把RabbitMQ的JMS客户端依赖包全部导入JMeter，初始化JMS连接工厂，还要配置ConnectionFactory。我按照网上教程操作，光Classpath的依赖就折腾了大半天。

第二，定制能力受限。JMS Publisher的界面参数就那么几个，消息体内容的构造方式也不够灵活。如果你要在消息里附带特定的headers属性，或者要自定义RoutingKey的生成规则，用这个Sampler就非常别扭。

第三，性能表现不理想。我在压测过程中发现，JMS Publisher创建的连接和会话模型跟RabbitMQ原生的AMQP模型还是有差异，同样的并发条件下，吞吐量上不去，而且容易出现连接不稳定。

3.2 方案二：自定义Java Sampler的完整思路

于是我把重心转向了第二种方案：编写自定义Java Sampler。JMeter提供了一个AbstractJavaSamplerClient抽象类，你只需要继承它，实现runTest方法，JMeter就会在线程组里面执行你定义的具体逻辑。

package com.example.jmeter; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient; import org.apache.jmeter.protocol.java.sampler.JavaSamplerContext; import org.apache.jmeter.samplers.SampleResult; public class RabbitMqProducerSampler extends AbstractJavaSamplerClient { private Connection connection; private Channel channel; private String queueName; private String messageBody; @Override public void setupTest(JavaSamplerContext context) { try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost(context.getParameter("host", "localhost")); factory.setPort(Integer.parseInt(context.getParameter("port", "5672"))); factory.setUsername(context.getParameter("username", "admin")); factory.setPassword(context.getParameter("password", "admin123")); connection = factory.newConnection(); channel = connection.createChannel(); queueName = context.getParameter("queueName", "test.queue"); channel.queueDeclare(queueName, true, false, false, null); messageBody = context.getParameter("messageBody", "Hello RabbitMQ from JMeter!"); } catch (Exception e) { throw new RuntimeException("Failed to setup RabbitMQ connection", e); } } @Override public SampleResult runTest(JavaSamplerContext context) { SampleResult result = new SampleResult(); result.setSampleLabel("RabbitMQ Publish"); result.sampleStart(); try { byte[] body = messageBody.getBytes("UTF-8"); channel.basicPublish("", queueName, null, body); result.setSuccessful(true); result.setResponseData("Message published to " + queueName, "UTF-8"); } catch (Exception e) { result.setSuccessful(false); result.setResponseData(e.getMessage(), "UTF-8"); } finally { result.sampleEnd(); } return result; } @Override public void teardownTest(JavaSamplerContext context) { try { if (channel != null) channel.close(); if (connection != null) connection.close(); } catch (Exception e) { // log warning } } }

这段代码里，setupTest负责建立连接和Channel，runTest是压测线程真正执行的方法，每次发送一条消息，teardownTest在测试结束后清理资源。

挑选这个方案的原因在于，代码完全可控，连接模型也是RabbitMQ官方推荐的最佳实践，压测结果更贴近真实场景。而且参数全部通过JMeter的界面进行配置，后续调整参数非常灵活。

4. 完整实操：从零搭一个RabbitMQ生产者压测计划

方案定了之后，实操就顺理成章了。下面把完整的过程复述一遍，包括代码怎么写、JMeter怎么配置、参数怎么设计，以及压测结果怎么分析。

4.1 编写Sampler代码与依赖导入

首先是工程的搭建。我用Maven创建一个标准的Java项目，在pom.xml里引入两个依赖：

<dependencies> <dependency> <groupId>org.apache.jmeter</groupId> <artifactId>ApacheJMeter_core</artifactId> <version>5.6.3</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.jmeter</groupId> <artifactId>ApacheJMeter_java</artifactId> <version>5.6.3</version> <scope>provided</scope> </dependency> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.20.0</version> </dependency> </dependencies>

注意前两个JMeter相关的依赖scope用provided，因为JMeter运行环境里本来就有这些类库，你只需要在编译时用到。第三个amqp-client是RabbitMQ官方Java客户端，这个必须打包进去，最终构建的时候把依赖带全。

写完代码后执行mvn clean package，然后在JMeter的lib/ext目录下放一个专门放自定义Sampler的文件夹，把打好的jar包以及依赖的amqp-clientjar包都拷贝进去，重启JMeter就能在“Java Request”采样器里看到这个类了。

提示：依赖缺了最常见的报错是ClassNotFoundException: com.rabbitmq.client.ConnectionFactory，所以amqp-client的jar务必带上。

4.2 JMeter测试计划参数设计

Java Request Sampler配置好之后，下一步就是搭建测试计划。线程组是整个压测的核心，它的参数直接决定了压测的压力模型。

首先是线程数。这个非常关键，它代表了同时发送消息的生产者数量。注意，线程数和RabbitMQ的Channel是挂钩的，我在setupTest里面是每个线程各建一个Channel，这个设计是符合RabbitMQ官方推荐的模型。如果线程数设得过高，比如1000个并发，就需要考虑机器本身的连接数和文件句柄限制，之前我遇到过“Too many open files”的报错，调高了操作系统的文件句柄上限才解决。

其次是Ramp-Up时间。它控制了线程启动的缓冲时间，就是要花多长时间把设定的线程全部启动起来。如果设为0，JMeter会瞬间启动所有线程，非常容易打崩连接池或者触发系统的TCP SYN队列溢出。我一般建议设置5-10秒，让连接建立过程有个缓冲。

然后是循环次数。每个线程发送多少条消息，如果你希望无限发送直到手动停止，勾选“永远”就行。如果希望控制总量，就把线程数乘以循环次数估算总消息数。

我自己常用的压测模型是这样的：线程数200、Ramp-Up 10秒、循环次数1000。这样总共会发送20万条消息，足够评估一个中等配置RabbitMQ集群的基本吞吐能力了。

再来是Sampler的参数。在界面上可以看到我之前代码里定义的那些参数：

如果想模拟不同大小的消息怎么办？消息体大小这个参数往往很关键。我在代码里用的是固定字符串，但如果要做更贴近业务的压测，建议改造一下代码，用随机字节数组生成特定大小的消息体，比如1KB、10KB，这样才能测出不同消息体量级下的性能差异。我自己的做法是加了一个messageSize参数，当它大于0时，忽略messageBody，改用随机字节数组。

还需要补充一个经常被忽略的细节——如果在runTest里每次都执行queueDeclare，会白白增加AMQP协议的往返交互开销，严重影响性能。我的做法是只在setupTest里声明队列，发送时直接basicPublish，这个优化让TPS大概提升了15%左右。

4.3 压测执行与结果分析

配置完成后，点击运行按钮，然后打开“聚合报告”监听器观察实时的TPS、响应时间、错误率这几个关键指标。

我这边做的一次实际压测数据是这样：200个线程同时发送，消息体1KB，RabbitMQ部署在4核8G的单节点Docker容器里，最终聚合报告显示平均TPS在8200左右，平均响应时间约230毫秒，p99响应时间约680毫秒，错误率0。整体表现算是不错，但也有值得优化的空间，比如消息持久化开启后TPS会下降，这个在后续调优时需要考虑。

压测过程中，建议同时打开RabbitMQ的Web管理界面，在“Queues”页面监控队列的消息速率（publish rate）和队列堆积情况。如果发现队列的Unacked消息数持续增长，说明消费端处理不过来。如果你的压测只发消息不启动消费者，队列堆积是正常的，但如果消费端存在而堆积仍上涨，就要排查消费逻辑了。

5. 压测过程中的高频问题与排查思路

这部分内容是我在多次压测中最想分享的，因为很多东西不是看文档能看到的，必须自己踩过坑才记得住。

5.1 连接失败与认证问题

最开始跑的时候，最容易碰到的是连接失败。如果你用的是我前面那个Docker启动命令，默认创建的账号是admin/admin123，权限只分配给了默认的vhost/。如果你的JMeter参数里写的host是localhost，但RabbitMQ部署在远程机器上，记得检查防火墙是否开放了5672端口。

另外一个隐蔽的问题是vhost不匹配。如果业务创建了独立的vhost，比如/order，那么ConnectionFactory里面还需要调用factory.setVirtualHost("/order")，否则即使账号密码正确，也会报ACCESS_REFUSED。

5.2 吞吐量上不去，怎么排查

明明线程数已经加到很高了，但TPS就是上不去，这种情况很多见。我总结下来的排查优先级是这样的：

第一，看JMeter机器本身的资源。压测的瓶颈经常不在RabbitMQ，而在发起压测的机器上。打开任务管理器或者top命令，如果CPU已经100%，说明JMeter所在机器已经顶不住了，这时候加线程数没有意义，反而会因为线程上下文切换导致TPS下降。

第二，检查消息体大小。1KB和100KB的消息体对吞吐量的影响完全不是一个量级。小的消息用短字符串代替，往往会让压测结果过于乐观。这里一定要结合实际业务的消息大小来设置。

第三，看RabbitMQ的日志和监控。如果RabbitMQ所在节点的CPU和内存都在合理范围，但TPS还是上不去，考虑网络延迟和带宽限制，特别是跨机房压测的场景，网络IO很容易成为瓶颈。

5.3 Channel关闭与"connection closed"错误

还有一种很常见的情况，就是压测跑到一半，JMeter的日志开始疯狂报channel is already closed或者connection closed unexpectedly，从监控看连接被服务端断开了。

这个问题的根源一般是连接空闲超时。RabbitMQ默认有heartbeat超时机制，如果连接在指定时间内没有AMQP协议帧交互，服务端会主动断开连接。解决办法有两种：一是在代码里调大heartbeat间隔，factory.setRequestedHeartbeat(60)；二是确保压测过程中不要有太长的think time，让JMeter线程保持持续发送的节奏。

另外，如果单次压测的消息总量非常大，且一次性创建了太多Channel，也有可能被RabbitMQ的连接数限制给拦截了。可以检查服务端日志，如果出现connection limit reached之类的信息，就需要调整RabbitMQ的channel_max参数或者适当降低线程数。

5.4 消息落盘与持久化配置的影响

还有一类问题容易被忽略：RabbitMQ的消息持久化对吞吐量的影响远比想象中更大。如果发送消息的时候指定了MessageProperties.PERSISTENT_TEXT_PLAIN，每条消息都会等待磁盘fsync确认后才返回，吞吐量会明显下降。

我做了一组对比测试：非持久化消息TPS约8300，持久化消息TPS掉到3200左右，损失超过60%。所以压测之前要想清楚你的业务是否真的需要持久化，如果不需要，生产者和队列都不要开启持久化，否则你的压测指标会误导容量评估。

5.5 快速问题参考表

为了方便排查，我把自己踩过的坑整理成了一张速查表：

6. 我在实际压测中的一些体会

最后再分享一些跟脚本本身无关，但对压测结果影响很深的心得。

一个是压测环境的隔离。RabbitMQ的压测必须尽量在干净的环境里做，不要和研发共用一个集群，更不要在生产环境直接压。因为压测产生的消息量是平时业务量的数倍甚至数十倍，很可能把共享集群打挂，影响线上业务。我一般是单独用Docker起一个专用容器，压完直接销毁。

另一个是压测指标一定要结合业务来解读。TPS高不代表系统质量好，还要关注消息端到端的延迟，也就是消息从生产者发出到消费者最终消费的间隔。我之前遇到过生产端TPS很好，但消费者处理逻辑里面有慢SQL，导致消息大量积压，业务方感受到的延迟暴涨。这种问题，光看JMeter的聚合报告是发现不了的，必须结合RabbitMQ的队列监控和业务侧日志一起看。

还有一个小技巧，JMeter压测结果的CSV文件一定要留下来。后续如果要输出正式的性能测试报告，或者跟新一轮压测做对比，这些原始数据是最有力的证据。我习惯在每次压测跑完后，把线程数、消息大小、TPS、p99响应时间这些关键数据记到一张表格里，时间长了之后，就能摸清自己这套系统在不同压力模型下的性能边界，这才是压测真正有价值的产出。

这个方案做完之后，我把它打包成了那个apache-jmeter-rabbitMQ测试.zip，里面的自定义Sampler、JMeter测试计划和说明文档都整理好了。如果你也需要对RabbitMQ做类似的生产者压测，直接照着我这个思路操作，应该能在一天之内跑出第一份有效数据。

本文还有配套的精品资源，点击获取