游戏陪玩app开发，消息可靠性的实现

在游戏陪玩app开发中&＃xff0c;随着业务的不断复杂和调用链路的不断增长&＃xff0c;我们可能会慢慢引入越来越多的中间件来更好的服务于我们的系统&＃xff0c;但是每样技术都是一把双刃剑&＃xff0c;在提高游戏陪玩app开发系统性能的同时&＃xff0c;也要想办法来减少它对系统带来稳定性的影响&＃xff0c;今天要带来的是如何让RabbitMQ的可靠性达到保证。

要想了解如何保证RabbitMQ的可靠性&＃xff0c;首先要从它的执行流程开始了解。

执行流程

• 生产者发送消息或者消费者进行消费消息都会先与游戏陪玩app开发的主机建立起一条长连接&＃xff0c;由长连接里的channal来传送消息。
• 长连接优点&＃xff1a;消费者如果出现宕机或者下线&＃xff0c;mq会感知到&＃xff0c;没法继续派发后会把这条消息再次存储起来&＃xff0c;避免造成消息大面积丢失
• 消息由消息头&＃43;消息体&＃43;路由键组成。
• 消息发送出去后&＃xff0c;首先进入游戏陪玩app开发的mq服务器指定的一个虚拟主机中&＃xff0c;由虚拟主机中的exchange交换机收到后&＃xff0c;通过消息的路由键和绑定关系&＃xff0c;最终决定发往那个队列。
• 消费者通过监听指定队列拿到消息。

由执行流程就可以看到&＃xff0c;消息不管是在生产者发送到MQ服务器的过程中或者是在消费的过程中都存在着丢失的风险&＃xff0c;那怎么办呢&＃xff1f;

消息确认机制-可靠抵达

事务

提到保证可靠性的问题&＃xff0c;小伙伴们肯定首先可以想到的是事务机制&＃xff0c;RabbitMQ也提供了事务消息&＃xff0c;不过官方文档也写到事务消息让MQ的性能下降250倍&＃xff0c;所以说在游戏陪玩app开发对性能要求很高的情况下&＃xff0c;显然不适合去使用事务消息。

那只能从其他几个方面来保证可靠性了。

这是一张消息发送到消费的简图&＃xff0c;要保证可靠性的话&＃xff0c;要从三个方面来来考虑。

publisher → Broker&＃xff0c;confirmCallback机制

• 在创建connectionFactory的时候设置publisherConfirm&＃xff08;true&＃xff09;选项&＃xff0c;开始confirmCallback。
• 消息只要被broker收到就会执行confirmCallback&＃xff0c;如果是cluster模式&＃xff0c;需要所有broker都接收到才会调用confirmCallback。
• 被broker接收到只能表示message已经抵达服务器&＃xff0c;并不能保证消息一定被投递到目标queue里。所以需要用到接下来的returnCallback。

#配置yml文件
spring:rabbitmq:#开启发送单确认publisher-confirms: true


//定制abbitTemplate
&＃64;PostConstruct //MyRabbitConfig初始化完成后&＃xff0c;执行这个方法
public void initRabbitTemplate(){// 服务器收到消息确认回调/*correlationData 消息的唯一idack 消息是否成功收到cause 失败原因*/rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {log.info("confirm---->correlationData{},-------->ack{},-------->cause{}",correlationData,ack,cause);}));
}


Exchange → Queue&＃xff0c;returnCallback 机制

• confirm模式只能保证消息抵达broker&＃xff0c;不能保证消息准确投递到目标queue里。在游戏陪玩app开发的一些业务场景下&＃xff0c;需要保证消息一定要投递到目标queue里&＃xff0c;此时就需要用到return退回模式。
• 这样如果未能投递到目标queue里将会调用returnCallback
  &＃xff0c;可以记录下详细到投递数据&＃xff0c;定期的巡检或者自动纠错都将需要这些数据。

#配置yml文件
spring:rabbitmq:#开启发送端消息抵达队列确认publisher-returns: true#只要抵达队列&＃xff0c;以异步方式优先回调这个returnconfirmtemplate:mandatory: true


//定制abbitTemplate
&＃64;PostConstruct //MyRabbitConfig初始化完成后&＃xff0c;执行这个方法
public void initRabbitTemplate(){//设置消息抵达queue的确认回调 (消息没有投递给指定队列&＃xff0c;才会触发这个失败回调)/*message 投递失败的详细信息replyCode 回复的状态码replyText 回复的文本内容exchange 这个消息发送给哪个交换机routingKey 这个消息用的哪个路由键*/rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {log.info("return---->message{},-->replyCode{},-->replyText{},-->exchange{},-->routingKey{}",message,replyCode,replyText,exchange,routingKey);});
}


Queue → Consumer&＃xff0c;ack消息确认机制&＃xff08;消费端&＃xff09;

• 默认自动ack&＃xff0c;消息被消费者收到&＃xff0c;就会从broker的queue中移除
• 问题&＃xff1a;收到很多消息&＃xff0c;自动回复给服务器ack&＃xff0c;只处理一个消息&＃xff0c;游戏陪玩app开发服务器就宕机了。这时消息就会全部丢失,所以要关闭默认的自动ack机制。
• 消费者获取到消息&＃xff0c;成功处理&＃xff0c;可以回复ack给broker
• basic.ack用于肯定确认&＃xff1b;broker将移除此消息
• basic.nack用于否定确认&＃xff1b;可以指出broker是否丢弃此消息&＃xff0c;可以批量
• basic.reject用于否定确认&＃xff1b;同上&＃xff0c;但不能批量
• queue无消费者&＃xff0c;消息依然会被存储&＃xff0c;直到消费者消费
• 消费者收到消息&＃xff0c;默认会自动ack。但是如果无法确认此消息是否被处理完成&＃xff0c;或者成功处理。我们可以开启手动ack模式
• 消息处理成功&＃xff0c;ack()&＃xff0c;接受下一个消息&＃xff0c;此消息broker就会移除
• 消息处理失败&＃xff0c;nack()/reject()&＃xff0c;重新发送给其他人进行处理&＃xff0c;或者容错处理后ack
• 消息一直都没有调用ack/nack方法&＃xff0c;broker认为此消息正在被处理&＃xff0c;不会投递给别人。此时客户端断开。消息不会被broker移除&＃xff0c;会投递给别人
• 手动ack机制下&＃xff0c;只要没有明确告诉mq消息被消费&＃xff0c;没有ack&＃xff0c;游戏陪玩app开发消息就一直是unacked状态。即使consumer宕机&＃xff0c;消息不会丢失&＃xff0c;会变为ready状态&＃xff0c;下次一有新的consumer连接进来就发给他

#配置yml文件
spring:rabbitmq:#切换为手动acklistener:direct:acknowledge-mode: manual


完成以上三个配置后&＃xff0c;我们用于消费消息的代码就会变成这样

/*** &＃64;author lp* &＃64;date 2020/8/9 15:05*/
&＃64;Service
&＃64;Slf4j
&＃64;RabbitListener(queues &＃61; "demo.queue")
public class DemoListener {&＃64;Autowiredprivate DemoService service;&＃64;RabbitHandlerpublic void listener(DemoEntity entity, Channel channel, Message message) throws IOException {log.info("-----------开始消费消息----------");try {//具体业务service.doSomething(entity);//成功处理回复aclchannel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {//失败处理重新返回queuechannel.basicReject(message.getMessageProperties().getDeliveryTag(), true);}}
}


当然在游戏陪玩app开发中仅仅是这三步配置还是不够的&＃xff0c;因为上述步骤并没有做到消息持久化&＃xff0c;在做好持久化方案后&＃xff0c;我们的消息将无敌。&＃xff08;夸张的修辞手法&＃xff09;

小结&＃xff1a;

• 消息发送出去&＃xff0c;由于网络问题没有抵达游戏陪玩app开发服务器
• 做好容错方法&＃xff08;try-catch&＃xff09;&＃xff0c;发送消息可能会网络失败&＃xff0c;失败后要有重试机制&＃xff0c;可记录到数据库&＃xff0c;采用定期扫描重发的方式。
• 做好日志记录(给数据库保存每一个消息的详细信息)&＃xff0c;每个消息状态是否都被服务器收到&＃xff0c;应记录。
• 做好定期重发&＃xff0c;如果消息没有发送成功&＃xff0c;定期去数据库扫描未成功的消息进行重发。
• 消息抵达Broker&＃xff0c;Broker要将消息写入磁盘&＃xff08;持久化&＃xff09;才算成功。此时Broker尚未持久化完成&＃xff0c;宕机
• publisher也必须加入确认回调机制&＃xff0c;确认成功的消息&＃xff0c;修改数据库消息状态。
• 自动ack的状态下。消费者收到消息&＃xff0c;但还没来得及处理消息&＃xff0c;宕机
• 一定开启手动ack&＃xff0c;消息消费成功后才移除&＃xff0c;失败或者没来得及处理就noAck并重新入队。

做好这些后&＃xff0c;游戏陪玩app开发中消息丢失的可能性已经很小很小了&＃xff0c;但是又会有新的问题出现&＃xff0c;比如说重复消费&＃xff0c;消息过多导致消费者宕机等。

防止消息重复

• 消息消费成功&＃xff0c;事务已经提交&＃xff0c;ack时&＃xff0c;机器宕机。导致没有ack成功&＃xff0c;Broker的消息重新由unack变为ready&＃xff0c;并发送给其他消费者
• 消费失败时&＃xff0c;由于重试机制&＃xff0c;自动又将消息发送出去
• 成功消费&＃xff0c;ack时宕机&＃xff0c;消息由unack变为ready,Broker又重新发送
• 消费者的业务消费接口应该设计为幂等性的。比如扣库存有工作单的状态标志
• 使用防重表&＃xff08;redis/mysql&＃xff09;&＃xff0c;发送消息每一个都有业务的唯一标识&＃xff0c;处理过就不用处理
• rabbitMq每一次消息都有redelivered字段&＃xff0c;可以获取是否被重新投递过来的&＃xff0c;而不是第一次被投递过来的

消息积压

• 产生原因&＃xff1a;
• 消费者宕机
• 消费者能力不足积压
• 发送者发送流量太大
• 如何解决&＃xff1a;
• 上线更多消费者&＃xff0c;进行正常消费
• 上线专门的消息队列服务&＃xff0c;将消息先批量取出来&＃xff0c;记录到数据库&＃xff0c;离线慢慢处理