SpringBoot 中使用RabbtiMq 詳解
前言
如圖使用redisTemplate 一樣的簡單方便
模擬發送郵件的情況
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 ## 根據自己情況而定,可以不用 spring.rabbitmq.listener.simple.acknowledge-mode=manual spring.rabbitmq.listener.simple.prefetch=100
寫在配置文件中,由 RabbitProperties 這個類進行讀取,封裝到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;//最大重試次數 public static final Integer MSG_TIMEOUT = 1;//消息超時時間 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; //發送郵件的 @Autowired MailSendLogService mailSendLogService; @Bean RabbitTemplate rabbitTemplate() { RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory); //手動應答返回的標志 rabbitTemplate.setConfirmCallback((data, ack, cause) -> { String msgId = data.getId(); if (ack) { logger.info(msgId + ":消息發送成功"); mailSendLogService.updateMailSendLogStatus(msgId, 1);//修改數據庫中的記錄,消息投遞成功 } else { logger.info(msgId + ":消息發送失敗"); } }); rabbitTemplate.setReturnCallback((msg, repCode, repText, exchange, routingkey) -> { logger.info("消息發送失敗"); }); 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(定時任務,發送)
@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);//直接設置該條消息發送失敗 }else{ mailSendLogService.updateCount(mailSendLog.getMsgId(), new Date()); Employee emp = employeeService.getEmployeeById(mailSendLog.getEmpId()); /** * 參數1:交換機名稱 * 參數2 :路由key * 參數三:數據 * 參數4:作為唯一標識 * */ 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,說明該消息已經被消費過 logger.info(msgId + ":消息已經被消費"); channel.basicAck(tag, false);//確認消息已消費 return; } //收到消息,發送郵件 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()); //根據模板發送 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 + ":郵件發送成功"); } catch (MessagingException e) { //手動應答, tag 消息id ,、 channel.basicNack(tag, false, true); e.printStackTrace(); logger.error("郵件發送失敗:" + e.getMessage()); } } }
使用總結
- 0. rabbtMq的本地服務,得開啟。(跟redis差不多)
- 1. 寫 application.properties中的rabbitMq的連接配置等
- 2. rabbitConfig配置文件。(包括:交換機選擇與隊列的配置,綁定),選擇的模式在這裡配置
- 3. 直接使用,導入rabbitTemplate類,使用rabbitTemplate.convertAndSend()方法
- 4. 接收類
@RabbitListener(queues = MailConstants.MAIL_QUEUE_NAME) public void handler(Message message, Channel channel) throws IOException { 業務邏輯瞭 手動接收等等 }
到此這篇關於SpringBoot 中使用RabbtiMq 詳解的文章就介紹到這瞭,更多相關SpringBoot RabbtiMq 內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- 詳解SpringBoot中使用RabbitMQ的RPC功能
- 聊聊RabbitMQ發佈確認高級問題
- rabbitmq中routingkey的作用說明
- RabbitMQ 3.9.7 鏡像模式集群與Springboot 2.5.5 整合
- SpringBoot集成RabbitMQ和概念介紹