Springboot整合RabbitMQ實(shí)現(xiàn)發(fā)送驗(yàn)證碼的示例代碼
1. RabbitMQ的介紹
- MQ全稱為Message Queue,即消息隊(duì)列, RabbitMQ是由erlang語言開發(fā),基于AMQP(Advanced Message Queue 高級(jí)消息隊(duì)列協(xié)議)協(xié)議實(shí)現(xiàn)的消息隊(duì)列,它是一種應(yīng)用程序之間的通信方法,消息隊(duì)列在分布式系統(tǒng)開 發(fā)中應(yīng)用非常廣泛。RabbitMQ官方地址:http://www.rabbitmq.com/
- 開發(fā)中消息隊(duì)列通常有如下應(yīng)用場景:
1、任務(wù)異步處理。 將不需要同步處理的并且耗時(shí)長的操作由消息隊(duì)列通知消息接收方進(jìn)行異步處理。提高了應(yīng)用程序的響應(yīng)時(shí)間。
2、應(yīng)用程序解耦合 MQ相當(dāng)于一個(gè)中介,生產(chǎn)方通過MQ與消費(fèi)方交互,它將應(yīng)用程序進(jìn)行解耦合。并且有如下優(yōu)點(diǎn)。
1.使得簡單,功能強(qiáng)大。
2.基于AMQP協(xié)議。
3.社區(qū)活躍,文檔完善。
4.高并發(fā)性能好,這主要得益于Erlang語言。
5.Spring Boot默認(rèn)已集成RabbitMQ
- 組成部分說明如下:
- Broker:消息隊(duì)列服務(wù)進(jìn)程,此進(jìn)程包括兩個(gè)部分:
- Exchange和Queue。
- Exchange:消息隊(duì)列交換機(jī),按一定的規(guī)則將消息路由轉(zhuǎn)發(fā)到某個(gè)隊(duì)列,對(duì)消息進(jìn)行過慮。
Queue:消息隊(duì)列,存儲(chǔ)消息的隊(duì)列,消息到達(dá)隊(duì)列并轉(zhuǎn)發(fā)給指定的消費(fèi)方。
Producer:消息生產(chǎn)者,即生產(chǎn)方客戶端,生產(chǎn)方客戶端將消息發(fā)送到MQ。
Consumer:消息消費(fèi)者,即消費(fèi)方客戶端,接收MQ轉(zhuǎn)發(fā)的消息。
消息發(fā)布接收流程:
-----發(fā)送消息-----
1、生產(chǎn)者和Broker建立TCP連接。
2、生產(chǎn)者和Broker建立通道。
3、生產(chǎn)者通過通道消息發(fā)送給Broker,由Exchange將消息進(jìn)行轉(zhuǎn)發(fā)。
4、Exchange將消息轉(zhuǎn)發(fā)到指定的Queue(隊(duì)列)
----接收消息-----
1、消費(fèi)者和Broker建立TCP連接
2、消費(fèi)者和Broker建立通道
3、消費(fèi)者監(jiān)聽指定的Queue(隊(duì)列)
4、當(dāng)有消息到達(dá)Queue時(shí)Broker默認(rèn)將消息推送給消費(fèi)者。
5、消費(fèi)者接收到消息
2. 搭建環(huán)境
- 實(shí)現(xiàn)原理:
- 在綁定(Binding)Exchange與Queue的同時(shí),一般會(huì)指定一個(gè)binding key;消費(fèi)者將消息發(fā)送給Exchange時(shí),一般會(huì)指定一個(gè)routing key;當(dāng)binding key與routing key相匹配時(shí),消息將會(huì)被路由到對(duì)應(yīng)的Queue中。
- 多個(gè)消費(fèi)者可以訂閱同一個(gè)Queue,這時(shí)Queue中的消息會(huì)被平均分?jǐn)偨o多個(gè)消費(fèi)者進(jìn)行處理,而不是每個(gè)消費(fèi)者都收到所有的消息并處理。
2.1引入jar包
<groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> </dependency> <!--redis--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>
2.2生產(chǎn)者配置
2.2.1Rabbit配置類
package com.cui.user.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** rabbitmq配置類 配置交換機(jī),消息隊(duì)列,并且綁定交換機(jī)和queue * @Author Cui * @Date 2020-4-9 14:55 */ @Configuration public class RabbitmqConfig { //隊(duì)列bean的名稱 cms 用來發(fā)送短信驗(yàn)證碼 public static final String QUEUE_INFORM_CMS= "queue_inform_cms"; //隊(duì)列bean的名稱 email 用來發(fā)送郵件 //public static final String QUEUE_INFORM_EMAIL= "queue_inform_email"; //交換機(jī)的名稱 public static final String EXCHANGE_TOPIC_INFORM_="exchange_topic_inform"; //隊(duì)列的名稱 @Value("${cxp.mq.queue}") public String queue_cms_postpage_name; //routingKey @Value("${cxp.mq.routingKey}") public String routingKey; /** * 交換機(jī)配置使用direct類型 * @return the exchange */ @Bean(EXCHANGE_TOPIC_INFORM_) public Exchange EXCHANGE_TOPICS_INFORM() { //durable(true) 持久化,mq重啟之后交換機(jī)還在 return ExchangeBuilder.directExchange(EXCHANGE_TOPIC_INFORM_).durable(true).build(); } //聲明隊(duì)列 @Bean(QUEUE_INFORM_CMS) public Queue QUEUE_CMS_POSTPAGE() { Queue queue = new Queue(QUEUE_INFORM_CMS); return queue; * 綁定隊(duì)列到交換機(jī) * * @param queue the queue * @param exchange the exchange * @return the binding @Bean public Binding BINDING_QUEUE_INFORM_SMS(@Qualifier(QUEUE_INFORM_CMS) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM_) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs(); }
2.2.2 application.yml文件配置
server: port: ${PORT:8002} spring: application: name: cxp-service-manage-user #Redis配置 redis: host: 127.0.0.1 port: 6379 jedis: pool: max-active: 8 max-wait: -1 max-idle: 500 min-idle: 0 lettuce: shutdown-timeout: 0 datasource: url: jdbc:mysql://localhost:3306/system_user?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC username: root password: 123456 driver-class-name: com.mysql.jdbc.Driver rabbitmq: port: 5672 username: guest password: guest virtualHost: / cxp: mq: #cms客戶端監(jiān)控的隊(duì)列名稱(不同的客戶端監(jiān)控的隊(duì)列不能重復(fù)) queue: queue_inform_cms routingKey: inform.#.sms.# #此routingKey郵件消費(fèi)者和信息消費(fèi)者通用 mybatis: mapper-locations: classpath:mapper/*Mapper.xml type-aliases-package: com.cui.model.entity.user mapper: mappers: com.cui.model.BaseMapper #通用基類配置 identity: mysql pagehelper: helperDialect: mysql reasonable: true supportMethodsArguments: true params: count=countSql eureka: client: registerWithEureka: true #服務(wù)注冊(cè)開關(guān) fetchRegistry: true #服務(wù)發(fā)現(xiàn)開關(guān) serviceUrl: #Eureka客戶端與Eureka服務(wù)端進(jìn)行交互的地址,多個(gè)中間用逗號(hào)分隔 defaultZone: ${EUREKA_SERVER:http://localhost:50101/eureka/,http://localhost:50102/eureka/} instance: prefer-ip-address: true #將自己的ip地址注冊(cè)到Eureka服務(wù)中 ip-address: ${IP_ADDRESS:127.0.0.1} instance-id: ${spring.application.name}:${server.port} #指定實(shí)例id ribbon: MaxAutoRetries: 2 #最大重試次數(shù),當(dāng)Eureka中可以找到服務(wù),但是服務(wù)連不上時(shí)將會(huì)重試,如果eureka中找不到服務(wù)則直接走斷路器 MaxAutoRetriesNextServer: 3 #切換實(shí)例的重試次數(shù) OkToRetryOnAllOperations: false #對(duì)所有操作請(qǐng)求都進(jìn)行重試,如果是get則可以,如果是post,put等操作沒有實(shí)現(xiàn)冪等的情況下是很危險(xiǎn)的,所以設(shè)置為false ConnectTimeout: 5000 #請(qǐng)求連接的超時(shí)時(shí)間 ReadTimeout: 6000 #請(qǐng)求處理的超時(shí)時(shí)間
2.3消費(fèi)者配置
引入jar包,這里需引入阿里云通信多的jar包和Redis的jar包
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!--阿里云通信--> <dependency> <groupId>com.aliyun</groupId> <artifactId>aliyun-java-sdk-core</artifactId> <version>4.4.0</version> <artifactId>aliyun-java-sdk-dysmsapi</artifactId> <version>1.0.0</version> <!-- 導(dǎo)入Eureka客戶端的依賴 --> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> <!-- feign相關(guān)依賴 --> <artifactId>spring-cloud-starter-openfeign</artifactId> <dependency> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <artifactId>spring-boot-starter-amqp</artifactId>
2.3.1 消費(fèi)者配置類(同生產(chǎn)者)
package com.cui.sms.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** rabbitmq配置類 配置交換機(jī),消息隊(duì)列,并且綁定交換機(jī)和queue * @Author Cui * @Date 2020-4-9 14:55 */ @Configuration public class RabbitmqConfig { //隊(duì)列bean的名稱 cms 用來發(fā)送短信驗(yàn)證碼 public static final String QUEUE_INFORM_CMS= "queue_inform_cms"; //隊(duì)列bean的名稱 email 用來發(fā)送郵件 //public static final String QUEUE_INFORM_EMAIL= "queue_inform_email"; //交換機(jī)的名稱 public static final String EXCHANGE_TOPIC_INFORM_="exchange_topic_inform"; //隊(duì)列的名稱 @Value("${cxp.mq.queue}") public String queue_cms_postpage_name; //routingKey @Value("${cxp.mq.routingKey}") public String routingKey; /** * 交換機(jī)配置使用direct類型 * @return the exchange */ @Bean(EXCHANGE_TOPIC_INFORM_) public Exchange EXCHANGE_TOPICS_INFORM() { //durable(true) 持久化,mq重啟之后交換機(jī)還在 return ExchangeBuilder.directExchange(EXCHANGE_TOPIC_INFORM_).durable(true).build(); } //聲明隊(duì)列 @Bean(QUEUE_INFORM_CMS) public Queue QUEUE_CMS_POSTPAGE() { Queue queue = new Queue(QUEUE_INFORM_CMS); return queue; * 綁定隊(duì)列到交換機(jī) * * @param queue the queue * @param exchange the exchange * @return the binding @Bean public Binding BINDING_QUEUE_INFORM_SMS(@Qualifier(QUEUE_INFORM_CMS) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM_) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs(); }
2.3.2 application.yml文件配置
server: port: 8103 spring: application: name: cxp-manager-service-sms rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest virtualHost: / #Redis配置 redis: port: 6379 password: 123456 jedis: pool: max-active: 8 max-wait: -1 max-idle: 500 min-idle: 0 lettuce: shutdown-timeout: 0 aliyun: sms: accessKeyId: XXXXXXXXXXXXXXXXXXXX accessKeySecret: XXXXXXXXXXXXXXXXXXXX template_code: XXXXXXXXXXX sign_name: XXXX cxp: mq: #cms客戶端監(jiān)控的隊(duì)列名稱(不同的客戶端監(jiān)控的隊(duì)列不能重復(fù)) queue: queue_inform_cms routingKey: inform.sms #此routingKey用來監(jiān)聽信息 eureka: client: registerWithEureka: true #服務(wù)注冊(cè)開關(guān) fetchRegistry: true #服務(wù)發(fā)現(xiàn)開關(guān) serviceUrl: #Eureka客戶端與Eureka服務(wù)端進(jìn)行交互的地址,多個(gè)中間用逗號(hào)分隔 defaultZone: ${EUREKA_SERVER:http://localhost:50101/eureka/,http://localhost:50102/eureka/} instance: prefer-ip-address: true #將自己的ip地址注冊(cè)到Eureka服務(wù)中 ip-address: ${IP_ADDRESS:127.0.0.1} instance-id: ${spring.application.name}:${server.port} #指定實(shí)例id ribbon: MaxAutoRetries: 2 #最大重試次數(shù),當(dāng)Eureka中可以找到服務(wù),但是服務(wù)連不上時(shí)將會(huì)重試,如果eureka中找不到服務(wù)則直接走斷路器 MaxAutoRetriesNextServer: 3 #切換實(shí)例的重試次數(shù) OkToRetryOnAllOperations: false #對(duì)所有操作請(qǐng)求都進(jìn)行重試,如果是get則可以,如果是post,put等操作沒有實(shí)現(xiàn)冪等的情況下是很危險(xiǎn)的,所以設(shè)置為false ConnectTimeout: 5000 #請(qǐng)求連接的超時(shí)時(shí)間 ReadTimeout: 6000 #請(qǐng)求處理的超時(shí)時(shí)間
3.寫發(fā)送短信驗(yàn)證碼的代碼
3.1寫一個(gè)controller來調(diào)用發(fā)送驗(yàn)證碼的接口
/** * 發(fā)送短信驗(yàn)證碼 * @param phone * @return */ @ApiOperation(value = "發(fā)送短信驗(yàn)證碼",notes = "發(fā)送短信驗(yàn)證碼") @GetMapping("/sendSms") public ResponseResult sendSms(String phone){ LOGGER.info("要發(fā)送的手機(jī)號(hào)為:{}", phone); userService.sendSms(phone); return new ResponseResult(UserMsg.SUCCESS.getMsgCd(), UserMsg.SUCCESS.getMsgInfo()); }
3.2 生成驗(yàn)證碼
后臺(tái)生成六位數(shù)的隨機(jī)驗(yàn)證碼,并且將驗(yàn)證碼存入Redis中,設(shè)置五分鐘的過期時(shí)間(用于用戶注冊(cè)時(shí)的校對(duì)),將驗(yàn)證碼存到RabbitMQ中,當(dāng)調(diào)用發(fā)送接口時(shí),生產(chǎn)端將信息發(fā)送到綁定的隊(duì)列中。
/** * 向注冊(cè)用戶發(fā)送發(fā)送驗(yàn)證碼 * @param phone 注冊(cè)的用戶的手機(jī)號(hào) */ @Override public void sendSms(String phone) { //1.生成六位隨機(jī)驗(yàn)證碼 Random random = new Random();//隨機(jī)函數(shù) int code = random.nextInt(999999);//設(shè)置隨機(jī)數(shù)的最大值 if(code<100000){ //如果驗(yàn)證碼小于六位數(shù),加100000保證驗(yàn)證碼為6位數(shù) code+=100000; } //System.out.println("短信驗(yàn)證碼:"+code); LOGGER.info("生成的短信驗(yàn)證碼為:{{}}", code); //2.將驗(yàn)證碼存入redis redisTemplate.boundValueOps("code_"+phone).set(code+""); redisTemplate.boundValueOps("code_"+phone).expire(5, TimeUnit.MINUTES);//設(shè)置驗(yàn)證碼五分鐘到期 //3.將驗(yàn)證碼存入RabbitMQ Map<String,String> map = new HashMap<String, String>(); map.put("phone", phone); map.put("code", code+""); //以json格式存到RabbitMQ消息隊(duì)列中 rabbitTemplate.convertAndSend(EXCHANGE_TOPIC_INFORM_, routingKey, JSON.toJSONString(map)); }
3.3發(fā)送短信驗(yàn)證碼
在RabbitMQ的消費(fèi)者端監(jiān)聽短信的routingKey ,當(dāng)收到生產(chǎn)端發(fā)來的消息后,便會(huì)調(diào)用阿里云通信向用戶發(fā)送短信
package com.cui.sms.mq; import com.alibaba.fastjson.JSON; import com.aliyuncs.CommonResponse; import com.cui.sms.utils.SmsUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import java.util.Map; /** * @Author Cui * @Date 2020-4-9 15:40 * 監(jiān)聽MQ,發(fā)送短信驗(yàn)證碼 */ @Component public class SmsMessageConsumer { private static final Logger LOGGER = LoggerFactory.getLogger(SmsMessageConsumer.class); @Autowired private SmsUtil smsUtil; @Value("${aliyun.sms.template_code}") private String templateCode; @Value("${aliyun.sms.param}") private String param; //短信參數(shù) @RabbitListener(queues = {"${cxp.mq.queue}"}) public void onMessage(Message message) { String jsonString= new String(message.getBody());//得到mq中存入的json格式的消息 Map<String,String> map = JSON.parseObject(jsonString, Map.class);//將json格式轉(zhuǎn)換為Map格式 String phone = map.get("phone");//mq中存入的手機(jī)號(hào) String code = map.get("code");//mq中存入的驗(yàn)證碼 //System.out.println("手機(jī)號(hào)"+phone+"驗(yàn)證碼"+code); LOGGER.info("發(fā)送的手機(jī)號(hào)為:{} ,發(fā)送的驗(yàn)證碼為 :{}",phone, code); //調(diào)用阿里云通信 CommonResponse commonResponse = smsUtil.sendSms(phone, templateCode, param.replace("[value]", code)); } }
3.4 實(shí)現(xiàn)驗(yàn)證碼的校對(duì)
用戶收到驗(yàn)證碼并且填寫完相應(yīng)的信息后,點(diǎn)擊注冊(cè),將自己的信息發(fā)送到后臺(tái),后臺(tái)收到信息后,取出存在Redis中的驗(yàn)證碼,和用戶的驗(yàn)證碼進(jìn)行比較,然后將結(jié)果返回給前端。代碼如下所示:
@PostMapping("/save") @ApiOperation(value = "新增用戶",notes = "新增用戶") public ResponseResult add(@RequestBody User user, String smsCode){ LOGGER.info("新增的用戶的信息為:{},用戶收到的驗(yàn)證碼為:{}", user.toString(),smsCode); //對(duì)用戶密碼進(jìn)行加密后在存入數(shù)據(jù)庫 BCryptPasswordEncoder encoder = new BCryptPasswordEncoder(); String newPassword = encoder.encode(user.getPassword()); user.setPassword(newPassword); userService.add(user,smsCode ); return new ResponseResult(UserMsg.SUCCESS.getMsgCd(), UserMsg.SUCCESS.getMsgInfo()); }
/** * 用戶注冊(cè) * @param user 用戶對(duì)象信息 * @param smsCode 短信驗(yàn)證碼 */ @Override public void add(User user, String smsCode) { //獲取系統(tǒng)驗(yàn)證碼 String sysCode = (String) redisTemplate.boundValueOps("code_" + user.getPhone()).get(); //比較短信驗(yàn)證碼 LOGGER.info("從Redis中取到的短信驗(yàn)證碼為:{{}}",smsCode+" 用戶收到的的短信驗(yàn)證碼為:{{}}",smsCode); if(sysCode==null||"".equals(smsCode)){ throw new RuntimeException("驗(yàn)證碼未發(fā)送或已過期!請(qǐng)稍后重試"); } if(!smsCode.equals(sysCode)){ throw new RuntimeException("驗(yàn)證碼不正確,請(qǐng)重新輸入!"); } if(user.getUsername()==null){ user.setUsername(user.getPhone()); } User searchUser = new User(); //將用戶傳來的手機(jī)號(hào)傳給searchUser,去查詢數(shù)據(jù)庫中是否存在該手機(jī)號(hào) searchUser.setPhone(user.getPhone()); if(userDao.selectCount(searchUser)>0){ throw new RuntimeException("該手機(jī)號(hào)已被注冊(cè)!"); } //設(shè)置user的其他參數(shù) user.setCreated(new Date()); user.setUpdated(new Date()); user.setPoints(0);//積分初始值為0 user.setStatus("1");//狀態(tài)1 user.setIsEmailCheck("0");//郵箱認(rèn)證 user.setIsMobileCheck("1");//手機(jī)認(rèn)證 userDao.insert(user); }
到此這篇關(guān)于Springboot整合RabbitMQ實(shí)現(xiàn)發(fā)送驗(yàn)證碼的功能的文章就介紹到這了,更多相關(guān)Springboot整合RabbitMQ發(fā)送驗(yàn)證碼內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Spring Boot整合MyBatis連接Oracle數(shù)據(jù)庫的步驟全紀(jì)錄
這篇文章主要給大家介紹了關(guān)于Spring Boot整合MyBatis連接Oracle數(shù)據(jù)庫的相關(guān)資料,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2018-07-07SpringBoot配置數(shù)據(jù)庫密碼加密的實(shí)現(xiàn)
這篇文章主要介紹了SpringBoot配置數(shù)據(jù)庫密碼加密的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03Java畢業(yè)設(shè)計(jì)實(shí)戰(zhàn)之食品溯源系統(tǒng)的實(shí)現(xiàn)
這是一個(gè)使用了java+Springboot+Maven+mybatis+Vue+mysql+wd開發(fā)的食品溯源系統(tǒng),是一個(gè)畢業(yè)設(shè)計(jì)的實(shí)戰(zhàn)練習(xí),具有食品溯源該有的所有功能,感興趣的朋友快來看看吧2022-01-01Mybatis?resultMap標(biāo)簽繼承、復(fù)用、嵌套方式
這篇文章主要介紹了Mybatis?resultMap標(biāo)簽繼承、復(fù)用、嵌套方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-03-03java網(wǎng)絡(luò)編程基礎(chǔ)知識(shí)介紹
這篇文章主要介紹了java網(wǎng)絡(luò)編程基礎(chǔ)知識(shí)介紹,涉及OSI分層模型和TCP/IP分層模型的對(duì)應(yīng)關(guān)系、IP地址、端口號(hào)、tcp、udp等相關(guān)內(nèi)容,還是比較不錯(cuò)的,這里分享給大家,供需要的朋友參考。2017-11-11詳解Java中的do...while循環(huán)語句的使用方法
這篇文章主要介紹了Java中的do...while循環(huán)語句的使用方法,是Java入門學(xué)習(xí)中的基礎(chǔ)知識(shí),需要的朋友可以參考下2015-10-10