This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch new-official-website
in repository https://gitbox.apache.org/repos/asf/rocketmq-site.git
The following commit(s) were added to refs/heads/new-official-website by this
push:
new e405b20a0 Polish docs
new 3ccdb9205 Merge pull request #504 from aaron-ai/polish_docs
e405b20a0 is described below
commit e405b20a0ba661493329c920b6bfcff1bb4b371d
Author: Aaron Ai <[email protected]>
AuthorDate: Thu Feb 2 11:24:40 2023 +0800
Polish docs
---
.../version-5.0/02-quickStart/01quickstart.md | 164 ++++++++---------
.../04-featureBehavior/06consumertype.md | 92 +++++-----
.../version-5.0/02-quickStart/01quickstart.md | 200 +++++++++++----------
.../04-featureBehavior/06consumertype.md | 93 +++++-----
4 files changed, 275 insertions(+), 274 deletions(-)
diff --git
a/i18n/en/docusaurus-plugin-content-docs/version-5.0/02-quickStart/01quickstart.md
b/i18n/en/docusaurus-plugin-content-docs/version-5.0/02-quickStart/01quickstart.md
index 0e98ca6fe..853e0aedd 100644
---
a/i18n/en/docusaurus-plugin-content-docs/version-5.0/02-quickStart/01quickstart.md
+++
b/i18n/en/docusaurus-plugin-content-docs/version-5.0/02-quickStart/01quickstart.md
@@ -108,91 +108,93 @@ We can also try to use the client sdk to send and receive
messages, you can see
4. In the Java project you have created, create a program that sends messages
and run it with the following code:
- ```java
- import org.apache.rocketmq.client.apis.*;
- import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
- import org.apache.rocketmq.client.apis.consumer.MessageListener;
- import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
- import org.apache.rocketmq.client.apis.message.Message;
- import org.apache.rocketmq.client.apis.message.MessageBuilder;
- import org.apache.rocketmq.client.apis.message.MessageView;
- import org.apache.rocketmq.client.apis.producer.Producer;
- import org.apache.rocketmq.client.apis.producer.SendReceipt;
- import java.time.Duration;
- import java.util.List;
- public class ProducerExample {
- public static void main(String[] args) throws ClientException {
- String endpoint = "localhost:8081";
- String topic = "Your Topic";
- ClientServiceProvider provider =
ClientServiceProvider.loadService();
- ClientConfigurationBuilder builder =
ClientConfiguration.newBuilder().setEndpoints(endpoint);
- ClientConfiguration configuration = builder.build();
- Producer producer = provider.newProducerBuilder()
- .setTopics(topic)
- .setClientConfiguration(configuration)
- .build();
- Message message = provider.newMessageBuilder()
- .setTopic(topic)
- .setKeys("messageKey")
- .setTag("messageTag")
- .setBody("messageBody".getBytes())
- .build();
- try {
- SendReceipt sendReceipt = producer.send(message);
- System.out.println(sendReceipt.getMessageId());
- } catch (ClientException e) {
- e.printStackTrace();
- }
- }
- }
+ ```java
+ import java.io.IOException;
+ import org.apache.rocketmq.client.apis.ClientConfiguration;
+ import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
+ import org.apache.rocketmq.client.apis.ClientException;
+ import org.apache.rocketmq.client.apis.ClientServiceProvider;
+ import org.apache.rocketmq.client.apis.message.Message;
+ import org.apache.rocketmq.client.apis.producer.Producer;
+ import org.apache.rocketmq.client.apis.producer.SendReceipt;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ public class ProducerExample {
+ private static final Logger logger =
LoggerFactory.getLogger(ProducerExample.class);
+
+ public static void main(String[] args) throws ClientException,
IOException {
+ String endpoint = "localhost:8081";
+ String topic = "YourTopic";
+ ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ ClientConfigurationBuilder builder =
ClientConfiguration.newBuilder().setEndpoints(endpoint);
+ ClientConfiguration configuration = builder.build();
+ Producer producer = provider.newProducerBuilder()
+ .setTopics(topic)
+ .setClientConfiguration(configuration)
+ .build();
+ Message message = provider.newMessageBuilder()
+ .setTopic(topic)
+ .setKeys("messageKey")
+ .setTag("messageTag")
+ .setBody("messageBody".getBytes())
+ .build();
+ try {
+ SendReceipt sendReceipt = producer.send(message);
+ logger.info("Send message successfully, messageId={}",
sendReceipt.getMessageId());
+ } catch (ClientException e) {
+ logger.error("Failed to send message", e);
+ }
+ // producer.close();
+ }
+ }
```
-
5. In the Java project you have created, create a consumer demo program and
run it. Apache RocketMQ support
[SimpleConsumer](../04-featureBehavior/06consumertype.md) and
[PushConsumer](../04-featureBehavior/06consumertype.md).
- ```java
- import org.apache.rocketmq.client.apis.*;
- import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
- import org.apache.rocketmq.client.apis.consumer.FilterExpression;
- import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
- import org.apache.rocketmq.client.apis.consumer.PushConsumer;
-
- import java.io.IOException;
- import java.util.Collections;
- import org.slf4j.Logger;
- import org.slf4j.LoggerFactory;
-
- public class PushConsumerExample {
- private static final Logger LOGGER =
LoggerFactory.getLogger(PushConsumerExample.class);
-
- private PushConsumerExample() {
- }
-
- public static void main(String[] args) throws ClientException,
IOException, InterruptedException {
- final ClientServiceProvider provider =
ClientServiceProvider.loadService();
- String endpoints = "localhost:8081";
- ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
- .setEndpoints(endpoints)
- .build();
- String tag = "*";
- FilterExpression filterExpression = new FilterExpression(tag,
FilterExpressionType.TAG);
- String consumerGroup = "Your ConsumerGroup";
- String topic = "Your Topic";
- PushConsumer pushConsumer = provider.newPushConsumerBuilder()
- .setClientConfiguration(clientConfiguration)
- .setConsumerGroup(consumerGroup)
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- .setMessageListener(messageView -> {
- // LOGGER.info("Consume message={}", messageView);
- System.out.println("Consume message!!");
- return ConsumeResult.SUCCESS;
- })
- .build();
- Thread.sleep(Long.MAX_VALUE);
- //pushConsumer.close();
- }
- }
- ```
+ ```java
+ import java.io.IOException;
+ import java.util.Collections;
+ import org.apache.rocketmq.client.apis.ClientConfiguration;
+ import org.apache.rocketmq.client.apis.ClientException;
+ import org.apache.rocketmq.client.apis.ClientServiceProvider;
+ import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
+ import org.apache.rocketmq.client.apis.consumer.FilterExpression;
+ import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
+ import org.apache.rocketmq.client.apis.consumer.PushConsumer;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ public class PushConsumerExample {
+ private static final Logger logger =
LoggerFactory.getLogger(PushConsumerExample.class);
+
+ private PushConsumerExample() {
+ }
+
+ public static void main(String[] args) throws ClientException,
IOException, InterruptedException {
+ final ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ String endpoints = "localhost:8081";
+ ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
+ .setEndpoints(endpoints)
+ .build();
+ String tag = "*";
+ FilterExpression filterExpression = new FilterExpression(tag,
FilterExpressionType.TAG);
+ String consumerGroup = "YourConsumerGroup";
+ String topic = "YourTopic";
+ PushConsumer pushConsumer = provider.newPushConsumerBuilder()
+ .setClientConfiguration(clientConfiguration)
+ .setConsumerGroup(consumerGroup)
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ .setMessageListener(messageView -> {
+ logger.info("Consume message successfully, messageId={}",
messageView.getMessageId());
+ return ConsumeResult.SUCCESS;
+ })
+ .build();
+ Thread.sleep(Long.MAX_VALUE);
+ // pushConsumer.close();
+ }
+ }
+ ```
## 6. Shutdown Servers
diff --git
a/i18n/en/docusaurus-plugin-content-docs/version-5.0/04-featureBehavior/06consumertype.md
b/i18n/en/docusaurus-plugin-content-docs/version-5.0/04-featureBehavior/06consumertype.md
index 24607afc6..3dbb2d08f 100644
---
a/i18n/en/docusaurus-plugin-content-docs/version-5.0/04-featureBehavior/06consumertype.md
+++
b/i18n/en/docusaurus-plugin-content-docs/version-5.0/04-featureBehavior/06consumertype.md
@@ -56,25 +56,24 @@ Sample code:
```java
// Message consumption example: Use a PushConsumer consumer to consume
messages.
ClientServiceProvider provider = ClientServiceProvider.loadService();
- String topic = "Your Topic";
- FilterExpression filterExpression = new FilterExpression("Your Filter
Tag", FilterExpressionType.TAG);
- PushConsumer pushConsumer = provider.newPushConsumerBuilder()
- // Configure consumer groups.
- .setConsumerGroup("Your ConsumerGroup")
- // Specify the access point.
-
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your
Endpoint").build())
- // Specify the pre-bound subscriptions.
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- // Set the message listener.
- .setMessageListener(new MessageListener() {
- @Override
- public ConsumeResult consume(MessageView messageView) {
- // Consume the messages and return the consumption
result.
- return ConsumeResult.SUCCESS;
- }
- })
- .build();
-
+String topic = "YourTopic";
+FilterExpression filterExpression = new FilterExpression("YourFilterTag",
FilterExpressionType.TAG);
+PushConsumer pushConsumer = provider.newPushConsumerBuilder()
+ // Configure consumer group.
+ .setConsumerGroup("YourConsumerGroup")
+ // Specify the access point.
+
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())
+ // Specify the pre-bound subscriptions.
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ // Set the message listener.
+ .setMessageListener(new MessageListener() {
+ @Override
+ public ConsumeResult consume(MessageView messageView) {
+ // Consume the messages and return the consumption result.
+ return ConsumeResult.SUCCESS;
+ }
+ })
+ .build();
```
The message listener for a PushConsumer consumer returns one of the following
results:
@@ -133,37 +132,36 @@ SimpleConsumer is a consumer type that supports atomic
operations for message pr
SimpleConsumer involves multiple API operations. The corresponding operations
are called as needed to obtain and distribute messages to business threads for
processing. Then, the commit operation is called to commit message processing
results. Sample code:
```java
- // Consumption example: When a SimpleConsumer consumer consumes normal
messages, the consumer obtain messages and commit message consumption results.
- ClientServiceProvider provider1 = ClientServiceProvider.loadService();
- String topic1 = "Your Topic";
- FilterExpression filterExpression1 = new FilterExpression("Your Filter
Tag", FilterExpressionType.TAG);
- SimpleConsumer simpleConsumer = provider1.newSimpleConsumerBuilder()
- // Configure consumer groups.
- .setConsumerGroup("Your ConsumerGroup")
- // Specify the access point.
-
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your
Endpoint").build())
- // Specify the pre-bound subscriptions.
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- // Specify the max await time when receive messages from the
server.
- .setAwaitDuration(Duration.ofSeconds(1))
- .build();
- List<MessageView> messageViewList = null;
+// Consumption example: When a SimpleConsumer consumer consumes normal
messages, the consumer obtain messages and commit message consumption results.
+ClientServiceProvider provider = ClientServiceProvider.loadService();
+String topic = "YourTopic";
+FilterExpression filterExpression = new FilterExpression("YourFilterTag",
FilterExpressionType.TAG);
+SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
+ // Configure consumer group.
+ .setConsumerGroup("YourConsumerGroup")
+ // Specify the access point.
+
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())
+ // Specify the pre-bound subscriptions.
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ // Specify the max await time when receive messages from the server.
+ .setAwaitDuration(Duration.ofSeconds(1))
+ .build();
+try {
+ // A SimpleConsumer consumer must obtain and process messages.
+ List<MessageView> messageViewList = simpleConsumer.receive(10,
Duration.ofSeconds(30));
+ messageViewList.forEach(messageView -> {
+ System.out.println(messageView);
+ // After consumption is complete, the consumer must invoke ACK to
submit the consumption result.
try {
- // A SimpleConsumer consumer must obtain and process messages.
- messageViewList = simpleConsumer.receive(10,
Duration.ofSeconds(30));
- messageViewList.forEach(messageView -> {
- System.out.println(messageView);
- // After consumption is complete, the consumer must invoke ACK
to submit the consumption result.
- try {
- simpleConsumer.ack(messageView);
- } catch (ClientException e) {
- e.printStackTrace();
- }
- });
+ simpleConsumer.ack(messageView);
} catch (ClientException e) {
- // If the pull fails due to system traffic throttling or other
reasons, the consumer must re-initiate the request to obtain the message.
- e.printStackTrace();
+ logger.error("Failed to ack message, messageId={}",
messageView.getMessageId(), e);
}
+ });
+} catch (ClientException e) {
+ // If the pull fails due to system traffic throttling or other reasons,
the consumer must re-initiate the request to obtain the message.
+ logger.error("Failed to receive message", e);
+}
```
diff --git a/versioned_docs/version-5.0/02-quickStart/01quickstart.md
b/versioned_docs/version-5.0/02-quickStart/01quickstart.md
index 1b0144443..61059f87d 100644
--- a/versioned_docs/version-5.0/02-quickStart/01quickstart.md
+++ b/versioned_docs/version-5.0/02-quickStart/01quickstart.md
@@ -110,109 +110,111 @@ $ sh bin/tools.sh
org.apache.rocketmq.example.quickstart.Consumer
4. 在已创建的Java工程中，创建发送普通消息程序并运行，示例代码如下：
- ```java
- import org.apache.rocketmq.client.apis.*;
- import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
- import org.apache.rocketmq.client.apis.consumer.MessageListener;
- import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
- import org.apache.rocketmq.client.apis.message.Message;
- import org.apache.rocketmq.client.apis.message.MessageBuilder;
- import org.apache.rocketmq.client.apis.message.MessageView;
- import org.apache.rocketmq.client.apis.producer.Producer;
- import org.apache.rocketmq.client.apis.producer.SendReceipt;
- import java.time.Duration;
- import java.util.List;
- public class ProducerExample {
- public static void main(String[] args) throws ClientException {
- //接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
- String endpoint = "localhost:8081";
- //消息发送的目标Topic名称，需要提前创建。
- String topic = "TestTopic";
- ClientServiceProvider provider =
ClientServiceProvider.loadService();
- ClientConfigurationBuilder builder =
ClientConfiguration.newBuilder().setEndpoints(endpoint);
- ClientConfiguration configuration = builder.build();
- //初始化Producer时需要设置通信配置以及预绑定的Topic。
- Producer producer = provider.newProducerBuilder()
- .setTopics(topic)
- .setClientConfiguration(configuration)
- .build();
- //普通消息发送。
- Message message = provider.newMessageBuilder()
- .setTopic(topic)
- //设置消息索引键，可根据关键字精确查找某条消息。
- .setKeys("messageKey")
- //设置消息Tag，用于消费端根据指定Tag过滤消息。
- .setTag("messageTag")
- //消息体。
- .setBody("messageBody".getBytes())
- .build();
- try {
- //发送消息，需要关注发送结果，并捕获失败等异常。
- SendReceipt sendReceipt = producer.send(message);
- System.out.println(sendReceipt.getMessageId());
- } catch (ClientException e) {
- e.printStackTrace();
- }
- }
- }
- ```
+ ```java
+ import org.apache.rocketmq.client.apis.ClientConfiguration;
+ import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
+ import org.apache.rocketmq.client.apis.ClientException;
+ import org.apache.rocketmq.client.apis.ClientServiceProvider;
+ import org.apache.rocketmq.client.apis.message.Message;
+ import org.apache.rocketmq.client.apis.producer.Producer;
+ import org.apache.rocketmq.client.apis.producer.SendReceipt;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ public class ProducerExample {
+ private static final Logger logger =
LoggerFactory.getLogger(ProducerExample.class);
+
+ public static void main(String[] args) throws ClientException {
+ // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
+ String endpoint = "localhost:8081";
+ // 消息发送的目标Topic名称，需要提前创建。
+ String topic = "TestTopic";
+ ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ ClientConfigurationBuilder builder =
ClientConfiguration.newBuilder().setEndpoints(endpoint);
+ ClientConfiguration configuration = builder.build();
+ // 初始化Producer时需要设置通信配置以及预绑定的Topic。
+ Producer producer = provider.newProducerBuilder()
+ .setTopics(topic)
+ .setClientConfiguration(configuration)
+ .build();
+ // 普通消息发送。
+ Message message = provider.newMessageBuilder()
+ .setTopic(topic)
+ // 设置消息索引键，可根据关键字精确查找某条消息。
+ .setKeys("messageKey")
+ // 设置消息Tag，用于消费端根据指定Tag过滤消息。
+ .setTag("messageTag")
+ // 消息体。
+ .setBody("messageBody".getBytes())
+ .build();
+ try {
+ // 发送消息，需要关注发送结果，并捕获失败等异常。
+ SendReceipt sendReceipt = producer.send(message);
+ logger.info("Send message successfully, messageId={}",
sendReceipt.getMessageId());
+ } catch (ClientException e) {
+ logger.error("Failed to send message", e);
+ }
+ // producer.close();
+ }
+ }
+ ```
5. 在已创建的Java工程中，创建订阅普通消息程序并运行。Apache RocketMQ
支持[SimpleConsumer](../04-featureBehavior/06consumertype.md)和[PushConsumer](../04-featureBehavior/06consumertype.md)两种消费者类型，您可以选择以下任意一种方式订阅消息。
- ```java
- import org.apache.rocketmq.client.apis.*;
- import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
- import org.apache.rocketmq.client.apis.consumer.FilterExpression;
- import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
- import org.apache.rocketmq.client.apis.consumer.PushConsumer;
-
- import java.io.IOException;
- import java.util.Collections;
- import org.slf4j.Logger;
- import org.slf4j.LoggerFactory;
-
- public class PushConsumerExample {
- private static final Logger LOGGER =
LoggerFactory.getLogger(PushConsumerExample.class);
-
- private PushConsumerExample() {
- }
-
- public static void main(String[] args) throws ClientException,
IOException, InterruptedException {
- final ClientServiceProvider provider =
ClientServiceProvider.loadService();
- //接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
- String endpoints = "localhost:8081";
- ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
- .setEndpoints(endpoints)
- .build();
- //订阅消息的过滤规则，表示订阅所有Tag的消息。
- String tag = "*";
- FilterExpression filterExpression = new FilterExpression(tag,
FilterExpressionType.TAG);
- //为消费者指定所属的消费者分组，Group需要提前创建。
- String consumerGroup = "Your ConsumerGroup";
- //指定需要订阅哪个目标Topic，Topic需要提前创建。
- String topic = "TestTopic";
- //初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
- PushConsumer pushConsumer = provider.newPushConsumerBuilder()
- .setClientConfiguration(clientConfiguration)
- //设置消费者分组。
- .setConsumerGroup(consumerGroup)
- //设置预绑定的订阅关系。
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- //设置消费监听器。
- .setMessageListener(messageView -> {
- //处理消息并返回消费结果。
- // LOGGER.info("Consume message={}", messageView);
- System.out.println("Consume message!!");
- return ConsumeResult.SUCCESS;
- })
- .build();
- Thread.sleep(Long.MAX_VALUE);
- //如果不需要再使用PushConsumer，可关闭该进程。
- //pushConsumer.close();
- }
- }
- ```
+ ```java
+ import java.io.IOException;
+ import java.util.Collections;
+ import org.apache.rocketmq.client.apis.ClientConfiguration;
+ import org.apache.rocketmq.client.apis.ClientException;
+ import org.apache.rocketmq.client.apis.ClientServiceProvider;
+ import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
+ import org.apache.rocketmq.client.apis.consumer.FilterExpression;
+ import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
+ import org.apache.rocketmq.client.apis.consumer.PushConsumer;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ public class PushConsumerExample {
+ private static final Logger logger =
LoggerFactory.getLogger(PushConsumerExample.class);
+
+ private PushConsumerExample() {
+ }
+
+ public static void main(String[] args) throws ClientException,
IOException, InterruptedException {
+ final ClientServiceProvider provider =
ClientServiceProvider.loadService();
+ // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
+ String endpoints = "localhost:8081";
+ ClientConfiguration clientConfiguration =
ClientConfiguration.newBuilder()
+ .setEndpoints(endpoints)
+ .build();
+ // 订阅消息的过滤规则，表示订阅所有Tag的消息。
+ String tag = "*";
+ FilterExpression filterExpression = new FilterExpression(tag,
FilterExpressionType.TAG);
+ // 为消费者指定所属的消费者分组，Group需要提前创建。
+ String consumerGroup = "YourConsumerGroup";
+ // 指定需要订阅哪个目标Topic，Topic需要提前创建。
+ String topic = "TestTopic";
+ // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
+ PushConsumer pushConsumer = provider.newPushConsumerBuilder()
+ .setClientConfiguration(clientConfiguration)
+ // 设置消费者分组。
+ .setConsumerGroup(consumerGroup)
+ // 设置预绑定的订阅关系。
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ // 设置消费监听器。
+ .setMessageListener(messageView -> {
+ // 处理消息并返回消费结果。
+ logger.info("Consume message successfully, messageId={}",
messageView.getMessageId());
+ return ConsumeResult.SUCCESS;
+ })
+ .build();
+ Thread.sleep(Long.MAX_VALUE);
+ // 如果不需要再使用 PushConsumer，可关闭该实例。
+ // pushConsumer.close();
+ }
+ }
+ ```
## 6. 关闭服务器
diff --git a/versioned_docs/version-5.0/04-featureBehavior/06consumertype.md
b/versioned_docs/version-5.0/04-featureBehavior/06consumertype.md
index e54304e0a..82c387b95 100644
--- a/versioned_docs/version-5.0/04-featureBehavior/06consumertype.md
+++ b/versioned_docs/version-5.0/04-featureBehavior/06consumertype.md
@@ -56,26 +56,26 @@ PushConsumer的使用方式比较固定，在消费者初始化时注册一个
示例代码如下：
```java
-//消费示例：使用PushConsumer消费普通消息。
+// 消费示例：使用PushConsumer消费普通消息。
ClientServiceProvider provider = ClientServiceProvider.loadService();
- String topic = "Your Topic";
- FilterExpression filterExpression = new FilterExpression("Your Filter
Tag", FilterExpressionType.TAG);
- PushConsumer pushConsumer = provider.newPushConsumerBuilder()
- //设置消费者分组。
- .setConsumerGroup("Your ConsumerGroup")
- //设置接入点。
-
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your
Endpoint").build())
- //设置预绑定的订阅关系。
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- //设置消费监听器。
- .setMessageListener(new MessageListener() {
- @Override
- public ConsumeResult consume(MessageView messageView) {
- //消费消息并返回处理结果。
- return ConsumeResult.SUCCESS;
- }
- })
- .build();
+String topic = "YourTopic";
+FilterExpression filterExpression = new FilterExpression("YourFilterTag",
FilterExpressionType.TAG);
+PushConsumer pushConsumer = provider.newPushConsumerBuilder()
+ // 设置消费者分组。
+ .setConsumerGroup("YourConsumerGroup")
+ // 设置接入点。
+
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())
+ // 设置预绑定的订阅关系。
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ // 设置消费监听器。
+ .setMessageListener(new MessageListener() {
+ @Override
+ public ConsumeResult consume(MessageView messageView) {
+ // 消费消息并返回处理结果。
+ return ConsumeResult.SUCCESS;
+ }
+ })
+ .build();
```
@@ -138,37 +138,36 @@ SimpleConsumer 是一种接口原子型的消费者类型，消息的获取、
SimpleConsumer
的使用涉及多个接口调用，由业务逻辑按需调用接口获取消息，然后分发给业务线程处理消息，最后按照处理的结果调用提交接口，返回服务端当前消息的处理结果。示例如下：
```java
- //消费示例：使用SimpleConsumer消费普通消息，主动获取消息处理并提交。
- ClientServiceProvider provider1 = ClientServiceProvider.loadService();
- String topic1 = "Your Topic";
- FilterExpression filterExpression1 = new FilterExpression("Your Filter
Tag", FilterExpressionType.TAG);
- SimpleConsumer simpleConsumer = provider1.newSimpleConsumerBuilder()
- //设置消费者分组。
- .setConsumerGroup("Your ConsumerGroup")
- //设置接入点。
-
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your
Endpoint").build())
- //设置预绑定的订阅关系。
- .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
- //设置从服务端接受消息的最大等待时间
- .setAwaitDuration(Duration.ofSeconds(1))
- .build();
- List<MessageView> messageViewList = null;
+// 消费示例：使用 SimpleConsumer 消费普通消息，主动获取消息处理并提交。
+ClientServiceProvider provider = ClientServiceProvider.loadService();
+String topic = "YourTopic";
+FilterExpression filterExpression = new FilterExpression("YourFilterTag",
FilterExpressionType.TAG);
+SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
+ // 设置消费者分组。
+ .setConsumerGroup("YourConsumerGroup")
+ // 设置接入点。
+
.setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build())
+ // 设置预绑定的订阅关系。
+ .setSubscriptionExpressions(Collections.singletonMap(topic,
filterExpression))
+ // 设置从服务端接受消息的最大等待时间
+ .setAwaitDuration(Duration.ofSeconds(1))
+ .build();
+try {
+ // SimpleConsumer 需要主动获取消息，并处理。
+ List<MessageView> messageViewList = simpleConsumer.receive(10,
Duration.ofSeconds(30));
+ messageViewList.forEach(messageView -> {
+ System.out.println(messageView);
+ // 消费处理完成后，需要主动调用 ACK 提交消费结果。
try {
- //SimpleConsumer需要主动获取消息，并处理。
- messageViewList = simpleConsumer.receive(10,
Duration.ofSeconds(30));
- messageViewList.forEach(messageView -> {
- System.out.println(messageView);
- //消费处理完成后，需要主动调用ACK提交消费结果。
- try {
- simpleConsumer.ack(messageView);
- } catch (ClientException e) {
- e.printStackTrace();
- }
- });
+ simpleConsumer.ack(messageView);
} catch (ClientException e) {
- //如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
- e.printStackTrace();
+ logger.error("Failed to ack message, messageId={}",
messageView.getMessageId(), e);
}
+ });
+} catch (ClientException e) {
+ // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
+ logger.error("Failed to receive message", e);
+}
```
SimpleConsumer主要涉及以下几个接口行为：
