[
https://issues.apache.org/jira/browse/ROCKETMQ-80?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15875403#comment-15875403
]
ASF GitHub Bot commented on ROCKETMQ-80:
----------------------------------------
Github user vongosling commented on a diff in the pull request:
https://github.com/apache/incubator-rocketmq/pull/53#discussion_r101981097
--- Diff:
broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
---
@@ -442,6 +449,202 @@ private RemotingCommand sendMessage(final
ChannelHandlerContext ctx, //
return response;
}
+ private RemotingCommand sendBatchMessage(final ChannelHandlerContext
ctx, //
+ final RemotingCommand request, //
+ final SendMessageContext
sendMessageContext, //
+ final SendMessageRequestHeader
requestHeader) throws RemotingCommandException {
+
+ final RemotingCommand response =
RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
+ final SendMessageResponseHeader responseHeader =
(SendMessageResponseHeader) response.readCustomHeader();
+
+
+ response.setOpaque(request.getOpaque());
+
+ response.addExtField(MessageConst.PROPERTY_MSG_REGION,
this.brokerController.getBrokerConfig().getRegionId());
+ response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH,
String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
+
+ if (log.isDebugEnabled()) {
+ log.debug("receive SendMessage request command, " + request);
+ }
+
+ final long startTimstamp =
this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
+ if (this.brokerController.getMessageStore().now() < startTimstamp)
{
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setRemark(String.format("broker unable to service,
until %s", UtilAll.timeMillisToHumanString2(startTimstamp)));
+ return response;
+ }
+
+ response.setCode(-1);
+ super.msgCheck(ctx, requestHeader, response);
--- End diff --
Why not put msgCheck this precondition method into the first row of the
outer method
> Add batch feature
> -----------------
>
> Key: ROCKETMQ-80
> URL: https://issues.apache.org/jira/browse/ROCKETMQ-80
> Project: Apache RocketMQ
> Issue Type: New Feature
> Affects Versions: 4.1.0-incubating
> Reporter: dongeforever
> Assignee: dongeforever
> Fix For: 4.1.0-incubating
>
>
> Tests show that Kafka's million-level TPS is mainly owed to batch. When set
> batch size to 1, the TPS is reduced an order of magnitude. So I try to add
> this feature to RocketMQ.
> For a minimal effort, it works as follows:
> Only add synchronous send functions to MQProducer interface, just like
> send(final Collection msgs).
> Use MessageBatch which extends Message and implements Iterable<Message>.
> Use byte buffer instead of list of objects to avoid too much GC in Broker.
> Split the decode and encode logic from lockForPutMessage to avoid too many
> race conditions.
> Tests:
> On linux with 24 Core 48G Ram and SSD, using 50 threads to send 50Byte(body)
> message in batch size 50, we get about 150w TPS until the disk is full.
> Potential problems:
> Although the messages can be accumulated in the Broker very quickly, it need
> time to dispatch to the consume queue, which is much slower than accepting
> messages. So the messages may not be able to be consumed immediately.
> We may need to refactor the ReputMessageService to solve this problem.
> And if guys have some ideas, please let me know or just share it in this
> issue.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
