Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Lil_Lazy
Rocketmq
提交
e4daba5f
R
Rocketmq
项目概览
Lil_Lazy
/
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,发现更多精彩内容 >>
未验证
提交
e4daba5f
编写于
7月 15, 2021
作者:
H
Heng Du
提交者:
GitHub
7月 15, 2021
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #3042 from haoran-pku/master
Add SyncProducer Example
上级
89d9ed76
a99cf373
变更
1
显示空白变更内容
内联
并排
Showing
1 changed file
with
136 addition
and
0 deletion
+136
-0
docs/cn/Example_Simple_cn.md
docs/cn/Example_Simple_cn.md
+136
-0
未找到文件。
docs/cn/Example_Simple_cn.md
0 → 100644
浏览文件 @
e4daba5f
# Basic Sample
------
基本示例中提供了以下两个功能
*
RocketMQ可用于以三种方式发送消息:可靠的同步、可靠的异步和单向传输。前两种消息类型是可靠的,因为无论它们是否成功发送都有响应。
*
RocketMQ可以用来消费消息。
### 1 添加依赖
maven:
```
java
<
dependency
>
<
groupId
>
org
.
apache
.
rocketmq
</
groupId
>
<
artifactId
>
rocketmq
-
client
</
artifactId
>
<
version
>
4.3
.
0
</
version
>
</
dependency
>
```
gradle:
```
java
compile
'
org
.
apache
.
rocketmq
:
rocketmq
-
client:
4.3
.
0
'
```
### 2 发送消息
##### 2.1 使用Producer发送同步消息
可靠的同步传输被广泛应用于各种场景,如重要的通知消息、短消息通知等。
```
java
public
class
SyncProducer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
// Instantiate with a producer group name
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"please_rename_unique_group_name"
);
// Specify name server addresses
producer
.
setNamesrvAddr
(
"localhost:9876"
);
// Launch the producer instance
producer
.
start
();
for
(
int
i
=
0
;
i
<
100
;
i
++)
{
// Create a message instance with specifying topic, tag and message body
Message
msg
=
new
Message
(
"TopicTest"
/* Topic */
,
"TagA"
/* Tag */
,
(
"Hello RocketMQ "
+
i
).
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)
/* Message body */
);
// Send message to one of brokers
SendResult
sendResult
=
producer
.
send
(
msg
);
// Check whether the message has been delivered by the callback of sendResult
System
.
out
.
printf
(
"%s%n"
,
sendResult
);
}
// Shut down once the producer instance is not longer in use
producer
.
shutdown
();
}
}
```
##### 2.2 发送异步消息
异步传输通常用于响应时间敏感的业务场景。这意味着发送方无法等待代理的响应太长时间。
```
java
public
class
AsyncProducer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
// Instantiate with a producer group name
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"please_rename_unique_group_name"
);
// Specify name server addresses
producer
.
setNamesrvAddr
(
"localhost:9876"
);
// Launch the producer instance
producer
.
start
();
producer
.
setRetryTimesWhenSendAsyncFailed
(
0
);
for
(
int
i
=
0
;
i
<
100
;
i
++)
{
final
int
index
=
i
;
// Create a message instance with specifying topic, tag and message body
Message
msg
=
new
Message
(
"TopicTest"
,
"TagA"
,
"OrderID188"
,
"Hello world"
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
));
// SendCallback: receive the callback of the asynchronous return result.
producer
.
send
(
msg
,
new
SendCallback
()
{
@Override
public
void
onSuccess
(
SendResult
sendResult
)
{
System
.
out
.
printf
(
"%-10d OK %s %n"
,
index
,
sendResult
.
getMsgId
());
}
@Override
public
void
onException
(
Throwable
e
)
{
System
.
out
.
printf
(
"%-10d Exception %s %n"
,
index
,
e
);
e
.
printStackTrace
();
}
});
}
// Shut down once the producer instance is not longer in use
producer
.
shutdown
();
}
}
```
##### 2.3 以单向模式发送消息
单向传输用于需要中等可靠性的情况,如日志收集。
```
java
public
class
OnewayProducer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
// Instantiate with a producer group name
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"please_rename_unique_group_name"
);
// Specify name server addresses
producer
.
setNamesrvAddr
(
"localhost:9876"
);
// Launch the producer instance
producer
.
start
();
for
(
int
i
=
0
;
i
<
100
;
i
++)
{
// Create a message instance with specifying topic, tag and message body
Message
msg
=
new
Message
(
"TopicTest"
/* Topic */
,
"TagA"
/* Tag */
,
(
"Hello RocketMQ "
+
i
).
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)
/* Message body */
);
// Send in one-way mode, no return result
producer
.
sendOneway
(
msg
);
}
// Shut down once the producer instance is not longer in use
producer
.
shutdown
();
}
}
```
### 3 消费消息
```
java
public
class
Consumer
{
public
static
void
main
(
String
[]
args
)
throws
InterruptedException
,
MQClientException
{
// Instantiate with specified consumer group name
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"please_rename_unique_group_name"
);
// Specify name server addresses
consumer
.
setNamesrvAddr
(
"localhost:9876"
);
// Subscribe one or more topics and tags for finding those messages need to be consumed
consumer
.
subscribe
(
"TopicTest"
,
"*"
);
// Register callback to execute on arrival of messages fetched from brokers
consumer
.
registerMessageListener
(
new
MessageListenerConcurrently
()
{
@Override
public
ConsumeConcurrentlyStatus
consumeMessage
(
List
<
MessageExt
>
msgs
,
ConsumeConcurrentlyContext
context
)
{
System
.
out
.
printf
(
"%s Receive New Messages: %s %n"
,
Thread
.
currentThread
().
getName
(),
msgs
);
// Mark the message that have been consumed successfully
return
ConsumeConcurrentlyStatus
.
CONSUME_SUCCESS
;
}
});
// Launch the consumer instance
consumer
.
start
();
System
.
out
.
printf
(
"Consumer Started.%n"
);
}
}
```
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录