Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
码出高效码出未来
Rocketmq
提交
278d3285
R
Rocketmq
项目概览
码出高效码出未来
/
Rocketmq
与 Fork 源项目一致
Fork自
Apache RocketMQ / Rocketmq
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
Rocketmq
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
278d3285
编写于
6月 23, 2021
作者:
C
CuCu
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
add Chinese version
上级
df1d93fc
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
85 addition
and
0 deletion
+85
-0
docs/cn/Example_Delay.md
docs/cn/Example_Delay.md
+85
-0
未找到文件。
docs/cn/Example_Delay.md
0 → 100644
浏览文件 @
278d3285
# Schedule example
### 1 启动消费者等待传入的订阅消息
```
java
import
org.apache.rocketmq.client.consumer.DefaultMQPushConsumer
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext
;
import
org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
;
import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently
;
import
org.apache.rocketmq.common.message.MessageExt
;
import
java.util.List
;
public
class
ScheduledMessageConsumer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
// Instantiate message consumer
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"ExampleConsumer"
);
// Subscribe topics
consumer
.
subscribe
(
"TestTopic"
,
"*"
);
// Register message listener
consumer
.
registerMessageListener
(
new
MessageListenerConcurrently
()
{
@Override
public
ConsumeConcurrentlyStatus
consumeMessage
(
List
<
MessageExt
>
messages
,
ConsumeConcurrentlyContext
context
)
{
for
(
MessageExt
message
:
messages
)
{
// Print approximate delay time period
System
.
out
.
println
(
"Receive message[msgId="
+
message
.
getMsgId
()
+
"] "
+
(
System
.
currentTimeMillis
()
-
message
.
getStoreTimestamp
())
+
"ms later"
);
}
return
ConsumeConcurrentlyStatus
.
CONSUME_SUCCESS
;
}
});
// Launch consumer
consumer
.
start
();
}
}
```
### 2 发送延迟消息
```
java
import
org.apache.rocketmq.client.producer.DefaultMQProducer
;
import
org.apache.rocketmq.common.message.Message
;
public
class
ScheduledMessageProducer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
// Instantiate a producer to send scheduled messages
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"ExampleProducerGroup"
);
// Launch producer
producer
.
start
();
int
totalMessagesToSend
=
100
;
for
(
int
i
=
0
;
i
<
totalMessagesToSend
;
i
++)
{
Message
message
=
new
Message
(
"TestTopic"
,
(
"Hello scheduled message "
+
i
).
getBytes
());
// This message will be delivered to consumer 10 seconds later.
message
.
setDelayTimeLevel
(
3
);
// Send the message
producer
.
send
(
message
);
}
// Shutdown producer after use.
producer
.
shutdown
();
}
}
```
### 3 确认
您应该会看到消息在其存储时间后大约 10 秒被消耗。
### 4 延迟消息的使用场景
例如在电子商务中,如果提交订单,可以发送延迟消息,1小时后可以查看订单状态。 如果订单仍未付款,则可以取消订单并释放库存。
### 5 使用延迟消息的限制
```
java
// org/apache/rocketmq/store/config/MessageStoreConfig.java
private
String
messageDelayLevel
=
"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
;
```
当前 RocketMQ 不支持任意时间的延迟。 生产者发送延迟消息前需要设置几个固定的延迟级别,分别对应1s到2h的1到18个延迟级,消息消费失败会进入延迟消息队列,消息发送时间与设置的延迟级别和重试次数有关。
See
`SendMessageProcessor.java`
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录