【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的问题可以私信我一起研究

相关内容

热门资讯

安卓系统账号怎么改,轻松实现账... 你是不是也和我一样,在使用安卓手机的时候,突然觉得账号名字太老土了,想要来个焕然一新的改变呢?别急,...
离线语音系统安卓版下载,随时随... 你有没有想过,在手机上也能实现语音助手的功能,而且完全不需要联网?没错,就是那种离线语音系统,听起来...
华为os系统是基于安卓系统吗,... 你有没有想过,华为的手机里那个神秘的OS系统,它是不是就是安卓系统呢?别急,今天就来揭开这个谜底,让...
各品牌安卓系统多大,探索各大品... 你有没有想过,那些我们每天离不开的安卓手机,它们背后的系统到底有多大呢?这可不是一个小问题哦,因为系...
不是安卓系统电视机,探索非安卓... 你有没有想过,家里的电视是不是安卓系统呢?现在市面上,安卓系统电视机可是越来越流行了。但是,你知道吗...
谷歌安卓系统开源免费用,免费体... 你知道吗?在科技的世界里,有时候最让人惊喜的就是那些免费又好用的东西。今天,就让我来给你揭秘一个超级...
安卓电脑版怎么装系统,轻松实现... 你有没有想过,你的安卓电脑版突然间卡得像蜗牛一样,慢得让人抓狂?别急,今天就来教你怎么给它来个焕然一...
安卓系统有几种语音,揭秘多样化... 你知道吗?安卓系统里的语音功能可真是让人爱不释手呢!想象你只需要动动嘴,就能完成各种操作,是不是觉得...
扫码系统推荐安卓游戏,畅享指尖... 你有没有发现,最近手机里的游戏推荐越来越贴心了?没错,就是那个神奇的扫码系统!今天,就让我带你一探究...
手机系统安卓做的最好,探索最佳... 你有没有发现,现在手机市场上,安卓系统简直是要风得风,要雨得雨,简直是手机系统中的佼佼者!不信?那就...
安卓系统越拉越难用,越拉越难用... 你有没有发现,最近安卓系统越来越难用了?是不是每次打开手机,都感觉卡得要命,各种广告弹窗让人头疼不已...
卓安系统创始人,引领科技浪潮的... 你有没有想过,在这个信息爆炸的时代,谁才是那个站在风口浪尖,引领科技潮流的先锋?今天,我要带你认识一...
安卓10系统经常断网,安卓10... 最近是不是你也遇到了安卓10系统经常断网的问题?这可真是让人头疼啊!手机断网,就像失去了方向感的指南...
开发安卓系统用什么,从工具到实... 你有没有想过,为什么你的手机里装了那么多应用,却还是觉得不够用呢?其实,这一切都离不开一个强大的操作...
安卓系统如何双开抖音,安卓系统... 如何在安卓系统上双开抖音:让你的娱乐生活更加丰富多彩在数字化时代,手机已经成为我们生活中不可或缺的一...
安卓8.0系统内测地址,探索系... 你有没有听说最近安卓8.0系统内测的消息?这可是科技圈的大事啊!作为一个紧跟潮流的数码爱好者,我可是...
安卓系统可以账号同步吗,安卓系... 你有没有想过,你的安卓手机里的各种账户信息,比如微信、QQ、支付宝,能不能像变魔术一样,一键同步到另...
安卓系统的平板怎么下载,畅享应... 你有没有想过,拥有一款安卓系统的平板电脑,那感觉简直就像拥有了整个世界?想象随时随地都能畅游网络海洋...
win平板上装安卓系统咋样,开... 亲爱的读者们,你是否曾想过在Windows平板上体验安卓系统的魅力?想象那双熟悉的Windows界面...
给安卓车机写系统,安卓车机系统... 你有没有想过,你的安卓车机系统其实可以焕然一新?想象当你驾驶着爱车,车机界面流畅得像丝滑的巧克力,功...