根据官网我们知道rabbitmq支持事务，所以我们能用事务来解决消息丢失的问题（由于事务的性能差所以不推荐使用）
使用事务会降低250倍的性能
推荐使用确认机制
RabbitMQ结构
生产者，消费者，交换机Exchange，路由键，队列（queue），通道（channal）
RabbitM消息确认机制
P（消费者）----------->Broker【Exchange---------->Queue】--------------------->C(消费者)
消费者到达RabbitMQ的确认机制为ConfirmCallback
特点：只要消息到达服务器就会触发
服务器接收到消息持久化到硬盘的机制是returnCallback
特点：失败才触发
配置
@Configuration
public class MyRabbitmqConfig {
@Resource
RabbitTemplate rabbitTemplate;
@Bean
MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
@PostConstruct //MyRabbitConfig对象创建完成以后，执行这个方法
public void initRabbitTemplate(){
//设置确认回调
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
System.out.println("confirm...."+s);
System.out.println(" ==========>"+b);
}
});
//设置消息抵达队列的确认回调
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
System.out.println("失败的信息"+returnedMessage.getMessage()+
"replyCode===>"+returnedMessage.getReplyCode()+
"exchange=====>"+returnedMessage.getExchange()+
"routingKey=====>"+returnedMessage.getRoutingKey());
//将没有持久化的消息发送到数据库中
String xx= returnedMessage.getMessage().toString();
String encrypt = MD5.encrypt(xx);
try{
lostMapper.addmessage(encrypt,xx);
}catch (Exception e){
System.out.println("消息已经存入数据库中");
}
}
});
}
}
Service层
@Service
public class GjxServiceImpl implements GjxService {
@Autowired
RabbitTemplate rabbitTemplate;
@Autowired
LostService lostService;
@Override
public void sendmessage() {
try {
rabbitTemplate.convertAndSend("zlifeexchange", "zlife", "郭峻旭");
} catch (Exception exception) {
System.out.println("rabbitmq服务器宕机");
String gjx = MD5.encrypt("郭峻旭");
System.out.println(gjx);
try{
lostService.addmessage(gjx, "郭峻旭");
}catch (Exception e){
System.out.println("丢失的消息已经存入数据库当中");
}
}
}
}
主要的思想是如果消息没有发送到RabbitMQ服务器上就将发送的消息存入到数据库中通过定时任务定时将数据库中发送失败的消息重新发送
定时任务
springboot在主启动类上添加注解@EnableScheduling
通过注解@Scheduled完成定时任务
@Service
public class CsImpl implements Cs {
@Autowired
LostMapper lostMapper;
@Autowired
RabbitTemplate rabbitTemplate;
@Autowired
LostService lostService;
@Override
@Scheduled(cron ="0/5 * * * * ?")
public void CsText(){
List losts = lostMapper.selectList(null);
for (int i = 0; i
System.out.println("====================>");
System.out.println(losts.get(i).getMessage());
String xx=losts.get(i).getMessage();
try {
rabbitTemplate.convertAndSend("zlifeexchange", "zlife", xx);
QueryWrapper queryWrapper=new QueryWrapper();
queryWrapper.like("message",xx);
lostMapper.delete(queryWrapper);
}catch (Exception e){
System.out.println("rabbitmq服务器宕机");
String gjx = MD5.encrypt(xx);
System.out.println(gjx);
try {
lostService.addmessage(gjx, xx);
}catch (Exception exception){
System.out.println("丢失的消息已经存入数据库当中");
}
}
}
}
}
配置文件
server:
port: 8084
spring:
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
username: root
password:
redis:
host: 127.0.0.1
port: 6379
rabbitmq:
host: 127.0.0.1
port: 5672
virtual-host: /
# 开启发送端确认机制
publisher-confirm-type: correlated
# 开启发送端消息抵达队列确认
publisher-returns: true
# 只要抵达队列以异步方法优先回调returnconfirm确认
template:
mandatory: true
# 手动ACK 手动签收
listener:
simple:
acknowledge-mode: manual
main:
# 运行循环调用
allow-circular-references: true
#mybatis:
# mapper-locations: classpath:mapper/*.xml
mybatis:
mapper-locations: classpath:mapper/*.xml
logging:
level:
com.gjx: debug
同理加上手动ack机制确保消息不会丢失
消息丢失的问题解决了，又出现了新的问题消息重复消费怎么解决？
例子：比如当我们已经消费了消息还没来得及ack即中间的时间段 RabbitMQ服务器突然宕机这个时候我们明明已经消费了消息但是没来得及ack，这个消息就会重新回到队列当中RabbitMQ服务重新启动后这个消息又会再消费一次，重复消费了
