RIP 22 RocketMQ Stage Message
Status
* Current State: Proposed
* Authors: [email protected]
* Shepherds: [email protected]
* Mailing List discussion: [email protected]
* Pull Request: #PR_NUMBER
* Released: 4.9.0
Background & Motivation
What do we need to do
* Will we add a new module?
No
* Will we add new APIs?
Yes
* Will we add new feature?
Yes
Why should we do that
* What can we benefit proposed changes?
EN
Let me give an example to illustrate the waste of performance caused by
`MessageListenerOrderly`. For example, in the e-commerce scenario, we use
`MessageListenerOrderly` to consume the order MQ. The first 100 consumed MQ
(the "consumed MQ" hereinafter referred to as "MQ") can get different extra
rewards, while the MQ after 100 can get no extra rewards. At this point,
`MessageListenerOrderly` can guarantee the order of the first 100 MQS and the
order of the after 100 MQS, but it is meaningless to guarantee the order of the
after 100 MQS. Therefore, for MQ after 100, we should use
`MessageListenerConcurrently` to consume concurrently, so as to improve the
performance.
Furthermore, suppose that the first 10 MQS can get an additional laptop
(called "stage 1"), the first 10-30 MQS can get an additional tablet (called
"stage 2"), the first 30-100 MQS can get an additional mobile phone (called
"stage 3"), and the after 100 MQS are called "stage 4". Because the reward of
each stage is the same, we can use `MessageListenerConcurrently` to consume
concurrently in the stages (for example, 20 MQ concurrent consumptions in stage
2). We only need to ensure the order between stages (only when all MQS in stage
1 is consumed can MQS in stage 2 be consumed).
Hard to understand? It doesn't matter. Let's give a more specific
example. Suppose that consuming one MQ takes 1s, single read queue and thread
pool size of `MessageListenerConcurrently` is 20, then for this example,
consuming 100 MQ using `MessageListenerOrderly` takes `1 * 100 = 100s` The time
required to consume 100 MQS by using `MessageListenerConcurrently` periodically
is as follows:
Stage 1 (1-10): 10 / 20 ?? 1s
Stage 2 (11-30): 20 / 20 ?? 1s
Stage 3 (31-100): 70 / 20 ?? 4S
Total time: 1 + 1 + 4 = 6S
Compared with 100s, it is ten times faster!
To sum up, I propose `MessageListenerStagedConcurrently`, which
can ensure the sequence of phases and concurrent consumption within phases.
Compared with `MessageListenerOrderly`, its performance is greatly improved;
Compared with `MessageListenerConcurrently`, it ensures the necessary order. Of
course, this idea of `staged concurrency` can also be used in producer, brush
disk and other places, and please free your imagination.
CN
????????????????????????`MessageListenerOrderly`??????????????????????????????????????`MessageListenerOrderly`??????????MQ????100??????????MQ(??????????MQ????????????MQ??)????????????????????????100????????MQ??????????????????`MessageListenerOrderly`??????????100??MQ????????????????100????????MQ????????????????100????????MQ??????????????????????????????100????????MQ????????`MessageListenerConcurrently`??????????????????????????
????????????????????10??MQ??????????????????????????(??????????1??)????10-30??MQ????????????????????????(??????????2??)????30-100??MQ????????????????????(??????????3??)??100????????MQ??????????4??????????????????????????????????????????????????`MessageListenerConcurrently`????????????(????????2??20??MQ????????)????????????????????????????????(????1??MQ????????????????2??MQ????????????)??
?????????????????????????????????????????????????1??MQ????????1s????????????`MessageListenerConcurrently`????????????????20????????????????????????????`MessageListenerOrderly`????100??MQ????????`1*100=100s`??????????????????`MessageListenerConcurrently`????100??MQ????????????????
????1(1-10)??10 / 20 ?? 1s
????2(11-30)??20 / 20 ?? 1s
????3(31-100)??70 / 20 ?? 4s
????????1 + 1 + 4 = 6s
????100s????????????????
????????????????`MessageListenerStagedConcurrently`??????????????????????????????????????????????????????`MessageListenerOrderly`????????????????????????????`MessageListenerConcurrently`??????????????????????????????????`??????????`??????????????????producer??????????????????????????????????????????????
Goals
EN: Greatly improve the performance of order-message.
CN: ??????????????????????????
Non-Goals
* What problem is this proposal NOT designed to solve?
EN: Not change the original order-message interface.
CN: ????????????????????????????????
* Are there any limits of this proposal?
EN
1.In order to maximize the performance of order-message, users need to
define stages reasonably.
2.Just like `MessageListenerOrderly`, the
`MessageListenerStagedConcurrently` only ensures that each queue (partition) is
orderly.
CN
1.??????????????????????????????????????????????????????????
2.??`MessageListenerOrderly`??????`MessageListenerStagedConcurrently`????????????????????????????????
Changes
We need add some codes in common, client and broker component
which include adding priority concurrency frameworks and consumer choices.Read
below sections to get more details about the Stage Message for RocketMQ.
Interface Design/Change
```java
public interface MessageListenerStagedConcurrently extends MessageListener {
ConsumeOrderlyStatus consumeMessage(final
List<MessageExt> msgs,
final
ConsumeStagedConcurrentlyContext context);
/**
* If returns empty collection, {@link
MessageListenerStagedConcurrently} will degenerate into {@link
* MessageListenerConcurrently}; If returns a
collection whose elements are all 1, {@link
* MessageListenerStagedConcurrently} will temporarily
evolve into {@link MessageListenerOrderly};
*/
List<Integer> getStageDefinitions();
/**
* can be used to reset the current stage by CAS
*/
void resetCurrentStageOffsetIfNeed(final String topic, final
AtomicInteger currentStageOffset);
}
```
```java
/**
* Refer to {@link org.apache.rocketmq.client.consumer.store.OffsetStore},
manage the stage consumption progress.
*/
public interface StageOffsetStore {
void load() throws MQClientException;
void updateStageOffset(final MessageQueue mq, final int
stageOffset, final boolean increaseOnly);
int readStageOffset(final MessageQueue mq, final
ReadOffsetType type);
void persistAll(final Set<MessageQueue> mqs);
void persist(final MessageQueue mq);
void removeStageOffset(MessageQueue mq);
Map<MessageQueue, Integer> cloneStageOffsetTable(String
topic);
void updateConsumeStageOffsetToBroker(MessageQueue mq, int
stageOffset, boolean isOneway) throws RemotingException,
MQBrokerException,
InterruptedException, MQClientException;
}
```
* Method behavior changes
Nothing specific.
* CLI command changes
Nothing specific.
* Log format or content changes
Nothing specific.
Compatibility, Deprecation, and Migration Plan
* Are backward and forward compatibility taken into consideration?
EN: Not change the original order-message interface.
CN: ??????????????????????????????????
* Are there deprecated APIs?
Nothing specific.
* How do we do migration?
Nothing specific.
Implementation Outline
We will implement the proposed changes by 3 phases.
Phase 1
Implement Stage Message feature in Consumer
Phase 2
Implement Stage Message feature in Producer(To be honest, I
haven't thought about it yet. I'm looking forward to your idea, ha ha)
Phase 3
Implement Stage Message feature in Broker(Looking forward to your
idea, too)
Rejected Alternatives
* How does alternatives solve the issue you proposed?
Nothing specific.
* Pros and Cons of alternatives
Nothing specific.
* Why should we reject above alternatives
