前言 今天又再一次面试了字节，在复习准备不足的情况下，我还是接受了面试邀请。
复盘 缓存穿透/缓存击穿 这两个概念面试的时候搞混了，还是没复习好，说实话接触的系统没能有这种场景。既不在缓存中也不在数据库中的数据 ，会导致数据库压力增大，这种情况一般是黑客在攻击或者数据被误删除了，相应的解决方案有限制非法请求，缓存控制或者默认值，使用布隆过滤器判断数据是否存在。一些热点数据的过期，很容易导致大量请求到db上。 解决方案一是加互斥锁或者分布式锁去更新缓存，没能获取的锁的返回空置或默认值，二是不给热点数据设置过期时间，由后台去统一更新缓存。
redis大key的解决方案。 大key指的是value大小超过一定阈值的key。这个阈值根据系统可能指标是不同的。大key可能导致的问题可能有一是数据倾斜，如在redis集群里，大key所在的节点上内存占用率过高。二是服务器资源耗费比较严重，包括网络带宽CPU和内存。三就是redis是单线程的会导致阻塞。
rabbitmq如何保证消息的可靠性，不丢失。 首先明确消息的传递阶段有哪些。一是从生产者到rabbmitmq。二是从exchange到queue。三是未持久化消息导致意外丢失。四是消费者消费异常。下面一个个来分析。
生产者到mq。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 @Configuration public class RabbitMQConfig { /** * 配置事务管理器 */ @Bean public RabbitTransactionManager transactionManager(ConnectionFactory connectionFactory) { return new RabbitTransactionManager(connectionFactory); } } @Service public class RabbitMQServiceImpl { @Autowired private RabbitTemplate rabbitTemplate; @Transactional // 事务注解 public void sendMessage() { // 开启事务 rabbitTemplate.setChannelTransacted(true); // 发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); } }
使用事务去确认发送消息成功是一个同步操作，会阻塞等待mq应答。
1 2 3 4 5 6 7 spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制 none：表示禁用发送方确认机制 correlated：表示开启发送方确认机制 simple：表示开启发送方确认机制，并支持 waitForConfirms() 和 waitForConfirmsOrDie() 的调用。
simple是串行的应答，与事务机制一样性能较差。这里主要讨论correlated。可以通过setConfirmCallback去实现异步confirm 。
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 @Service public class RabbitMQServiceImpl { @Autowired private RabbitTemplate rabbitTemplate; @Override public void sendMessage() { rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * MQ确认回调方法 * @param correlationData 消息的唯一标识 * @param ack 消息是否成功收到 * @param cause 失败原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { // 记录日志 log.info("ConfirmCallback...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]"); if (!ack) { // 出错处理 ... } } }); } }
exchange到queue投递失败。
1 2 3 4 5 6 7 spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制 publisher-returns: true # 开启消息返回 template: mandatory: true # 消息投递失败返回客户端
mandatory 分为 true 失败后返回客户端 和 false 失败后自动删除两种策略。通过调用 setReturnCallback() 方法设置路由失败后的回调方法
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 @Service public class RabbitMQServiceImpl { @Autowired private RabbitTemplate rabbitTemplate; @Override public void sendMessage() { rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { } }); // 设置路由失败回调方法 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { /** * MQ没有将消息投递给指定的队列回调方法 * @param message 投递失败的消息详细信息 * @param replyCode 回复的状态码 * @param replyText 回复的文本内容 * @param exchange 消息发给哪个交换机 * @param routingKey 消息用哪个路邮键 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { // 记录日志 log.info("Fail Message["+message+"]==>replyCode["+replyCode+"]" +"==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]"); // 出错处理 ... } }); } }
持久化消息。rabbitmq支持将消息持久化保证服务异常后可恢复。在定义queue和exchange时就可以指定队列和交换机的持久化参数。
1 2 3 4 5 6 7 8 9 10 11 @Bean public Queue queue() { // 四个参数：name（队列名）、durable（持久化）、 exclusive（独占）、autoDelete（自动删除） return new Queue(MESSAGE_QUEUE, true); } @Bean public DirectExchange exchange() { // 四个参数：name（交换机名）、durable（持久化）、autoDelete（自动删除）、arguments（额外参数） return new DirectExchange(Direct_Exchange, true, false); }
至于消息持久化，可以在发送时指定消息类型。
1 2 Message message = MessageBuilder.withBody("test".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build(); rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message);
因此如果需要持久化，必须交换机，队列，消息都进行持久化，否则该丢失的还是会丢失。
保证消费者消费的消息不丢失。rabbitmq也提供了消费者确认机制感知消费者是否消费成功。消费成功后才删除消息，否则会继续投递。
1 2 3 4 5 spring: rabbitmq: listener: simple: acknowledge-mode: manual
默认情况下acknowledge-mode的参数是auto，即自动确认，一般情况下使用@RabbitListener注解的方法没有抛出异常，则会自动进行确认。可以结合springboot中提供的retry来实现消息重试策略。
1 2 3 4 5 6 7 8 9 10 11 spring: rabbitmq: listener: simple: acknowledge-mode: auto # 开启自动确认消费机制 retry: enabled: true # 开启消费者失败重试 initial-interval: 5000ms # 初始失败等待时长为5秒 multiplier: 1 # 失败的等待时长倍数（下次等待时长 = multiplier * 上次等待时间） max-attempts: 3 # 最大重试次数 stateless: true # true无状态；false有状态（如果业务中包含事务，这里改为false）
这里重试次数达到上限后，会被自动ack。如果存在RepublishMessageRecoverer那么会被投递到指定交换机。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 @RabbitListener(queues = RabbitMQConfig.MESSAGE_QUEUE) public void onMessage(Message message, Channel channel) { long deliveryTag = message.getMessageProperties().getDeliveryTag(); // 解析消息 byte[] body = message.getBody(); ... try { // 业务处理 // 手动确认 channel.basicAck(deliveryTag, false); }catch (Exception e) { // 记录日志 log.info("出现异常：{}", e.getMessage()); try { channel.basicNack(deliveryTag, false, false); } catch (IOException ex) { log.info("nack消息异常"); } } }
即便有了以上方案，其实也无法保证消息100%不丢失，最好的方案还是消息落库，再加消息补偿的机制去保证消息的100%正常处理。
rabbitmq的事务消息。 rabbitmq的事务我没有接触过，所以直接就回答不清楚，仔细一查结果还真有。
1 2 3 4 5 6 7 8 9 try { channel.txSelect(); channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes()); int result = 1 / 0; channel.txCommit(); } catch (Exception e) { e.printStackTrace(); channel.txRollback(); }
上面的代码和3中的rabbitTemplate.setChannelTransacted(true)是一样的，aop会自动去txSelect和txCommit。rocketmq文档 。
其他 系统复杂度主要体现在哪些地方。
随便聊聊 我其实很不解，一度陷入自我怀疑，难道我三年来真的没长进吗？不，我想不是的。
