网站建设公司做销售好不好？,网站设计制作怎样可以快速,网站建设的描述,有谁知道知乎网站是谁做的文章目录前言依赖配置代码参考前言
SpringCloud项目中整合RocketMQ是为了削峰填谷。 这里我使用RocketMQ的作用用于接收项目中产生的消息#xff0c;然后异步的发送邮件给客户#xff0c;这是这个项目的产生的背景。
依赖配置
dependenciesdependency…
文章目录前言依赖配置代码参考前言
SpringCloud项目中整合RocketMQ是为了削峰填谷。 这里我使用RocketMQ的作用用于接收项目中产生的消息然后异步的发送邮件给客户这是这个项目的产生的背景。
依赖配置
dependenciesdependencygroupIdcom.alibaba.cloud/groupId!-- 引入基于 RocketMQ 的 Spring Cloud Bus 的实现的依赖并实现对其的自动配置 --artifactIdspring-cloud-starter-bus-rocketmq/artifactId/dependencydependencygroupIdcom.alibaba.cloud/groupIdartifactIdspring-cloud-starter-stream-rocketmq/artifactIdexclusionsexclusiongroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactId/exclusionexclusiongroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-acl/artifactId/exclusion/exclusions/dependencydependencygroupIdcom.alibaba.cloud/groupId!-- 引入基于 RocketMQ 的 Spring Cloud Bus 的实现的依赖并实现对其的自动配置 --artifactIdspring-cloud-starter-bus-rocketmq/artifactId/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.9.4/version/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-acl/artifactIdversion4.9.4/version/dependency/dependencies项目导入上面依赖之后即可开始代码的编写
代码
然后让我们先看一眼配置文件
# 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:userIdaccountId:accountId);streamBridge.send(sendSmsToUser-out-0, userId:userId accountId:accountId);}}
然后就是控制层
RestController
RequestMapping(/sys/sms)
public class SmsController {Autowiredprivate SmsProducer smsProducer;PostMapping(/send/admin)public RBoolean sendSmsToAdmin(RequestBody Valid SmsSendMessage message){smsProducer.sendSmsToAdmin(message);return R.ok();}PostMapping(/send/user)public RBoolean sendSmsToUser(RequestParam(userId)Long userId,RequestParam(accountId)Long accountId){smsProducer.sendSmsToUser(userId,accountId);return R.ok();}}
然后下面是事件消费者的第一种写法
Component
Slf4j
public class SmsSendConsumer //implements ConsumerSmsSendMessage
{//Override//public void accept(SmsSendMessage message) {// System.out.println(message);//}Beanpublic ConsumerString sendSmsToAdmin() {return reqest - {log.info(received: {} , reqest);};}Beanpublic ConsumerString sendSmsToUser(){return request - {log.info(received: {}, request);ListLong 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的问题可以私信我一起研究
