目录
一、问题描述
1.压测日志
2.客户端日志
二、解决发送失败情况
三、解决发送TPS过低情况
四、原因分析
1.刷盘流程回顾
2.主从复制回顾
3.流程模拟
4.原因总结
一、问题描述
早些时候写过性能测试和性能优化文章，主要基于异步刷盘/异步复制；由于业务需要需要搭建异步刷盘/同步复制集群；同时对性能进行压测。
压测结果显示集群几乎无法使用，TPS居然是个位数，客户端也在报错。
1.压测日志
2.客户端日志
2019-09-19 19:22:38,038 ERROR RocketmqClient - [BENCHMARK_PRODUCER] Send Exception
org.apache.rocketmq.client.exception.MQBrokerException: CODE: 2 DESC: [TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: 209ms, size of queue: 9
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
at org.apache.rocketmq.client.impl.MQClientAPIImpl.processSendResponse(MQClientAPIImpl.java:671)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageSync(MQClientAPIImpl.java:467)
at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:449)
at
二、解决发送失败情况
经排查，将transientStorePoolEnable关闭(默认为false)；压测显示最高TPS有1.9万。
brokerRole=SYNC_MASTER
#transientStorePoolEnable=true
三、解决发送TPS过低情况
最高TPS只有1.9万，依然过低，与预期相差甚远，我们预期压测应该可以到7到8万这样可以满足业务发展需要。再次检查broker端参数配置，没有发现有参数导致性能如此过低。
回顾性能调优的几个方面：系统调优、集群调优、JVM调优。
系统调优与集群调优都已经做过了，唯一没有优化的JVM调优，堆内存设置默认的8G。
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
将JVM堆内存提高4倍后，压测效果明显提升，基本可以达到预期7万多的TPS。
四、原因分析
1.为什么在异步刷盘/同步复制时开启堆外内存池transientStorePoolEnable后，集群压测几乎无法进行？
2.为什么在异步刷盘/同步复制时调大JVM堆内存后，性能明显提升呢？提升了的倍数几乎是堆内存增大的倍数。
1.刷盘流程回顾
消息追加回顾。参见：https://www.jianshu.com/p/5aa591a72fb4
刷盘回顾。参见：https://www.jianshu.com/p/62f876fda84c
异步刷盘未开启堆外缓存示意图
异步刷盘开启堆外缓存示意图
小结：异步刷盘未开启transientStorePoolEnable时，消息追加到mappedByteBuffer中，异步线程刷调用mappedByteBuffer.force落盘；异步刷盘开启transientStorePoolEnable时，消息写入wrtieBuffer中，异步线程将消息提交到fileChannel，然后异步线程调用fileChannel.force落盘。
2.主从复制回顾
主从复制参见：https://www.jianshu.com/p/8fd0cf5f2ce7
负责向Slave写入数据：HAConnection#WriteSocketService
//查找待拉取偏移量之后所有的可读消息
SelectMappedBufferResult selectResult = HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
// ...
SelectMappedBufferResult result = mappedFile.selectMappedBuffer(pos);
// ...
ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
byteBuffer.position(pos);
int size = readPosition - pos; //计算距离最大可读位置的大小
ByteBuffer byteBufferNew = byteBuffer.slice();
byteBufferNew.limit(size);
return new SelectMappedBufferResult()
小结：主从复制使用mappedByteBuffer向Slave同步数据。
3.流程模拟
开启堆外内存池流程
@Test
public void test01(){
// 堆外内存池transientStorePoolEnable开启后，消息追加操作
try {
File file = new File("/Users/yongliang/logs/temp.log");
FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
String data = "beautiful girl!";
// mmap 文件映射操作
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, data.length());
// 堆外内存transientStorePoolEnable开启
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(data.length());
// ----------------消息追加开始-----------------------
// 注意此时使用堆内内存分配
ByteBuffer msgStoreItemMemory = ByteBuffer.allocate(data.length());
msgStoreItemMemory.put(data.getBytes());
// 开启transientStorePoolEnable消息写入了ByteBuffer
byteBuffer.put(msgStoreItemMemory.array(),0,data.length());
// ----------------消息追加结束-----------------------
// ----------------消息提交开始-----------------------
byteBuffer.position(0);
byteBuffer.limit(data.length());
fileChannel.write(byteBuffer);
// ----------------消息提交结束-----------------------
// --------主从复制从mappedByteBuffer获取消息开始----------
mappedByteBuffer.position(0);
mappedByteBuffer.limit(data.length());
Charset charset = Charset.forName("UTF-8");
CharsetDecoder decoder = charset.newDecoder();
CharBuffer charBuffer = decoder.decode(mappedByteBuffer.asReadOnlyBuffer());
System.out.println(charBuffer.toString());
// --------主从复制从mappedByteBuffer获取消息结束----------
} catch (Exception e) {
e.printStackTrace();
}
}
小结：模拟开启堆外内存池transientStorePoolEnable的消息追加及主从复制流程。
未开启堆外内存池流程
@Test
public void test02(){
try {
File file = new File("/Users/yongliang/logs/temp1.log");
FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
String data = "beautiful girl!";
// mmap 文件映射操作
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, data.length());
// ----------------消息追加开始-----------------------
// 注意消息组装使用堆内内存分配
ByteBuffer msgStoreItemMemory = ByteBuffer.allocate(data.length());
msgStoreItemMemory.put(data.getBytes());
mappedByteBuffer.put(msgStoreItemMemory.array(),0,data.length());
// ----------------消息追加结束-----------------------
// --------主从复制从mappedByteBuffer获取消息开始----------
mappedByteBuffer.position(0);
mappedByteBuffer.limit(data.length());
Charset charset = Charset.forName("UTF-8");
CharsetDecoder decoder = charset.newDecoder();
CharBuffer charBuffer = decoder.decode(mappedByteBuffer.asReadOnlyBuffer());
System.out.println(charBuffer.toString());
// --------主从复制从mappedByteBuffer获取消息结束----------
} catch (Exception e) {
e.printStackTrace();
}
}
小结：模拟未开启堆外内存池transientStorePoolEnable的消息追加及主从复制流程。
4、原因总结
1.为什么在异步刷盘/同步复制时开启堆外内存transientStorePoolEnable后，集群压测几乎无法进行？
解释：
1>主从同步复制使用mappedByteBuffer；
2>开启堆外内存池transientStorePoolEnable后数据先落到WriteBuffer，再通过异步提交线程提交到FileChannel，再通过mmap将数据映射到mappedByteBuffer；
3>未开启堆外内存池transientStorePoolEnable数据直接写入到mappedByteBuffe；
由于开启堆外内存数据映射到mappedByteBuffer比直接写入mappedByteBuffer多了很多步骤，再加上发送队列处理事件默认只有200毫秒（waitTimeMillsInSendQueue=200），造成集群不能正常压测的原因。
2.为什么在异步刷盘/同步复制时调大JVM堆内存后，性能明显提升呢？提升了的倍数几乎是对内存增大的倍数。
解释：
从模拟流程中可以看出，在组装消息时使用堆内存，提高堆内存显著提高写入Tps的原因所在。
// 注意消息组装使用堆内内存分配
ByteBuffer msgStoreItemMemory = ByteBuffer.allocate(data.length());
作者老梁，哈啰出行高级技术专家，参与了《RocketMQ技术内幕》审稿工作。专注后端中间件方向，已陆续发表RocketMQ系列、Kafka系列、gRPC系列、Sentinel系列、Java NIO系列。其中RocketMQ系列已发表40余篇。源码、实战、原理、调优期待与你一起学习。
