Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
檀越@新空间
RocketMQ-demo
提交
f4cf9ed1
R
RocketMQ-demo
项目概览
檀越@新空间
/
RocketMQ-demo
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
RocketMQ-demo
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
提交
f4cf9ed1
编写于
5月 26, 2023
作者:
檀越@新空间
🐭
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix:配置
上级
f13f3ea7
变更
20
显示空白变更内容
内联
并排
Showing
20 changed file
with
47 addition
and
28 deletion
+47
-28
rocketmq-demo/pom.xml
rocketmq-demo/pom.xml
+7
-0
rocketmq-demo/shuyu.iml
rocketmq-demo/shuyu.iml
+13
-0
rocketmq-demo/src/main/java/base/consumer/Consumer.java
rocketmq-demo/src/main/java/base/consumer/Consumer.java
+6
-6
rocketmq-demo/src/main/java/base/producer/AsyncProducer.java
rocketmq-demo/src/main/java/base/producer/AsyncProducer.java
+4
-6
rocketmq-demo/src/main/java/base/producer/OneWayProducer.java
...etmq-demo/src/main/java/base/producer/OneWayProducer.java
+1
-1
rocketmq-demo/src/main/java/base/producer/SyncProducer.java
rocketmq-demo/src/main/java/base/producer/SyncProducer.java
+1
-1
rocketmq-demo/src/main/java/batch/Consumer.java
rocketmq-demo/src/main/java/batch/Consumer.java
+1
-1
rocketmq-demo/src/main/java/batch/Producer.java
rocketmq-demo/src/main/java/batch/Producer.java
+1
-1
rocketmq-demo/src/main/java/delay/Consumer.java
rocketmq-demo/src/main/java/delay/Consumer.java
+1
-1
rocketmq-demo/src/main/java/delay/Producer.java
rocketmq-demo/src/main/java/delay/Producer.java
+1
-1
rocketmq-demo/src/main/java/filter/sql/Consumer.java
rocketmq-demo/src/main/java/filter/sql/Consumer.java
+1
-1
rocketmq-demo/src/main/java/filter/sql/Producer.java
rocketmq-demo/src/main/java/filter/sql/Producer.java
+1
-1
rocketmq-demo/src/main/java/filter/tag/Consumer.java
rocketmq-demo/src/main/java/filter/tag/Consumer.java
+1
-1
rocketmq-demo/src/main/java/filter/tag/Producer.java
rocketmq-demo/src/main/java/filter/tag/Producer.java
+1
-1
rocketmq-demo/src/main/java/order/Consumer.java
rocketmq-demo/src/main/java/order/Consumer.java
+1
-1
rocketmq-demo/src/main/java/order/Producer.java
rocketmq-demo/src/main/java/order/Producer.java
+1
-1
rocketmq-demo/src/main/java/transaction/Consumer.java
rocketmq-demo/src/main/java/transaction/Consumer.java
+1
-1
rocketmq-demo/src/main/java/transaction/Producer.java
rocketmq-demo/src/main/java/transaction/Producer.java
+2
-1
springboot-rocketmq-consumer/src/main/resources/application.properties
...cketmq-consumer/src/main/resources/application.properties
+1
-1
springboot-rocketmq-producer/src/main/resources/application.properties
...cketmq-producer/src/main/resources/application.properties
+1
-1
未找到文件。
rocketmq-demo/pom.xml
浏览文件 @
f4cf9ed1
...
...
@@ -12,5 +12,12 @@
<maven.compiler.source>
8
</maven.compiler.source>
<maven.compiler.target>
8
</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>
org.apache.rocketmq
</groupId>
<artifactId>
rocketmq-client
</artifactId>
<version>
4.8.0
</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
rocketmq-demo/shuyu.iml
浏览文件 @
f4cf9ed1
...
...
@@ -11,5 +11,18 @@
</content>
<orderEntry
type=
"inheritedJdk"
/>
<orderEntry
type=
"sourceFolder"
forTests=
"false"
/>
<orderEntry
type=
"library"
name=
"Maven: org.apache.rocketmq:rocketmq-client:4.8.0"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: org.apache.rocketmq:rocketmq-common:4.8.0"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: org.apache.rocketmq:rocketmq-remoting:4.8.0"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: com.alibaba:fastjson:1.2.69"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: io.netty:netty-all:4.0.42.Final"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: org.apache.rocketmq:rocketmq-logging:4.8.0"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: io.netty:netty-tcnative-boringssl-static:1.1.33.Fork26"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: commons-validator:commons-validator:1.6"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: commons-beanutils:commons-beanutils:1.9.2"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: commons-digester:commons-digester:1.8.1"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: commons-logging:commons-logging:1.2"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: commons-collections:commons-collections:3.2.2"
level=
"project"
/>
<orderEntry
type=
"library"
name=
"Maven: org.apache.commons:commons-lang3:3.4"
level=
"project"
/>
</component>
</module>
\ No newline at end of file
rocketmq-demo/src/main/java/base/consumer/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -11,23 +11,23 @@ import java.util.List;
/**
* 消息的接受者
*
* @author : qinyingjie
* @version : 2.2.0
* @date : 2023/5/26 11:45
*/
public
class
Consumer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"base"
,
"*"
);
//设定消费模式:负载均衡|广播模式
consumer
.
setMessageModel
(
MessageModel
.
BROADCASTING
);
//4.设置回调函数,处理消息
consumer
.
registerMessageListener
(
new
MessageListenerConcurrently
()
{
//接受消息内容
@Override
public
ConsumeConcurrentlyStatus
consumeMessage
(
List
<
MessageExt
>
msgs
,
ConsumeConcurrentlyContext
context
)
{
...
...
rocketmq-demo/src/main/java/base/producer/AsyncProducer.java
浏览文件 @
f4cf9ed1
...
...
@@ -11,15 +11,13 @@ import java.util.concurrent.TimeUnit;
* 发送异步消息
*/
public
class
AsyncProducer
{
public
static
void
main
(
String
[]
args
)
throws
Exception
{
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
for
(
int
i
=
0
;
i
<
10
;
i
++)
{
//4.创建消息对象,指定主题Topic、Tag和消息体
/**
...
...
@@ -34,6 +32,7 @@ public class AsyncProducer {
* 发送成功回调函数
* @param sendResult
*/
@Override
public
void
onSuccess
(
SendResult
sendResult
)
{
System
.
out
.
println
(
"发送结果:"
+
sendResult
);
}
...
...
@@ -42,15 +41,14 @@ public class AsyncProducer {
* 发送失败回调函数
* @param e
*/
@Override
public
void
onException
(
Throwable
e
)
{
System
.
out
.
println
(
"发送异常:"
+
e
);
}
});
//线程睡1秒
TimeUnit
.
SECONDS
.
sleep
(
1
);
}
//6.关闭生产者producer
producer
.
shutdown
();
}
...
...
rocketmq-demo/src/main/java/base/producer/OneWayProducer.java
浏览文件 @
f4cf9ed1
...
...
@@ -15,7 +15,7 @@ public class OneWayProducer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/base/producer/SyncProducer.java
浏览文件 @
f4cf9ed1
...
...
@@ -16,7 +16,7 @@ public class SyncProducer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/batch/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -13,7 +13,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"BatchTopic"
,
"*"
);
...
...
rocketmq-demo/src/main/java/batch/Producer.java
浏览文件 @
f4cf9ed1
...
...
@@ -15,7 +15,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/delay/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -14,7 +14,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"DelayTopic"
,
"*"
);
...
...
rocketmq-demo/src/main/java/delay/Producer.java
浏览文件 @
f4cf9ed1
...
...
@@ -16,7 +16,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/filter/sql/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -14,7 +14,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"FilterSQLTopic"
,
MessageSelector
.
bySql
(
"i>5"
));
...
...
rocketmq-demo/src/main/java/filter/sql/Producer.java
浏览文件 @
f4cf9ed1
...
...
@@ -13,7 +13,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/filter/tag/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -13,7 +13,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"FilterTagTopic"
,
"Tag1 || Tag2 "
);
...
...
rocketmq-demo/src/main/java/filter/tag/Producer.java
浏览文件 @
f4cf9ed1
...
...
@@ -13,7 +13,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
...
...
rocketmq-demo/src/main/java/order/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -14,7 +14,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"OrderTopic"
,
"*"
);
...
...
rocketmq-demo/src/main/java/order/Producer.java
浏览文件 @
f4cf9ed1
...
...
@@ -14,7 +14,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
DefaultMQProducer
producer
=
new
DefaultMQProducer
(
"group1"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.启动producer
producer
.
start
();
//构建消息集合
...
...
rocketmq-demo/src/main/java/transaction/Consumer.java
浏览文件 @
f4cf9ed1
...
...
@@ -17,7 +17,7 @@ public class Consumer {
//1.创建消费者Consumer,制定消费者组名
DefaultMQPushConsumer
consumer
=
new
DefaultMQPushConsumer
(
"group1"
);
//2.指定Nameserver地址
consumer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
consumer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//3.订阅主题Topic和Tag
consumer
.
subscribe
(
"TransactionTopic"
,
"*"
);
...
...
rocketmq-demo/src/main/java/transaction/Producer.java
浏览文件 @
f4cf9ed1
package
transaction
;
import
org.apache.commons.lang3.StringUtils
;
import
org.apache.rocketmq.client.producer.*
;
import
org.apache.rocketmq.common.message.Message
;
import
org.apache.rocketmq.common.message.MessageExt
;
...
...
@@ -15,7 +16,7 @@ public class Producer {
//1.创建消息生产者producer,并制定生产者组名
TransactionMQProducer
producer
=
new
TransactionMQProducer
(
"group5"
);
//2.指定Nameserver地址
producer
.
setNamesrvAddr
(
"
192.168.25.135:9876;192.168.25.138
:9876"
);
producer
.
setNamesrvAddr
(
"
47.119.161.70:9876;47.119.163.226
:9876"
);
//添加事务监听器
producer
.
setTransactionListener
(
new
TransactionListener
()
{
...
...
springboot-rocketmq-consumer/src/main/resources/application.properties
浏览文件 @
f4cf9ed1
rocketmq.name-server
=
192.168.25.135:9876;192.168.25.138
:9876
rocketmq.name-server
=
47.119.161.70:9876;47.119.163.226
:9876
rocketmq.producer.group
=
my-group
\ No newline at end of file
springboot-rocketmq-producer/src/main/resources/application.properties
浏览文件 @
f4cf9ed1
rocketmq.name-server
=
192.168.25.135:9876;192.168.25.138
:9876
rocketmq.name-server
=
47.119.161.70:9876;47.119.163.226
:9876
rocketmq.producer.group
=
my-group
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录