Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
caopu16
whatsmars
提交
ea47f7d3
W
whatsmars
项目概览
caopu16
/
whatsmars
与 Fork 源项目一致
Fork自
武汉红喜 / whatsmars
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
W
whatsmars
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
ea47f7d3
编写于
12月 13, 2018
作者:
武汉红喜
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
RocketMQTemplate优化(+SendResult)
上级
f6300c09
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
46 addition
and
39 deletion
+46
-39
whatsmars-mq/whatsmars-mq-rocketmq-spring/src/main/java/org/hongxi/whatsmars/rocketmq/config/spring/RocketMQTemplate.java
...xi/whatsmars/rocketmq/config/spring/RocketMQTemplate.java
+19
-16
whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/rocketmq/spring/demo/Producer.java
...a/org/hongxi/whatsmars/rocketmq/spring/demo/Producer.java
+3
-1
whatsmars-mq/whatsmars-mq-sdk/src/main/java/org/hongxi/whatsmars/common/rocketmq/RocketMQTemplate.java
...rg/hongxi/whatsmars/common/rocketmq/RocketMQTemplate.java
+21
-20
whatsmars-mq/whatsmars-mq-sdk/src/test/java/org/hongxi/whatsmars/common/rocketmq/RocketMQTest.java
...va/org/hongxi/whatsmars/common/rocketmq/RocketMQTest.java
+3
-2
未找到文件。
whatsmars-mq/whatsmars-mq-rocketmq-spring/src/main/java/org/hongxi/whatsmars/rocketmq/config/spring/RocketMQTemplate.java
浏览文件 @
ea47f7d3
...
...
@@ -4,6 +4,7 @@ import lombok.extern.slf4j.Slf4j;
import
org.apache.commons.lang3.math.NumberUtils
;
import
org.apache.rocketmq.client.producer.DefaultMQProducer
;
import
org.apache.rocketmq.client.producer.MessageQueueSelector
;
import
org.apache.rocketmq.client.producer.SendResult
;
import
org.apache.rocketmq.common.message.Message
;
import
org.apache.rocketmq.common.message.MessageQueue
;
import
org.apache.rocketmq.remoting.common.RemotingHelper
;
...
...
@@ -22,17 +23,17 @@ public class RocketMQTemplate {
this
.
defaultMQProducer
=
producer
;
}
public
void
send
(
String
topic
,
String
body
)
{
send
(
topic
,
""
,
body
);
public
SendResult
send
(
String
topic
,
String
body
)
{
return
send
(
topic
,
""
,
body
);
}
public
void
send
(
String
topic
,
String
tags
,
String
body
)
{
send
(
topic
,
tags
,
""
,
body
);
public
SendResult
send
(
String
topic
,
String
tags
,
String
body
)
{
return
send
(
topic
,
tags
,
""
,
body
);
}
public
void
send
(
String
topic
,
String
tags
,
String
keys
,
String
body
)
{
public
SendResult
send
(
String
topic
,
String
tags
,
String
keys
,
String
body
)
{
try
{
send
(
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)));
return
send
(
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)));
}
catch
(
Exception
e
)
{
log
.
error
(
"send error, topic:{}, tags:{}, keys:{}, body:{}"
,
topic
,
tags
,
keys
,
body
,
e
);
...
...
@@ -40,18 +41,19 @@ public class RocketMQTemplate {
}
}
private
void
send
(
Message
message
)
throws
Exception
{
this
.
defaultMQProducer
.
send
(
message
);
log
.
info
(
"send successful:{}"
,
message
);
private
SendResult
send
(
Message
message
)
throws
Exception
{
SendResult
sendResult
=
this
.
defaultMQProducer
.
send
(
message
);
log
.
debug
(
"send result: {}"
,
sendResult
);
return
sendResult
;
}
public
void
sendOrderly
(
String
topic
,
String
keys
,
String
body
)
{
sendOrderly
(
topic
,
""
,
keys
,
body
);
public
SendResult
sendOrderly
(
String
topic
,
String
keys
,
String
body
)
{
return
sendOrderly
(
topic
,
""
,
keys
,
body
);
}
public
void
sendOrderly
(
String
topic
,
String
tags
,
String
keys
,
String
body
)
{
public
SendResult
sendOrderly
(
String
topic
,
String
tags
,
String
keys
,
String
body
)
{
try
{
sendOrderly
(
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)));
return
sendOrderly
(
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)));
}
catch
(
Exception
e
)
{
log
.
error
(
"send error, topic:{}, tags:{}, keys:{}, body:{}"
,
topic
,
tags
,
keys
,
body
,
e
);
...
...
@@ -59,8 +61,8 @@ public class RocketMQTemplate {
}
}
private
void
sendOrderly
(
Message
message
)
throws
Exception
{
this
.
defaultMQProducer
.
send
(
message
,
private
SendResult
sendOrderly
(
Message
message
)
throws
Exception
{
SendResult
sendResult
=
this
.
defaultMQProducer
.
send
(
message
,
new
MessageQueueSelector
()
{
@Override
public
MessageQueue
select
(
List
<
MessageQueue
>
mqs
,
Message
msg
,
Object
arg
)
{
...
...
@@ -69,7 +71,8 @@ public class RocketMQTemplate {
return
mqs
.
get
(
index
);
}
},
message
.
getKeys
());
log
.
info
(
"send successful:{}"
,
message
);
log
.
debug
(
"send result: {}"
,
sendResult
);
return
sendResult
;
}
}
whatsmars-mq/whatsmars-mq-rocketmq/src/main/java/org/hongxi/whatsmars/rocketmq/spring/demo/Producer.java
浏览文件 @
ea47f7d3
package
org.hongxi.whatsmars.rocketmq.spring.demo
;
import
org.apache.rocketmq.client.producer.SendResult
;
import
org.hongxi.whatsmars.rocketmq.config.spring.RocketMQTemplate
;
import
org.springframework.context.support.ClassPathXmlApplicationContext
;
...
...
@@ -10,7 +11,8 @@ public class Producer {
RocketMQTemplate
rocketMQTemplate
=
(
RocketMQTemplate
)
context
.
getBean
(
"rocketMQTemplate"
);
for
(
int
i
=
0
;
i
<
20
;
i
++)
{
try
{
rocketMQTemplate
.
send
(
"sdk-test"
,
"rocketMQTemplate"
);
SendResult
sendResult
=
rocketMQTemplate
.
send
(
"sdk-test"
,
"rocketMQTemplate"
);
System
.
out
.
println
(
sendResult
);
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
}
...
...
whatsmars-mq/whatsmars-mq-sdk/src/main/java/org/hongxi/whatsmars/common/rocketmq/RocketMQTemplate.java
浏览文件 @
ea47f7d3
...
...
@@ -5,6 +5,7 @@ import org.apache.commons.lang3.math.NumberUtils;
import
org.apache.rocketmq.client.log.ClientLogger
;
import
org.apache.rocketmq.client.producer.DefaultMQProducer
;
import
org.apache.rocketmq.client.producer.MessageQueueSelector
;
import
org.apache.rocketmq.client.producer.SendResult
;
import
org.apache.rocketmq.common.message.Message
;
import
org.apache.rocketmq.common.message.MessageQueue
;
import
org.apache.rocketmq.logging.InternalLogger
;
...
...
@@ -59,25 +60,25 @@ public class RocketMQTemplate {
return
producerMap
.
get
(
producerKey
);
}
public
static
void
send
(
String
topic
,
String
body
)
{
send
(
DEFAULT_PRODUCER_GROUP
,
topic
,
body
,
DEFAULT_SEND_MSG_TIMEOUT
);
public
static
SendResult
send
(
String
topic
,
String
body
)
{
return
send
(
DEFAULT_PRODUCER_GROUP
,
topic
,
body
,
DEFAULT_SEND_MSG_TIMEOUT
);
}
public
static
void
send
(
String
topic
,
String
body
,
int
sendMsgTimeout
)
{
send
(
DEFAULT_PRODUCER_GROUP
,
topic
,
body
,
sendMsgTimeout
);
public
static
SendResult
send
(
String
topic
,
String
body
,
int
sendMsgTimeout
)
{
return
send
(
DEFAULT_PRODUCER_GROUP
,
topic
,
body
,
sendMsgTimeout
);
}
public
static
void
send
(
String
producerGroup
,
String
topic
,
String
body
,
int
sendMsgTimeout
)
{
send
(
producerGroup
,
topic
,
""
,
body
,
sendMsgTimeout
);
public
static
SendResult
send
(
String
producerGroup
,
String
topic
,
String
body
,
int
sendMsgTimeout
)
{
return
send
(
producerGroup
,
topic
,
""
,
body
,
sendMsgTimeout
);
}
public
static
void
send
(
String
producerGroup
,
String
topic
,
String
tags
,
String
body
,
int
sendMsgTimeout
)
{
send
(
producerGroup
,
topic
,
tags
,
""
,
body
,
sendMsgTimeout
);
public
static
SendResult
send
(
String
producerGroup
,
String
topic
,
String
tags
,
String
body
,
int
sendMsgTimeout
)
{
return
send
(
producerGroup
,
topic
,
tags
,
""
,
body
,
sendMsgTimeout
);
}
public
static
void
send
(
String
producerGroup
,
String
topic
,
String
tags
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
public
static
SendResult
send
(
String
producerGroup
,
String
topic
,
String
tags
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
try
{
send
(
producerGroup
,
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)),
sendMsgTimeout
);
return
send
(
producerGroup
,
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)),
sendMsgTimeout
);
}
catch
(
Exception
e
)
{
log
.
error
(
"send error, producerGroup:{}, topic:{}, tags:{}, keys:{}, body:{}"
,
producerGroup
,
topic
,
tags
,
keys
,
body
,
e
);
...
...
@@ -85,21 +86,21 @@ public class RocketMQTemplate {
}
}
private
static
void
send
(
String
producerGroup
,
Message
message
,
int
sendMsgTimeout
)
throws
Exception
{
getProducer
(
producerGroup
,
sendMsgTimeout
).
send
(
message
);
private
static
SendResult
send
(
String
producerGroup
,
Message
message
,
int
sendMsgTimeout
)
throws
Exception
{
return
getProducer
(
producerGroup
,
sendMsgTimeout
).
send
(
message
);
}
public
static
void
sendOrderly
(
String
producerGroup
,
String
topic
,
String
keys
,
String
body
)
{
sendOrderly
(
producerGroup
,
topic
,
keys
,
body
,
DEFAULT_SEND_MSG_TIMEOUT
);
public
static
SendResult
sendOrderly
(
String
producerGroup
,
String
topic
,
String
keys
,
String
body
)
{
return
sendOrderly
(
producerGroup
,
topic
,
keys
,
body
,
DEFAULT_SEND_MSG_TIMEOUT
);
}
public
static
void
sendOrderly
(
String
producerGroup
,
String
topic
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
sendOrderly
(
producerGroup
,
topic
,
keys
,
body
,
sendMsgTimeout
);
public
static
SendResult
sendOrderly
(
String
producerGroup
,
String
topic
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
return
sendOrderly
(
producerGroup
,
topic
,
keys
,
body
,
sendMsgTimeout
);
}
public
static
void
sendOrderly
(
String
producerGroup
,
String
topic
,
String
tags
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
public
static
SendResult
sendOrderly
(
String
producerGroup
,
String
topic
,
String
tags
,
String
keys
,
String
body
,
int
sendMsgTimeout
)
{
try
{
sendOrderly
(
producerGroup
,
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)),
sendMsgTimeout
);
return
sendOrderly
(
producerGroup
,
new
Message
(
topic
,
tags
,
keys
,
body
.
getBytes
(
RemotingHelper
.
DEFAULT_CHARSET
)),
sendMsgTimeout
);
}
catch
(
Exception
e
)
{
log
.
error
(
"send error, producerGroup:{}, topic:{}, tags:{}, keys:{}, body:{}"
,
producerGroup
,
topic
,
tags
,
keys
,
body
,
e
);
...
...
@@ -107,8 +108,8 @@ public class RocketMQTemplate {
}
}
private
static
void
sendOrderly
(
String
producerGroup
,
Message
message
,
int
sendMsgTimeout
)
throws
Exception
{
getProducer
(
producerGroup
,
sendMsgTimeout
).
send
(
message
,
private
static
SendResult
sendOrderly
(
String
producerGroup
,
Message
message
,
int
sendMsgTimeout
)
throws
Exception
{
return
getProducer
(
producerGroup
,
sendMsgTimeout
).
send
(
message
,
new
MessageQueueSelector
()
{
@Override
public
MessageQueue
select
(
List
<
MessageQueue
>
mqs
,
Message
msg
,
Object
arg
)
{
...
...
whatsmars-mq/whatsmars-mq-sdk/src/test/java/org/hongxi/whatsmars/common/rocketmq/RocketMQTest.java
浏览文件 @
ea47f7d3
package
org.hongxi.whatsmars.common.rocketmq
;
import
org.apache.rocketmq.client.producer.SendResult
;
import
org.junit.Test
;
/**
...
...
@@ -16,13 +17,13 @@ public class RocketMQTest {
int
count
=
0
;
for
(
int
i
=
0
;
i
<
20
;
i
++)
{
try
{
RocketMQTemplate
.
send
(
TEST_TOPIC
,
"hello"
+
i
,
1000
);
SendResult
sendResult
=
RocketMQTemplate
.
send
(
TEST_TOPIC
,
"hello"
+
i
,
1000
);
System
.
out
.
println
(
sendResult
);
count
++;
}
catch
(
Exception
e
)
{
e
.
printStackTrace
();
}
}
System
.
out
.
println
(
"send successful: "
+
count
);
}
@Test
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录