dragon-zhang removed a comment on pull request #2895:
URL: https://github.com/apache/rocketmq/pull/2895#issuecomment-847577693
Thank you for your question. I'm very happy that you understand the meaning
of this PR.
About Q1, it depends on the user's own definition, and is free to define
which stages and how large they are, for example in
`org.apache.rocketmq.client.impl.consumer.ConsumeMessagePeriodicConcurrentlyServiceTest#test05MessageListenerOrderlyToConcurrently`,
I've defined 11 stages with sizes of 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 and
infinity.
About Q2, I wrote a simple example to answer your question:
```
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
public class Example {
public static class PriorityConcurrentEngine {
/**
* highest priority
*/
public static final Integer
MAX_PRIORITY = Integer.MIN_VALUE;
/**
* lowest priority
*/
public static final Integer
MIN_PRIORITY = Integer.MAX_VALUE;
private final ExecutorService
executor = Executors.newFixedThreadPool(32);
private final AtomicInteger
currentStage = new AtomicInteger(0);
/**
* You can also use other
class here, such as ***@***.*** ConcurrentNavigableMap<Integer,
Queue<Callable>>}
*/
private final
ConcurrentNavigableMap<Integer, Queue<Runnable>> priorityTasks = new
ConcurrentSkipListMap<>();
private volatile boolean run =
true;
private final Thread consumer =
new Thread(() -> {
while
(run) {
invokeAllNow();
}
});
public static <E> List<E>
pollAllTask(Queue<E> tasks) {
List<E> list = new ArrayList<>();
while
(tasks != null && !tasks.isEmpty()) {
E task = tasks.poll();
list.add(task);
}
return
list;
}
public void start() {
consumer.start();
}
public void shutdown() throws
InterruptedException {
run =
false;
executor.shutdown();
}
public int
getAndAddCurrentStage(int delta) {
return
currentStage.getAndAdd(delta);
}
public final void
runPriorityAsync(Integer priority, Runnable... tasks) {
if (null
== tasks || tasks.length == 0) {
return;
}
Queue<Runnable> queue = priorityTasks.putIfAbsent(priority, new
ConcurrentLinkedQueue<>());
if (null
== queue) {
queue = priorityTasks.get(priority);
}
for
(Runnable runnable : tasks) {
queue.offer(runnable);
}
}
public final void
runPriorityAsync(Integer priority, Collection<Runnable> tasks) {
if (null
== tasks || tasks.isEmpty()) {
return;
}
Queue<Runnable> queue = priorityTasks.putIfAbsent(priority, new
ConcurrentLinkedQueue<>());
if (null
== queue) {
queue = priorityTasks.get(priority);
}
for
(Runnable runnable : tasks) {
queue.offer(runnable);
}
}
public final void
runAsync(Queue<Runnable> tasks) {
runAsync(pollAllTask(tasks));
}
public final void
runAsync(Collection<Runnable> tasks) {
if (null
== tasks || tasks.isEmpty()) {
return;
}
List<CompletableFuture<Void>> list = new ArrayList<>(tasks.size());
for
(Runnable task : tasks) {
// Other thread pools can also be used here
list.add(CompletableFuture.runAsync(task, executor));
}
executeAsync(list);
}
public final <T> List<T>
executeAsync(Collection<CompletableFuture<T>> tasks) {
if (null
== tasks || tasks.isEmpty()) {
return new ArrayList<>();
}
try {
CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0])).join();
} catch
(Exception e) {
e.printStackTrace();
}
return
getResultIgnoreException(tasks);
}
public final <T> List<T>
getResultIgnoreException(Collection<CompletableFuture<T>> tasks) {
List<T> result = new ArrayList<>(tasks.size());
for
(CompletableFuture<T> completableFuture : tasks) {
if (null == completableFuture) {
continue;
}
try {
T response = completableFuture.get();
if (null != response) {
result.add(response);
}
} catch (Exception e) {
e.printStackTrace();
}
}
return
result;
}
public void invokeAllNow() {
synchronized (priorityTasks) {
// This ensures that tasks with high priority are executed first,
// and does not guarantee the execution order of tasks with the same priority
for (Queue<Runnable> queue : priorityTasks.values()) {
Queue<Runnable> runnableQueue = new ConcurrentLinkedQueue<>();
while (!queue.isEmpty()) {
Runnable element = queue.poll();
if (element != null) {
runnableQueue.offer(element);
}
}
runAsync(runnableQueue);
}
}
}
}
public static void main(String[] args) throws Exception {
PriorityConcurrentEngine engine =
new PriorityConcurrentEngine();
List<Integer> stageDefinitions
= new ArrayList<>();
for (int i = 1; i <= 10; i++) {
int sum =
0;
for (int
j = 1; j <= i; j++) {
sum = sum + j;
}
stageDefinitions.add(sum);
}
engine.start();
for (int i = 0; i < 100; i++) {
int
currentStage = engine.getAndAddCurrentStage(1);
int index
= -1;
for (int
j = 0; j < stageDefinitions.size(); j++) {
if (currentStage < stageDefinitions.get(j)) {
index = j;
break;
}
}
int
finalIndex = index;
if (index
== -1) {
engine.runPriorityAsync(PriorityConcurrentEngine.MIN_PRIORITY, () -> {
try {
//simulate business call
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(finalIndex);
});
} else {
engine.runPriorityAsync(index, () -> {
try {
//simulate business call
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(finalIndex);
});
}
}
//
1/32=1,2/32=1,3/32=1,4/32=1,5/32=1,6/32=1,7/32=1,8/32=1,9/32=1,10/32=1
// 45/32=2
//
MessageListenerPeriodicConcurrently takes about (1*10+2)*100=1200ms
Thread.sleep(1300);
engine.shutdown();
for (int i = 0; i < 100; i++) {
Thread.sleep(100);
System.out.println(i);
}
// MessageListenerOrderly takes
about 100*100=10000ms
}
}
```
------------------ 原始邮件 ------------------
发件人:
"apache/rocketmq"
***@***.***>;
发送时间: 2021年5月25日(星期二) 上午10:18
***@***.***>;
***@***.******@***.***>;
主题: Re: [apache/rocketmq] Stage Ordered and disordered comsumption
support? (#2895)
it's a very nice PR to improve order-message. and i have 2 questions here:
Q1. how to define the first period and why ?
Q2. using order-message means need comsume message orderly, and orderly do
NEXT-STEP withconsume-results, like CRUD in mysql. in one period, how to make
sure NEXT-STEP orderly?
—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub, or unsubscribe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
