SpringBoot中使用RabbtiMq 详解
短信预约 -IT技能 免费直播动态提醒
前言
如图使用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());
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 内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
