Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Polish example of simple consumer #351

Merged
merged 1 commit into from
Feb 6, 2023
Merged

Polish example of simple consumer #351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
60 changes: 31 additions & 29 deletions ...ent/src/main/java/org/apache/rocketmq/client/java/example/AsyncSimpleConsumerExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package org.apache.rocketmq.client.java.example;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
Expand Down Expand Up @@ -45,7 +44,8 @@ public class AsyncSimpleConsumerExample {
private AsyncSimpleConsumerExample() {
}

public static void main(String[] args) throws ClientException, IOException, InterruptedException {
@SuppressWarnings({"resource", "InfiniteLoopStatement"})
public static void main(String[] args) throws ClientException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();

// Credential provider is optional for client configuration.
Expand Down Expand Up @@ -81,34 +81,36 @@ public static void main(String[] args) throws ClientException, IOException, Inte
ExecutorService receiveCallbackExecutor = Executors.newCachedThreadPool();
// Set individual thread pool for ack callback.
ExecutorService ackCallbackExecutor = Executors.newCachedThreadPool();
final CompletableFuture<List<MessageView>> future0 = consumer.receiveAsync(maxMessageNum, invisibleDuration);
future0.whenCompleteAsync(((messages, throwable) -> {
if (null != throwable) {
log.error("Failed to receive message from remote", throwable);
// Return early.
return;
}
log.info("Received {} message(s)", messages.size());
// Using messageView as key rather than message id because message id may be duplicated.
final Map<MessageView, CompletableFuture<Void>> map =
messages.stream().collect(Collectors.toMap(message -> message, consumer::ackAsync));
for (Map.Entry<MessageView, CompletableFuture<Void>> entry : map.entrySet()) {
final MessageId messageId = entry.getKey().getMessageId();
final CompletableFuture<Void> future = entry.getValue();
future.whenCompleteAsync((v, t) -> {
if (null != t) {
log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
// Return early.
return;
}
log.info("Message is acknowledged successfully, messageId={}", messageId);
}, ackCallbackExecutor);
}
// Receive message.
do {
final CompletableFuture<List<MessageView>> future0 = consumer.receiveAsync(maxMessageNum,
invisibleDuration);
future0.whenCompleteAsync(((messages, throwable) -> {
if (null != throwable) {
log.error("Failed to receive message from remote", throwable);
// Return early.
return;
}
log.info("Received {} message(s)", messages.size());
// Using messageView as key rather than message id because message id may be duplicated.
final Map<MessageView, CompletableFuture<Void>> map =
messages.stream().collect(Collectors.toMap(message -> message, consumer::ackAsync));
for (Map.Entry<MessageView, CompletableFuture<Void>> entry : map.entrySet()) {
final MessageId messageId = entry.getKey().getMessageId();
final CompletableFuture<Void> future = entry.getValue();
future.whenCompleteAsync((v, t) -> {
if (null != t) {
log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
// Return early.
return;
}
log.info("Message is acknowledged successfully, messageId={}", messageId);
}, ackCallbackExecutor);
}

}), receiveCallbackExecutor);
// Block to avoid exist of background threads.
Thread.sleep(Long.MAX_VALUE);
}), receiveCallbackExecutor);
} while (true);
// Close the simple consumer when you don't need it anymore.
consumer.close();
// consumer.close();
}
}
29 changes: 16 additions & 13 deletions java/client/src/main/java/org/apache/rocketmq/client/java/example/SimpleConsumerExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package org.apache.rocketmq.client.java.example;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
Expand All @@ -40,7 +39,8 @@ public class SimpleConsumerExample {
private SimpleConsumerExample() {
}

public static void main(String[] args) throws ClientException, IOException {
@SuppressWarnings({"resource", "InfiniteLoopStatement"})
public static void main(String[] args) throws ClientException {
final ClientServiceProvider provider = ClientServiceProvider.loadService();

// Credential provider is optional for client configuration.
Expand Down Expand Up @@ -72,18 +72,21 @@ public static void main(String[] args) throws ClientException, IOException {
int maxMessageNum = 16;
// Set message invisible duration after it is received.
Duration invisibleDuration = Duration.ofSeconds(15);
final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
log.info("Received {} message(s)", messages.size());
for (MessageView message : messages) {
final MessageId messageId = message.getMessageId();
try {
consumer.ack(message);
log.info("Message is acknowledged successfully, messageId={}", messageId);
} catch (Throwable t) {
log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
// Receive message, multi-threading is more recommended.
do {
final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
log.info("Received {} message(s)", messages.size());
for (MessageView message : messages) {
final MessageId messageId = message.getMessageId();
try {
consumer.ack(message);
log.info("Message is acknowledged successfully, messageId={}", messageId);
} catch (Throwable t) {
log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
}
}
}
} while (true);
// Close the simple consumer when you don't need it anymore.
consumer.close();
// consumer.close();
}
}