MQ夺命11问

MQ夺命11问

为什么使用MQ

• 削峰填谷
• 异步调用

基于什么做的选型

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

所以选的RocketMq

消息可靠性保证？

生产者丢失

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

MQ丢失

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

消费者丢失

1. 默认消费者回复ACK确认

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

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

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

RocketMq实现原理

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

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

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

1. CAP理论,ZK满足CP,不能保证可用性 作为服务发现不能接受
2. 基于性能考虑 NameServer轻量 扩展方便 ZK的写是不可扩展的
3. ZK持久化机制问题 ZAB协议对每一个写请求会在每个 ZooKeeper 节点上保持写一个事务日志，同时再加上定期的将内存数据镜像（Snapshot）到磁盘来保证数据的一致性和持久性，而对于一个简单的服务发现的场景来说，这其实没有太大的必要
4. 消息发送应该弱依赖注册中心 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协议来进行的：

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

RocketMq为什么快？

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

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

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

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

1. 生产者先发送一条半事务消息到MQ
2. MQ收到消息后返回ack确认
3. 生产者开始执行本地事务
4. 如果事务执行成功发送commit到MQ，失败发送rollback
5. 如果MQ长时间未收到生产者的二次确认commit或者rollback，MQ对生产者发起消息回查
6. 生产者查询事务执行最终状态
7. 根据查询事务状态再次提交二次确认
   最终，如果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 {
}