SpringBoot 中使用RabbtiMq 详解
更新时间：2022年07月27日 15:49:23 投稿：hqx
这篇文章主要介绍了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 内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
下面小编就为大家带来一篇Java运行环境搭建的图文教程。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧2017-09-09
SpringBoot JavaMailSender发送邮件功能(实例详解)
JavaMailSender是Spring提供的，非常好用的，实现邮件发送的接口 ，这篇文章主要介绍了SpringBoot JavaMailSender发送邮件功能,需要的朋友可以参考下2024-03-03
Jpa中Specification的求和sum不生效原理分析
这篇文章主要为大家介绍了Jpa中Specification的求和sum不生效原理示例分析,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪2023-08-08
这篇文章主要介绍了java设计模式之适配器模式Adapter的相关资料，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-09-09
这篇文章主要介绍了Java中的this、package、import,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2023-06-06
这篇文章主要介绍了java智能问答图灵机器人AI接口（聚合数据），文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-02-02
本教程提供了如何创建一个简单的JDBC应用程序的示例。演示如何打开一个数据库连接，执行SQL查询，并显示结果2014-03-03
这篇文章主要介绍了java常用工具类之数据库连接类,可以连接多种数据库,代码中包含详细注释,需要的朋友可以参考下2014-07-07
FTP(File Transfer Protocol 文件传输协议)是Internet 上用来传送文件的协议，本文给大家分享Java实现FTP服务器功能实例代码，对java实现ftp服务器相关知识感兴趣的朋友一起学习吧2015-12-12
这篇文章主要介绍了Spring Data JPA分页复合查询原理解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2019-11-11
最新评论