MQ夺命11问

为什么使用MQ

削峰填谷
异步调用

基于什么做的选型

qps压力大性能第一
java开发语言
分布式架构
可能用到顺序消息事物消息

所以选的RocketMq

消息可靠性保证？

生产者丢失

同步发送能保证不推荐
异步发送无回调消息丢失
异步发送 + 回调 + 本地消息表 + job轮询

MQ丢失

同步刷盘不会丢失性能损耗
异步刷盘会丢失

消费者丢失

默认消费者回复ACK确认

消费一直失败导致消息积压怎么办？

消息积压达到磁盘上线消息被删除了怎么办？

生产者有发送记录可以比对的

RocketMq实现原理

由NameServer注册中心集群 Producer集群 Consumeer集群和若干Broker(mq进程)组成,架构原理:

Broker启动向NameServer注册并且保持长连接每30s发送心跳
Producer发送消息的时候从NameServer获取Broker服务器,根据负载均衡选择一台服务器发送
Consumer从NameServer获取Broker服务器,主动拉取消息消费

为什么Rocket不使用Zookeeper作为注册中心

CAP理论,ZK满足CP,不能保证可用性作为服务发现不能接受
基于性能考虑 NameServer轻量扩展方便 ZK的写是不可扩展的
ZK持久化机制问题 ZAB协议对每一个写请求会在每个 ZooKeeper 节点上保持写一个事务日志，同时再加上定期的将内存数据镜像（Snapshot）到磁盘来保证数据的一致性和持久性，而对于一个简单的服务发现的场景来说，这其实没有太大的必要
消息发送应该弱依赖注册中心 RocketMQ的设计理念也正是基于此，生产者在第一次发送消息的时候从NameServer获取到Broker地址后缓存到本地，如果NameServer整个集群不可用，短时间内对于生产者和消费者并不会产生太大影响

broker怎么保存数据的？

commitlog文件和consumequeue文件和 indexfile文件
broker收到消息后会把消息存到commitlog, 同时再分布式存储中每个broker会保存一定topic的数据 topic对应的messagequeue下会生成consumequeue文件用于保存commitlog的物理偏移量offset, indexfile保存key和offset的对应关系

master和slave怎么同步数据？

而消息在master和slave之间的同步是根据raft协议来进行的：

在broker收到消息后，会被标记为uncommitted状态
然后会把消息发送给所有的slave
slave在收到消息之后返回ack响应给master
master在收到超过半数的ack之后，把消息标记为committed
发送committed消息给所有slave，slave也修改状态为committed

RocketMq为什么快？

使用了顺序存储 PageCache 异步刷盘

写入commitlog是顺序写入的比随机写入快很多
写入commitlog不是直接写入磁盘而是写入操作系统的PageCache
由操作系统将Pagecache刷到磁盘

什么是事物消息、半事物消息，怎么实现的？

事务消息就是MQ提供的类似XA的分布式事务能力，通过事务消息可以达到分布式事务的最终一致性。
半事务消息就是MQ收到了生产者的消息，但是没有收到二次确认，不能投递的消息。
实现原理如下：

生产者先发送一条半事务消息到MQ
MQ收到消息后返回ack确认
生产者开始执行本地事务
如果事务执行成功发送commit到MQ，失败发送rollback
如果MQ长时间未收到生产者的二次确认commit或者rollback，MQ对生产者发起消息回查
生产者查询事务执行最终状态
根据查询事务状态再次提交二次确认
最终，如果MQ收到二次确认commit，就可以把消息投递给消费者，反之如果是rollback，消息会保存下来并且在3天后被删除。

redis多维度排序实现

zset 值用高低位组成
数组+链表实现
skipList实现

实际使用

生产者

@RabbitSettings(
        resourceCode = "BudgetProducer",
        exchangeName = ExchangeName.MARKETING_BUDGET
)
@ProducerSettings
public class BudgetProducer extends RabbitProducer {
}

配置

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface RabbitSettings {

    /**
     * resource code
     */
    String resourceCode() default "";

    /**
     * Exchange name.
     */
    String exchangeName();

    /***
     *
     * exchangeType
     */
    ExchangeType exchangeType() default ExchangeType.TOPIC;

    /**
     * Queue name.
     */
    String queueName() default "";

    /**
     * Message routing key.
     */
    String[] routingKey() default {};

    /***
     * Message publish Timeout
     *
     */
    int publishTimeoutMils() default -1;

    /***
     *
     * Connection in pool
     */
    int connectionPoolSize() default -1;

    /***
     *
     * channel in pool
     */
    int channelPoolSize() default -1;

    //todo cw check

    /**
     * 业务Q需要绑定的死信队列的Exchange
     */
    String xDeadLetterExchange() default "";

    /**
     * 业务Q消息转到死信队列的RoutingKey
     */
    String xDeadLetterRoutingKey() default "";

    int argsGray() default -1;

    String argsXMatch() default "";

}

消费者

@RabbitSettings(
        resourceCode = "RotatinonBudgetConsumer",
        exchangeName = "marketing_budget",
        routingKey = "marketing.budget.consume.rotation",
        queueName = "marketing.budget.consume.rotation.mq"

)
@ConsumerSettings(consumeMode = ConsumeMode.DELIVERY)
public class RotatinonBudgetConsumer extends RabbitConsumer {
}