Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Apache RocketMQ
Rocketmq
提交
e9a87ea9
R
Rocketmq
项目概览
Apache RocketMQ
/
Rocketmq
上一次同步 大约 3 年
通知
267
Star
16139
Fork
68
代码
文件
提交
分支
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看板
提交
e9a87ea9
编写于
1月 22, 2017
作者:
Z
Zhanhui Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Add javadoc to quick start examples.
上级
e447bd01
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
63 addition
and
11 deletion
+63
-11
example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java
...java/org/apache/rocketmq/example/quickstart/Consumer.java
+28
-0
example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
...java/org/apache/rocketmq/example/quickstart/Producer.java
+35
-11
未找到文件。
example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java
浏览文件 @
e9a87ea9
...
...
@@ -25,15 +25,39 @@ import org.apache.rocketmq.client.exception.MQClientException;
import
org.apache.rocketmq.common.consumer.ConsumeFromWhere
;
import
org.apache.rocketmq.common.message.MessageExt
;
/**
* This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}.
*/
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_4"
);
/*
* Specify name server addresses.
* <p/>
*
* Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
*/
consumer
.
setNamesrvAddr
(
"name-server1-ip:9876;name-server2-ip:9876"
);
/*
* Specify where to start in case the specified consumer group is a brand new one.
*/
consumer
.
setConsumeFromWhere
(
ConsumeFromWhere
.
CONSUME_FROM_FIRST_OFFSET
);
/*
* Subscribe one more more topics to consume.
*/
consumer
.
subscribe
(
"TopicTest"
,
"*"
);
/*
* Register callback to execute on arrival of messages fetched from brokers.
*/
consumer
.
registerMessageListener
(
new
MessageListenerConcurrently
()
{
@Override
...
...
@@ -44,7 +68,11 @@ public class Consumer {
}
});
/*
* Launch the consumer instance.
*/
consumer
.
start
();
System
.
out
.
printf
(
"Consumer Started.%n"
);
}
}
example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
浏览文件 @
e9a87ea9
...
...
@@ -18,36 +18,60 @@ package org.apache.rocketmq.example.quickstart;
import
org.apache.rocketmq.client.exception.MQClientException
;
import
org.apache.rocketmq.client.producer.DefaultMQProducer
;
import
org.apache.rocketmq.client.producer.LocalTransactionExecuter
;
import
org.apache.rocketmq.client.producer.LocalTransactionState
;
import
org.apache.rocketmq.client.producer.SendResult
;
import
org.apache.rocketmq.common.message.Message
;
import
org.apache.rocketmq.remoting.common.RemotingHelper
;
/**
* This class demonstrates how to send messages to brokers using provided {@link DefaultMQProducer}.
*/
public
class
Producer
{
public
static
void
main
(
String
[]
args
)
throws
MQClientException
,
InterruptedException
{
/*
* Instantiate with a producer group name.
*/
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"please_rename_unique_group_name"
);
/*
* Specify name server addresses.
* <p/>
*
* Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
*/
producer
.
setNamesrvAddr
(
"name-server1-ip:9876;name-server2-ip:9876"
);
/*
* Launch the instance.
*/
producer
.
start
();
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
try
{
Message
msg
=
new
Message
(
"TopicTest"
,
"TagA"
,
(
"Hello RocketMQ "
+
i
).
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)
/*
* Create a message instance, specifying topic, tag and message body.
*/
Message
msg
=
new
Message
(
"TopicTest"
/* Topic */
,
"TagA"
/* Tag */
,
(
"Hello RocketMQ "
+
i
).
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)
/* Message body */
);
/*
* Call send message to deliver message to one of brokers.
*/
SendResult
sendResult
=
producer
.
send
(
msg
);
LocalTransactionExecuter
tranExecuter
=
new
LocalTransactionExecuter
()
{
@Override
public
LocalTransactionState
executeLocalTransactionBranch
(
Message
msg
,
Object
arg
)
{
return
null
;
}
};
System
.
out
.
printf
(
"%s%n"
,
sendResult
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
Thread
.
sleep
(
1000
);
}
}
/*
* Shut down once the producer instance is not longer in use.
*/
producer
.
shutdown
();
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录