【Java】最新版本SpringCloudStream整合RocketMQ实现单项目中事件的发布与监听
创始人
2024-05-30 19:13:45
0

文章目录

  • 前言
  • 依赖配置
  • 代码

参考

前言

SpringCloud项目中整合RocketMQ是为了削峰填谷。
这里我使用RocketMQ的作用用于接收项目中产生的消息,然后异步的发送邮件给客户,这是这个项目的产生的背景。

依赖配置

com.alibaba.cloudspring-cloud-starter-bus-rocketmqcom.alibaba.cloudspring-cloud-starter-stream-rocketmqorg.apache.rocketmqrocketmq-clientorg.apache.rocketmqrocketmq-aclcom.alibaba.cloudspring-cloud-starter-bus-rocketmqorg.apache.rocketmqrocketmq-client4.9.4org.apache.rocketmqrocketmq-acl4.9.4

项目导入上面依赖之后即可开始代码的编写

代码

然后让我们先看一眼配置文件

# Tomcat
server:port: 9201# Spring
spring:application:# 应用名称name: towelove-systemprofiles:# 环境配置active: devcloud:nacos:discovery:# 服务注册地址server-addr: localhost:8848config:# 配置中心地址server-addr: localhost:8848# 配置文件格式file-extension: yaml# 共享配置shared-configs[0]:data-id: towelove-base-dev.yamlrefresh: trueshared-configs[1]:data-id: towelove-mysql-dev.yamlrefresh: trueshared-configs[2]:data-id: towelove-redis-dev.yamlrefresh: true# Spring Cloud Stream 配置项,对应 BindingServiceProperties 类stream:function:definition: mailSendConsumer;sendSmsToAdmin;sendSmsToUser; # 需要确保消费者类的名称和这里一样# Binding 配置项,对应 BindingProperties Mapbindings:sendSmsToAdmin-out-0: # 配置生产者destination: admin_sms_sendsendSmsToAdmin-in-0:destination: admin_sms_sendgroup: system_sms_send_consumer_groupsendSmsToUser-out-0: # 配置生产者destination: admin_sms_sendsendSmsToUser-in-0:destination: admin_sms_sendgroup: system_sms_send_consumer_group
#        smsSendConsumer-in-0: # 配置消费者
#          destination: admin_sms_send
#          group: system_sms_send_consumer_group#        smsSend-out-1:
#          destination: user_sms_send
#        smsSendConsumer-in-1:
#          destination: user_sms_send
#          group: system_sms_send_consumer_groupmailSend-out-0:destination: system_mail_sendmailSendConsumer-in-0: # 需要确保消费者类的名称和这里一样destination: system_mail_sendgroup: system_mail_send_consumer_group# Spring Cloud Stream RocketMQ 配置项rocketmq:# RocketMQ Binder 配置项,对应 RocketMQBinderConfigurationProperties 类binder:name-server: 192.168.146.115:9876 # RocketMQ Namesrv 地址#          access-key: # 用户名#          secret-key:  # 密码default: # 默认 bindings 全局配置producer: # RocketMQ Producer 配置项,对应 RocketMQProducerProperties 类group: system_producer_group # 生产者分组send-type: SYNC # 发送模式,SYNC 同步# 如果你项目里只对接一个中间件,那么不用定义binders# 当系统要定义多个不同消息中间件的时候,使用binders定义#      binders:#        my-rocketmq:#          type: rocketmq#          environment:#            rocketmq:#              name-server: 192.168.146.115:9876#          access-key: # 用户名#          secret-key:  # 密码# Spring Cloud Bus 配置项,对应 BusProperties 类bus:enabled: true # 是否开启,默认为 trueid: ${spring.application.name}:${server.port} # 编号,Spring Cloud Alibaba 建议使用“应用:端口”的格式destination: springCloudBus # 目标消息队列,默认为 springCloudBus

这里我截取了比较重要的配置,然后下面进行配置的讲解
在这里插入图片描述
首先就是我写了特别多注释的一个spring.cloud.stream.function.definition
这个东西是什么作用呢?
我的理解是,它用来声明你当前项目中的消费者,以及消费者类中的方法。
然后就是spring.cloud.stream.bindings中的好多个xxx-out-0和xxx-in-0
其中out对应的项目的输出,也就是消息的产生,对应的就是项目中的生产者,生产者发送消息的需要指定对应的信道,也就是你要告诉他往哪里发,其实就是对应的broker(再RocketMQ里面是这样子的),并且设定你发往的这个broker对应的topic,也就是destination。
那么同理,当生产者吧消息发送到broker中对应的topic后,我们就需要消费者去消费这个消息了。
那么此时就是使用in标签。
in标签里面的destination表示的也就是当前消费者需要去消费哪一个topic里面的消息。
你可能有一个疑问就是,那么为什么不用去指定对应的broker呢?
下面就是讲解这个in和out标签的声明的规则。
其实这也是一种约定优于配置的思想。
其中functionName就是你的消费者的类名或者你要提供消费的方法。
在命名规则的最后还有一个 index,它是 input 和 output 的序列,如果同一个 function name 只有一个 output 和一个 input,那么这个 index 永远都是 0。而如果你需要为一个 function 添加多个 input 和 output,就需要使用 index 变量来区分每个生产者消费者了。
Input 信道(消费者):< functionName > - in - < index >;

Output 信道(生产者):< functionName > - out - < index >。

讲解完这些,你大概就理解了这里的代码是为什么这么编写了。
那么下面我引入具体的业务代码。
我们从底层向上。
首先是消息的实体类。


@Data
public class SmsSendMessage {/*** 邮件日志编号*/@NotNull(message = "邮件日志编号不能为空")private Long logId;/*** 接收邮件地址*/@NotNull(message = "电话号码不能为空")private String phonenumber;/*** 邮件账号编号*/@NotNull(message = "邮件账号编号不能为空")private Long accountId;/*** 邮件发件人*/private String nickname;/*** 邮件标题*/@NotEmpty(message = "邮件标题不能为空")private String title;/*** 邮件内容*/@NotEmpty(message = "邮件内容不能为空")private String content;private Boolean isHtml;private File[] files;}

这个是消息的生产者

@Slf4j
@Service
public class SmsProducer {@Autowiredprivate StreamBridge streamBridge;public void sendSmsToAdmin(SmsSendMessage message) {log.info("要发送的短信内容为: {}", message);streamBridge.send("sendSmsToAdmin-out-0", message);}public void sendSmsToUser(Long userId,Long accountId) {log.info("要发送的短信内容为: {}", "userId:"+userId+"accountId:"+accountId);streamBridge.send("sendSmsToUser-out-0",  "userId:"+userId+"  accountId:"+accountId);}}

然后就是控制层

@RestController
@RequestMapping("/sys/sms")
public class SmsController {@Autowiredprivate SmsProducer smsProducer;@PostMapping("/send/admin")public R sendSmsToAdmin(@RequestBody @Valid SmsSendMessage message){smsProducer.sendSmsToAdmin(message);return R.ok();}@PostMapping("/send/user")public R sendSmsToUser(@RequestParam("userId")Long userId,@RequestParam("accountId")Long accountId){smsProducer.sendSmsToUser(userId,accountId);return R.ok();}}

然后下面是事件消费者的第一种写法

@Component
@Slf4j
public class SmsSendConsumer //implements Consumer
{//@Override//public void accept(SmsSendMessage message) {//    System.out.println(message);//}@Beanpublic Consumer sendSmsToAdmin() {return reqest -> {log.info("received: {} ", reqest);};}@Beanpublic Consumer sendSmsToUser(){return request -> {log.info("received: {}", request);List params = Arrays.stream(request.split(",")).map(Long::valueOf).collect(Collectors.toList());System.out.println(params);};}}

简单的介绍一下代码的逻辑,
其实就是我们向控制层发送一个请求并且携带上一些参数之后,控制层让生产者发送一个消息到对应的消息队列中。
发现了吗,这里消息的生产者发送的消息的目的地,就是我们设定的out标签。
在这里插入图片描述
那么消费者如何知道要去消费消息呢?
这就是为什么上面我说function.definition和in标签的作用了。
in标签这里的前缀就是我们的方法名,也就是对应的broker中的topic有消息后,对应的消费者会把消息拉过来,然后进行消费,而他之所以能知道要去消费哪一个消息也就是因为这里的绑定好的原因。
所以如果你一个类中声明了多个的消费方法,只需要再function.definition这个地方声明出你方法的名称,并且再代码里面使用@Bean的方式去声明出对应的方法即可
也就是如下图一样。
在这里插入图片描述
那么好奇的你可能会发现,这样子可以定义多个方法,还挺不错的,就是好像有点麻烦欸,要写的东西一下子就多了。
所以,如果你的消费者类只有一个方法,也就是你当前要消费的消费者只需要提供唯一的方法,那么我们可以把function.definition这里的方法名编写为消费者类的名称。
也就是下面这种代码的方式
在这里插入图片描述
而我们的生产者还是一样,只要确保其发送消息的信道是确定的即可
在这里插入图片描述
那么以这两种方式,如果你的消费者需要提供多个方法,那么就使用第一种方式,而如果你的消费者是单一的,只需要提供某一种方法,那么直接使用第二种方法去实现某个类即可。

当然,两种方式可以混合在一起实现
如果你在你的代码中出现了下图的问题
在这里插入图片描述
可以查看我下面这篇文章

解决上图的问题

类似的springcloudstream整合rocketmq的问题可以私信我一起研究

相关内容

热门资讯

安卓系统安装gps模块,安卓系... 你有没有想过,为什么你的安卓手机在户外旅行时总是能精准地告诉你位置呢?这背后可大有学问呢!今天,就让...
安卓手机如何修复系统,安卓手机... 手机突然卡顿,系统崩溃,是不是让你头疼不已?别担心,今天就来教你几招,让你的安卓手机焕然一新!一、重...
系统更新华为安卓,引领智能科技... 亲爱的手机控们,是不是最近你的华为手机突然变得有点儿“慢吞吞”了呢?别急,这可不是你的错觉哦!华为安...
安卓系统拦截怎么解除,轻松恢复... 你是不是也遇到了安卓系统拦截的问题,心里直痒痒想要解除它呢?别急,今天就来给你详细说说怎么搞定这个小...
老式平板装安卓系统,重温经典体... 你有没有想过,那些曾经陪伴我们度过无数时光的老式平板电脑,竟然还能装上安卓系统呢?这可不是什么天方夜...
系统如何与安卓互通,技术融合与... 你有没有想过,你的手机系统竟然能和安卓系统这么默契地互通有无?这就像是一场跨越科技界的友谊赛,让我们...
安卓系统 扫码枪,安卓系统下扫... 你有没有想过,在繁忙的超市收银台,那些快速流畅的扫码操作,背后其实隐藏着一个小小的英雄——安卓系统扫...
平板插卡推荐安卓系统,安卓系统... 你有没有想过,你的平板电脑是不是也能像智能手机一样,随时随地扩充存储空间呢?没错,这就是今天我要跟你...
安卓系统固件安装失败,原因排查... 最近是不是你也遇到了安卓系统固件安装失败的问题?别急,让我来给你详细说说这个让人头疼的小麻烦,让你一...
ios系统和安卓区别,系统差异... 你有没有发现,现在手机市场上,iOS系统和安卓系统就像是一对双胞胎,长得差不多,但性格却截然不同。今...
安卓系统2.3优酷,优酷的崛起... 你有没有发现,安卓系统2.3时代的那股怀旧风?那时候,优酷可是视频界的巨头,多少人都是看着优酷长大的...
安卓导航系统密封,安卓导航系统... 你有没有发现,现在手机导航系统越来越智能了?尤其是安卓系统的导航,简直就像一个贴心的导航小助手,带你...
a版安卓11系统,a版深度解析... 你知道吗?最近手机界可是炸开了锅,各大品牌纷纷发布了搭载a版安卓11系统的手机。这可不是什么小打小闹...
安卓系统的模拟吉他,随时随地弹... 你有没有想过,在手机上也能弹奏吉他呢?没错,就是那种模拟吉他的安卓系统应用,让你随时随地都能享受音乐...
王者适配的安卓系统,深度解析适... 你有没有发现,最近玩《王者荣耀》的小伙伴们都在议论纷纷,说新出的安卓系统简直是为王者量身定做的!没错...
安卓系统自动定位关闭,隐私保护... 你有没有发现,手机里的安卓系统有时候会自动定位,这可真是让人又爱又恨啊!有时候,我们并不想让别人知道...
安卓系统电量耗尽测试,全面解析... 手机电量耗尽,这可是每个手机用户都头疼的问题。你有没有想过,你的安卓手机在电量耗尽前,到底经历了哪些...
如何升级车载安卓系统,车载安卓... 亲爱的车主朋友们,你是不是也和我一样,对车载安卓系统升级这件事充满了好奇和期待呢?想象当你驾驶着爱车...
安卓办公哪个系统好,深度解析哪... 你有没有想过,在安卓办公的世界里,哪个系统才是你的最佳拍档呢?在这个信息爆炸的时代,选择一个既强大又...
安卓系统差劲怎么解决,重拾流畅... 你有没有发现,安卓系统有时候真的让人头疼得要命?手机卡顿、应用崩溃、电池续航短,这些问题是不是让你抓...