rabbitmq消息去重及防丢失解决方案

标签: 后端笔记 rabbitmq | 发表时间:2019-03-22 11:38 | 作者:
出处:https://fatboa.co/

-


-

个人笔记,如有描述不当,欢迎留言指出~

前提

我们知道一个电商项目里时刻都有海量的消息通知,比如顾客注册通知、签到通知、下单通知等等,而我们公司的电商项目更加复杂,包含了客户端、门店端以及供应商端三端,各种各样的消息通知游走在各个服务模块间。如果每个模块都要实现一套消息通知的功能,那无疑是多余的。所以我把各模块的消息功能提取出来独立成一个服务模块,就像一个快递员,把各模块的消息准确投递至各端。
我采用了自己熟悉的rabbitmq来实现消息功能,
当模块开发完交差时,组长冷不丁来了句:消息去重以及防丢失机制实现了没?w(゚Д゚)w好吧,赶紧去实现。

消息去重

@消息大致流程
依我的经验来看,在消费端去重比较好。因为即使生产端保证投递到rabbitmq上的消息是不重复的,但rabbitmq服务器有可能由于系统或网络原因导致消息重复推送到消费端,所以生产端去重是不可靠的,应当在消费端去重。

怎么解决呢?我的方案是在生产端投递消息的同时,传入correlationId关联id,在消费端接收消息后,从message的messageProperties中拿到correlationId,再根据correlationId从db中查询是否有相关记录。如果有,则说明这条消息已被我们消费过,直接ack,不进行业务处理;没有,那就把消息内容和correlationId存入表中,然后ack。

这里说明一下,我把消息的接收和业务处理分开来了。消息监听器只负责监听队列消息,并将其存至db中。在另外的任务线程里,从db中取消息记录进去业务处理,如果业务处理中出现异常,结合elasticsearch实现异常报警(这部分还没做,目前还只是记录下错误信息及消息内容)。

why?为啥分开处理,其实一开始的设计中消息接收和处理是写在一起的,消息处理成功回复ack,处理异常回复nack。但会有一个严重的问题,但测试环境中,我们发现总有那么几条消息卡在队列里,就因为处理异常回复nack,消息一直在重入队,严重消耗rabbitmq服务器的性能!所以说,大部分异常的消息,都不能指望把消息重推到别的消费端就能处理成功了,所以消息接收和处理分开来是比较好的。

方案是有了,但具体代码怎么实现呢?
生产端关键代码:

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
public void send(String routingKey, String msg) {      
RabbitTemplate rabbitTemplate = applicationContext.getBean("rabbitTemplate", RabbitTemplate.class);
rabbitTemplate.setReturnCallback(this);
log.info("消息发送内容 : " + msg);
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (!ack) {
throw new RuntimeException("send error " + cause);
} else {
log.info("send() 消息发送成功 ");
}
});
rabbitTemplate.convertAndSend("amq.topic", routingKey, msg, correlationId);
}

我们点开 rabbitTemplate.convertAndSend方法

1      
2
3
public void convertAndSend(String exchange, String routingKey, Object object, CorrelationData correlationData) throws AmqpException {      
this.send(exchange, routingKey, this.convertMessageIfNecessary(object), correlationData);
}

看到没,convertAndSend方法是以 Object来接收消息内容,它内部调用的send方法最终还是把Object类转成 Message
@Message.java
从上图可以看的出,Message包含了 ENCODING(编码方式)、 SERIALIZER_MESSAGE_CONVERTER(序列化消息转换器)、 messageProperties(消息属性)、 body(消息内容),队列里消息存放着这些东东。
我们再看看 MessageProperties里放着什么

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
public class MessageProperties implements Serializable {      
private static final long serialVersionUID = 1619000546531112290L;
public static final String CONTENT_TYPE_BYTES = "application/octet-stream";
public static final String CONTENT_TYPE_TEXT_PLAIN = "text/plain";
public static final String CONTENT_TYPE_SERIALIZED_OBJECT = "application/x-java-serialized-object";
public static final String CONTENT_TYPE_JSON = "application/json";
public static final String CONTENT_TYPE_JSON_ALT = "text/x-json";
public static final String CONTENT_TYPE_XML = "application/xml";
public static final String SPRING_BATCH_FORMAT = "springBatchFormat";
public static final String BATCH_FORMAT_LENGTH_HEADER4 = "lengthHeader4";
public static final String SPRING_AUTO_DECOMPRESS = "springAutoDecompress";
public static final String X_DELAY = "x-delay";
public static final String DEFAULT_CONTENT_TYPE = "application/octet-stream";
public static final MessageDeliveryMode DEFAULT_DELIVERY_MODE;
public static final Integer DEFAULT_PRIORITY;
private final Map<String, Object> headers = new HashMap();
private volatile Date timestamp;
private volatile String messageId;
private volatile String userId;
private volatile String appId;
private volatile String clusterId;
private volatile String type;
private volatile String correlationId;
private volatile String replyTo;
private volatile String contentType = "application/octet-stream";
private volatile String contentEncoding;
private volatile long contentLength;
private volatile boolean contentLengthSet;
private volatile MessageDeliveryMode deliveryMode;
private volatile String expiration;
private volatile Integer priority;
private volatile Boolean redelivered;
private volatile String receivedExchange;
private volatile String receivedRoutingKey;
private volatile String receivedUserId;
private volatile long deliveryTag;
private volatile boolean deliveryTagSet;
private volatile Integer messageCount;
private volatile String consumerTag;
private volatile String consumerQueue;
private volatile Integer receivedDelay;
private volatile MessageDeliveryMode receivedDeliveryMode;
private volatile boolean finalRetryForMessageWithNoId;
private transient volatile Type inferredArgumentType;
private transient volatile Method targetMethod;
private transient volatile Object targetBean;
}

果然 correlationId就在这里,然后看到这里我就没继续深入了,原以为rabbitTemplate.convertAndSend方法会自动将correlationId放入messageProperties中,结果表明我错了。在消费端拿到的correlationId为 null。也就是说,convertAndSend方法里correlationId根本就没有被放进去的,大家感兴趣的话可以看看源码,这里就不说了。

问题找出来就好办了

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public void send(String routingKey, String msg) {      
RabbitTemplate rabbitTemplate = applicationContext.getBean("rabbitTemplate", RabbitTemplate.class);
rabbitTemplate.setReturnCallback(this);
log.info("消息发送内容 : " + msg);
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (!ack) {
throw new RuntimeException("send error " + cause);
} else {
log.info("send() 消息发送成功 ");
}
});
Message message = MessageBuilder.withBody(msg.getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setCorrelationId(correlationId.toString())
.build();
rabbitTemplate.convertAndSend("amq.topic", routingKey, message, correlationId);
}

直接构建Message类,手动传入correlationId总行了吧。在消费端从Message里拿到correlationId,再从db查询就行了。好了,到这里去重机制就实现了

消息防丢失

rabbitmq是支持队列、消息的持久化的。即便rabbitmq突然挂了,那些尚在队列未能推送的消息在rabbitmq重启后也是能够继续推送的,所以丢失问题一般不出现在rabbitmq上。

rabbitmq将消息从队列推到消费端后,需要有一个回应告诉它队列里的这条消息的去留。主要有两种方式:

  • auto: 自动回应,消息在发送给消费端后立即确认
  • manual 手动回应,消息消费正常后由消费端返回ack;或消费异常返回nack,将消息重入队;或返回reject,丢弃该条消息

springboot的yaml配置

1      
2
3
4
5
6
7
8
9
10
11
12
rabbitmq:      
host: 127.0.0.1
port: 5672
username: admin
password: admin
publisher-confirms: true #开启confirmcallback
publisher-returns: true #开启returncallback
listener:
simple:
acknowledge-mode: manual
direct:
acknowledge-mode: MANUAL

另外,如果消息在消费的时候,消费端与rabbitmq的连接中断了,那这条消息会被重新放回队列进行推送,这个时候我们的去重机制就起作用了;如果消费的时候,消费端死机了,长时间不回应rabbitmq,这时候我们可以将该消息转至死信队列,防止原队列阻塞。死信队列,这里也不做介绍,有兴趣百度呗。
所以消费端出现消息丢失的可能性也不大,问题就可能出在生产端。看看下面这张图
@来源于网络
左边P代表生产端,中间是rabbitmq,右边是消费端,绿色的X是交换机,红色的是队列,用过rabbitmq的小伙伴肯定一目了然了。

rabbitmq 整个消息投递的路径为:
producer-> rabbitmq broker cluster-> exchange-> queue-> consumer

生产端投递消息到rabbitmq里,rabbitmq将消息发到交换机中,交换机再根据路由键将消息最终送到队列中,队列取出消息推送到消费端。只有最终抵达队列的消息才是可靠的,不会丢失。所以我们要实现的就是保证生产端的消息务必推送到rabbitmq的队列中。

那么生产端是怎么知道自己的消费准确投递到了队列中呢?rabbitmq返回了两个回调给生产端。

  • message 从 producer 到 rabbitmq broker cluster 则会返回一个 confirmCallback
  • message 从 exchange->queue 投递失败则会返回一个 returnCallback 。我们将利用这两个 callback 控制消息的最终一致性和部分纠错能力。

解决方案

生产端在投递消息前,先将消息内容、投递状态、重试次数记录在db中,然后在两个回调中修改记录状态。另外再开一个任务线程去取db中记录的失败消息,进行重新投递。

代码实现

失败消息记录实体类:

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
/**      
* @time: 2019/2/18 9:14
* @author: hl
* @descripe: 失败消息记录
* @version: 1.0
*/
Entity
@Table(name = "t_failure_mq_record")
@Data
@NoArgsConstructor
@EntityListeners(AuditingEntityListener.class)
public class FailureMqRecord extends Uuid {
/**
* 失败消息内容
*/
private String message;
/**
* 重试次数
*/
@Column(name = "retry_time")
private Integer retryTime;

/**
* 消息状态 1:投递成功 2:投递失败
*/
private Integer status;

/**
* 关联id
*/
private String correlationId;
/**
* 创建时间
*/
@CreatedDate
@Column(name = "create_time", updatable = false)
private Date createTime;

public FailureMqRecord(String message) {
this.message = message;
}
}

rabbitmq发送器:

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
/**      
* @time: 2019/2/13 9:47
* @author: hl
* @descripe:
* @version: 1.0
*/
@Component
@Slf4j
public class RabbitMqSender implements RabbitTemplate.ReturnCallback {
@Autowired
private ApplicationContext applicationContext;
@Autowired
private FailureMqRecordRepository failureMqRecordRepository;

public void send(String routingKey, FailureMqRecord failureMqRecord) {
RabbitTemplate rabbitTemplate = applicationContext.getBean("rabbitTemplate", RabbitTemplate.class);
//设置当前实例为rabbitmqtemplate的returncallback
rabbitTemplate.setReturnCallback(this);
rabbitTemplate.setConfirmCallback(((correlationData1, ack, cause) -> {
if (!ack) { //投递至broker失败
failureMqRecord.setStatus(2);//设为投递失败
failureMqRecordRepository.save(failureMqRecord);
}
}));
Message message = MessageBuilder.withBody(failureMqRecord.getMessage().getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setCorrelationId(failureMqRecord.getCorrelationId())
.build();
rabbitTemplate.convertAndSend("amq.topic", routingKey, message, new CorrelationData(failureMqRecord.getCorrelationId()));
failureMqRecord.setStatus(1);//设为投递成功
failureMqRecord.setRetryTime(failureMqRecord.getRetryTime() + 1);//重试次数+1
failureMqRecordRepository.save(failureMqRecord);
}

/**
* 消息由exchang未能正确投递到queue时触发回调
*
* @param message
* @param replyCode
* @param replyText
* @param exchange
* @param routingKey
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) throws BusinessException {
FailureMqRecord mq = failureMqRecordRepository.findByCorrelationId(message.getMessageProperties().getCorrelationId());
mq.setStatus(2);
failureMqRecordRepository.save(mq);
log.error("审批消息:{} 投递至路由:{}失败", message.getBody(), routingKey);
}
}

任务线程,实现消息重试机制:

1      
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
/**      
* @time: 2019/2/18 10:10
* @author: huanglong
* @descripe: 消息重发定时器
* @version: 1.0
*/
@Component
@Slf4j
public class MqScheduler {
@Autowired
private RabbitMqSender rabbitMqSender;
@Autowired
private ApprovalMqRepository failureMqRecordRepository;
@Autowired
private RedisDistributedLock redisDistributedLock;

/**
* 每3分钟执行一次
* 将投递失败消息重新投递到rabbitmq
*/
@Scheduled(cron = "* */3 * * * ?")
void push() {
List<FailureMqRecord> failureMqRecords = failureMqRecordRepository.findAll()
.stream()
.filter(failureMqRecord -> {
if (failureMqRecord.getRetryTime() == 3) {
log.error("警告!该消息已重投3次失败,请人工处理,消息记录uuid:{}", failureMqRecord.getUuid());
}
//过滤出重试次数不超过3次、状态为2的消息记录
if (failureMqRecord.getRetryTime() < 3 && failureMqRecord.getStatus() == 0) {
return true;
}
return false;
})
.collect(Collectors.toList());

Iterator<FailureMqRecord> mqIterator = failureMqRecords.iterator();
while (mqIterator.hasNext()) {
FailureMqRecord failureMqRecord = mqIterator.next();
//获取,过期时间5秒,不重复获取
if (redisDistributedLock.lock(failureMqRecord.getUuid(), 5000L, 0, 1000L)) {
// 因为有可能上一个线程刚释放该记录的锁,就被当前先线程获取到该记录的锁,导致记录已被
FailureMqRecord failurelatest = failureMqRecordRepository.findById(failureMqRecord.getUuid()).orElse(null);
ApprovalPushMessage pushMessage = JSON.parseObject(failurelatest.getMessage(), ApprovalPushMessage.class);
//当前时间距离该记录最近一次修改时间的间隔,防止上个线程重试过后,当前线程又重试一次
long lastUpdatePeriod = System.currentTimeMillis() - failurelatest.getUpdateTime().getTime(); //距离上次更新间隔
//重判断记录是否符合条件,重试次数小于3、状态为投递失败、距离上次重试不能少于2分钟
if (failurelatest.getRetryTime() < 3 && failurelatest.getStatus() == 2 && lastUpdatePeriod > 2 * 60 * 1000) {
rabbitMqSender.send("approval.create", failureMqRecord);
}
//释放
redisDistributedLock.releaseLock(failureMqRecord.getUuid());
}
}
}
}

聪明的小伙伴看到这里,会发现任务线程里还用到了分布式。为啥还要加分布式锁,因为是分布式架构啊,会有多个相同定时器从db里取记录处理,如果不加分布式锁,那真的要乱套了。因为redis用的多,就用redis来实现分布式锁了,zookeeper啥的,有空再研究了。redis分布式锁的代码实现,网上有很多资源,我这里就不贴了,嘿嘿

好了,到这里rabbitmq的去重以及防丢失方案已经实现了,如果你有更好的解决方案或者指出我方案的不足,欢迎留言讨论

相关 [rabbitmq 消息] 推荐:

消息中间件的技术选型心得-RabbitMQ、ActiveMQ和ZeroMQ

- - haohtml's blog
RabbitMQ、ActiveMQ和ZeroMQ都是极好的消息中间件,但是我们在项目中该选择哪个更适合呢. 下面我会对这三个消息中间件做一个比较,看了后你们就心中有数了. RabbitMQ是AMQP协议领先的一个实现,它实现了代理(Broker)架构,意味着消息在发送到客户端之前可以在中央节点上排队.

rabbitmq消息去重及防丢失解决方案

- - fatboa
个人笔记,如有描述不当,欢迎留言指出~. 我们知道一个电商项目里时刻都有海量的消息通知,比如顾客注册通知、签到通知、下单通知等等,而我们公司的电商项目更加复杂,包含了客户端、门店端以及供应商端三端,各种各样的消息通知游走在各个服务模块间. 如果每个模块都要实现一套消息通知的功能,那无疑是多余的. 所以我把各模块的消息功能提取出来独立成一个服务模块,就像一个快递员,把各模块的消息准确投递至各端.

消息中间件选型分析——从Kafka与RabbitMQ的对比来看全局

- - 程序猿DD
有很多网友留言:公司要做消息中间件选型,该如何选. 消息选型的确是一个大论题,实则说来话长的事情又如何长话短说. 对此笔者专门撰稿一篇内功心法: 如何看待消息中间件的选型,不过这篇只表其意未表其行,为了弥补这种缺陷,笔者最近特意重新撰稿一篇,以供参考. 温馨提示:本文一万多字,建议先马(关注)后看.

史上最透彻的 RabbitMQ 可靠消息传输实战 - 后端 - 掘金

- -
缓存架构之史上讲的最明白的RabbitMQ可靠消息传输实战演练. 一、背景介绍:消息可靠传递的重要性. 比如:某个广告主(如:天猫)想在我们的平台(如:今日头条)投放广告,当通过我们的广告系统新建广告的时候,该消息在同步给redis缓存(es)的时候丢失了,而我们又没有发现,造成该广告无法正常显示出来,那这损失就打了,如果1天都没有该广告的投放记录,那就有可能是上百万的损失了,所以消息的可靠传输多我们的广告系统也是很重要的.

【架构】关于RabbitMQ

- - 学习笔记
1      什么是RabbitMQ. RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗. 消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然:. 例如一个日志系统,很容易使用RabbitMQ简化工作量,一个Consumer可以进行消息的正常处理,另一个Consumer负责对消息进行日志记录,只要在程序中指定两个Consumer所监听的queue以相同的方式绑定到同一个exchange即可,剩下的消息分发工作由RabbitMQ完成.

RabbitMQ (三) 发布/订阅

- - CSDN博客架构设计推荐文章
转发请标明出处: http://blog.csdn.net/lmj623565791/article/details/37657225. 本系列教程主要来自于官网入门教程的翻译,然后自己进行了部分的修改与实验,内容仅供参考. 上一篇博客中,我们实现了工作队列,并且我们的工作队列中的一个任务只会发给一个工作者,除非某个工作者未完成任务意外被杀死,会转发给另外的工作者,如果你还不了解: RabbitMQ (二)工作队列.

rabbitmq java client api详解

- - 五四陈科学院
以下内容由 [五四陈科学院]提供. AMQP协议是一个高级抽象层消息通信协议,RabbitMQ是AMQP协议的实现. 每个rabbitmq-server叫做一个Broker,等着tcp连接进入. 在rabbitmq-server进程内有Exchange,定义了这个消息的发送类型. Queue是进程内的逻辑队列,有多个,有名字.

抽取rabbitmq网络层做的echo server

- 2sin18 - codedump
传说rabbitmq网络层实现的优雅高效,于是我就尝试着将其中的网络层抽取出来,模拟着做了一个echo服务器,代码放在这里.. rabbitmq的做法是内置状态机,通过切换callback的形式处理不同的业务,这样只有一个子进程处理一个链接,性能提高不少.. 测试这个echo服务器的客户端我使用的是telnet,telnet输入的数据会自动在后面加上”\r\n”发送到对端,于是代码中以这个来判断是否接收了一条消息,抽取出来回复给对端..

RabbitMQ关键性问题调研

- - Java - 编程语言 - ITeye博客
摘要:本篇是本人对RabbitMQ使用的关键性问题进行的调研,如性能上限、数据存储、集群等,.             具体的 RabbitMQ概念、使用方法、SpringAMQP配置,假设读者已有了基础. 1.1  RabbitMQ数据速率问题. 在边读边写的情况下:速率只与网络带宽正相关,网络使用率最高能达到接近100%,并且数据使用率很高(90%以上).

[转][RabbitMQ+Python入门经典] 兔子和兔子窝

- lostsnow - heiyeluren的blog(黑夜路人的开源世界)
高级消息队列协议(AMQP1)是一个异步消息传递所使用的应用层协议规范. AMQP的原始用途只是为金融界提供一个可以彼此协作的消息协议,而现在的目标则是为通用消息队列架构提供通用构建工具. RabbitMQ作为一个工业级的消息队列中间件,基于AMQP协议的实现,由erlang语言编写. 本文讲解 RabbitMQ+Python 的使用.