SpringBoot 中使用RabbtiMq?詳解
前言
如圖使用redisTemplate 一樣的簡單方便
模擬發(fā)送郵件的情況
pom.xml
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency>
application.properties
spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.host=192.168.91.128 spring.rabbitmq.port=5672 ## 根據(jù)自己情況而定,可以不用 spring.rabbitmq.listener.simple.acknowledge-mode=manual spring.rabbitmq.listener.simple.prefetch=100
寫在配置文件中,由 RabbitProperties 這個(gè)類進(jìn)行讀取,封裝到ConnectionFactory 中。
MailConstants (常量)
public class MailConstants { public static final Integer DELIVERING = 0;//消息投遞中 public static final Integer SUCCESS = 1;//消息投遞成功 public static final Integer FAILURE = 2;//消息投遞失敗 public static final Integer MAX_TRY_COUNT = 3;//最大重試次數(shù) public static final Integer MSG_TIMEOUT = 1;//消息超時(shí)時(shí)間 public static final String MAIL_QUEUE_NAME = "javaboy.mail.queue"; public static final String MAIL_EXCHANGE_NAME = "javaboy.mail.exchange"; public static final String MAIL_ROUTING_KEY_NAME = "javaboy.mail.routing.key"; }
RabbitConfig (rabbitMq的配置類)
import org.javaboy.vhr.model.MailConstants; import org.javaboy.vhr.service.MailSendLogService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitConfig { public final static Logger logger = LoggerFactory.getLogger(RabbitConfig.class); @Autowired CachingConnectionFactory cachingConnectionFactory; //發(fā)送郵件的 @Autowired MailSendLogService mailSendLogService; @Bean RabbitTemplate rabbitTemplate() { RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory); //手動(dòng)應(yīng)答返回的標(biāo)志 rabbitTemplate.setConfirmCallback((data, ack, cause) -> { String msgId = data.getId(); if (ack) { logger.info(msgId + ":消息發(fā)送成功"); mailSendLogService.updateMailSendLogStatus(msgId, 1);//修改數(shù)據(jù)庫中的記錄,消息投遞成功 } else { logger.info(msgId + ":消息發(fā)送失敗"); } }); rabbitTemplate.setReturnCallback((msg, repCode, repText, exchange, routingkey) -> { logger.info("消息發(fā)送失敗"); }); return rabbitTemplate; } @Bean Queue mailQueue() { return new Queue(MailConstants.MAIL_QUEUE_NAME, true); } @Bean DirectExchange mailExchange() { return new DirectExchange(MailConstants.MAIL_EXCHANGE_NAME, true, false); } @Bean Binding mailBinding() { return BindingBuilder.bind(mailQueue()).to(mailExchange()).with(MailConstants.MAIL_ROUTING_KEY_NAME); } }
MailSendTask(定時(shí)任務(wù),發(fā)送)
@Component public class MailSendTask { @Autowired MailSendLogService mailSendLogService; @Autowired RabbitTemplate rabbitTemplate; @Autowired EmployeeService employeeService; @Scheduled(cron = "0/10 * * * * ?") public void mailResendTask() { List<MailSendLog> logs = mailSendLogService.getMailSendLogsByStatus(); if (logs == null || logs.size() == 0) { return; } logs.forEach(mailSendLog->{ if (mailSendLog.getCount() >= 3) { mailSendLogService.updateMailSendLogStatus(mailSendLog.getMsgId(), 2);//直接設(shè)置該條消息發(fā)送失敗 }else{ mailSendLogService.updateCount(mailSendLog.getMsgId(), new Date()); Employee emp = employeeService.getEmployeeById(mailSendLog.getEmpId()); /** * 參數(shù)1:交換機(jī)名稱 * 參數(shù)2 :路由key * 參數(shù)三:數(shù)據(jù) * 參數(shù)4:作為唯一標(biāo)識(shí) * */ rabbitTemplate.convertAndSend(MailConstants.MAIL_EXCHANGE_NAME, MailConstants.MAIL_ROUTING_KEY_NAME, emp, new CorrelationData(mailSendLog.getMsgId())); } }); } }
MailReceiver(接收端)
@Component public class MailReceiver { public static final Logger logger = LoggerFactory.getLogger(MailReceiver.class); @Autowired JavaMailSender javaMailSender; @Autowired MailProperties mailProperties; @Autowired TemplateEngine templateEngine; @Autowired StringRedisTemplate redisTemplate; @RabbitListener(queues = MailConstants.MAIL_QUEUE_NAME) public void handler(Message message, Channel channel) throws IOException { Employee employee = (Employee) message.getPayload(); MessageHeaders headers = message.getHeaders(); Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG); String msgId = (String) headers.get("spring_returned_message_correlation"); if (redisTemplate.opsForHash().entries("mail_log").containsKey(msgId)) { //redis 中包含該 key,說明該消息已經(jīng)被消費(fèi)過 logger.info(msgId + ":消息已經(jīng)被消費(fèi)"); channel.basicAck(tag, false);//確認(rèn)消息已消費(fèi) return; } //收到消息,發(fā)送郵件 MimeMessage msg = javaMailSender.createMimeMessage(); MimeMessageHelper helper = new MimeMessageHelper(msg); try { helper.setTo(employee.getEmail()); helper.setFrom(mailProperties.getUsername()); helper.setSubject("入職歡迎"); helper.setSentDate(new Date()); Context context = new Context(); context.setVariable("name", employee.getName()); context.setVariable("posName", employee.getPosition().getName()); context.setVariable("joblevelName", employee.getJobLevel().getName()); context.setVariable("departmentName", employee.getDepartment().getName()); //根據(jù)模板發(fā)送 String mail = templateEngine.process("mail", context); helper.setText(mail, true); javaMailSender.send(msg); redisTemplate.opsForHash().put("mail_log", msgId, "javaboy"); channel.basicAck(tag, false); logger.info(msgId + ":郵件發(fā)送成功"); } catch (MessagingException e) { //手動(dòng)應(yīng)答, tag 消息id ,、 channel.basicNack(tag, false, true); e.printStackTrace(); logger.error("郵件發(fā)送失?。? + e.getMessage()); } } }
使用總結(jié)
- 0. rabbtMq的本地服務(wù),得開啟。(跟redis差不多)
- 1. 寫 application.properties中的rabbitMq的連接配置等
- 2. rabbitConfig配置文件。(包括:交換機(jī)選擇與隊(duì)列的配置,綁定),選擇的模式在這里配置
- 3. 直接使用,導(dǎo)入rabbitTemplate類,使用rabbitTemplate.convertAndSend()方法
- 4. 接收類
@RabbitListener(queues = MailConstants.MAIL_QUEUE_NAME) ?public void handler(Message message, Channel channel) throws IOException { ? ? ? ? 業(yè)務(wù)邏輯了 ? ? ? ? 手動(dòng)接收等等 }
到此這篇關(guān)于SpringBoot 中使用RabbtiMq 詳解的文章就介紹到這了,更多相關(guān)SpringBoot RabbtiMq 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Java8 Stream Collectors收集器使用方法解析
這篇文章主要介紹了Java8 Stream Collectors收集器使用方法解析,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-08-08Java關(guān)鍵字final的實(shí)現(xiàn)原理分析
這篇文章主要介紹了Java關(guān)鍵字final的實(shí)現(xiàn)原理分析,在JDK8之前,如果在匿名內(nèi)部類中需要訪問局部變量,那么這個(gè)局部變量一定是final修飾的,但final關(guān)鍵字可以省略,需要的朋友可以參考下2024-01-01阿里資深技術(shù)專家:在各階段中3年經(jīng)驗(yàn)的java程序員應(yīng)該具備哪些技術(shù)能力
這篇文章主要介紹了阿里資深技術(shù)專家:在各階段中3年經(jīng)驗(yàn)的java程序員應(yīng)該具備哪些技術(shù)能力,本文給大家列舉了一些內(nèi)容,大家可以根據(jù)自己需要有方法的掌握,感興趣的朋友跟隨小編一起看看吧2020-07-07JAVA中數(shù)組插入與刪除指定元素的實(shí)例代碼
下面小編就為大家分享一篇JAVA中數(shù)組插入與刪除指定元素的實(shí)例代碼,具有很好的參考價(jià)值,希望對大家有所幫助。一起跟隨小編過來看看吧2018-02-02idea如何修改文件的file is read-only問題
這篇文章主要介紹了idea如何修改文件的file is read-only問題,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-12-12