在线脑图

消息队列提供的功能

• 异步处理
• 流量控制
• 消息解耦

队列和主题的区别

最初的消息队列，就是一个严格意义上的队列。在计算机领域，“队列（Queue）”是一种数据结构，有完整而严格的定义。在维基百科中，队列的定义是这样的：

队列是先进先出（FIFO, First-In-First-Out）的线性表（Linear List）。在具体应用中通常用链表或者数组来实现。队列只允许在后端（称为 rear）进行插入操作，在前端（称为 front）进行删除操作。

**早期的消息队列，就是按照“队列”的数据结构来设计的。**我们一起看下这个图，生产者（Producer）发消息就是入队操作，消费者（Consumer）收消息就是出队也就是删除操作，服务端存放消息的容器自然就称为“队列”。

这就是最初的一种消息模型：队列模型。

如果需要将一份消息数据分发给多个消费者，要求每个消费者都能收到全量的消息，例如，对于一份订单数据，风控系统、分析系统、支付系统等都需要接收消息。这个时候，单个队列就满足不了需求，一个可行的解决方式是，为每个消费者创建一个单独的队列，让生产者发送多份。

为了解决这个问题，演化出了另外一种消息模型：“发布 - 订阅模型（Publish-Subscribe Pattern）”。

在发布 - 订阅模型中，消息的发送方称为发布者（Publisher），消息的接收方称为订阅者（Subscriber），服务端存放消息的容器称为主题（Topic）。发布者将消息发送到主题中，订阅者在接收消息之前需要先“订阅主题”。“订阅”在这里既是一个动作，同时还可以认为是主题在消费时的一个逻辑副本，每份订阅中，订阅者都可以接收到主题的所有消息。

保障消息队列不丢失

其实，现在主流的消息队列产品都提供了非常完善的消息可靠性保证机制，完全可以做到在消息传递过程中，即使发生网络中断或者硬件故障，也能确保消息的可靠传递，不丢消息。

绝大部分丢消息的原因都是由于开发者不熟悉消息队列，没有正确使用和配置消息队列导致的。虽然不同的消息队列提供的 API 不一样，相关的配置项也不同，但是在保证消息可靠传递这块儿，它们的实现原理是一样的。

检测消息丢失的方法

• 分布式链路跟踪系统
• 根据生产者标示+分区号+递增消息号判断
• 生产者保存需要核对是否丢失的数据，消费者消费完之后需要与生产者核对数据

像 Kafka 和 RocketMQ 这样的消息队列，它是不保证在 Topic 上的严格顺序的，只能保证分区上的消息是有序的，所以我们在发消息的时候必须要指定分区，并且，在每个分区单独检测消息序号的连续性。

如果你的系统中 Producer 是多实例的，由于并不好协调多个 Producer 之间的发送顺序，所以也需要每个 Producer 分别生成各自的消息序号，并且需要附加上 Producer 的标识，在 Consumer 端按照每个 Producer 分别来检测序号的连续性。

Consumer 实例的数量最好和分区数量一致，做到 Consumer 和分区一一对应，这样会比较方便地在 Consumer 内检测消息序号的连续性。

如何确保消息不丢失

一条消息从生产到消费完成这个过程，可以划分三个阶段：

1. 生产阶段

   在生产阶段，消息队列通过最常用的请求确认机制，来保证消息的可靠传递：当你的代码调用发消息方法时，消息队列的客户端会把消息发送到 Broker，Broker 收到消息后，会给客户端返回一个确认响应，表明消息已经收到了。客户端收到响应后，完成了一次正常消息的发送。

   只要 Producer 收到了 Broker 的确认响应，就可以保证消息在生产阶段不会丢失。有些消息队列在长时间没收到发送确认响应后，会自动重试，如果重试再失败，就会以返回值或者异常的方式告知用户。

   你在编写发送消息代码时，需要注意，正确处理返回值或者捕获异常，就可以保证这个阶段的消息不会丢失。

2. 存储阶段

   在存储阶段正常情况下，只要 Broker 在正常运行，就不会出现丢失消息的问题，但是如果 Broker 出现了故障，比如进程死掉了或者服务器宕机了，还是可能会丢失消息的。

   如果对消息的可靠性要求非常高，可以通过配置 Broker 参数来避免因为宕机丢消息。

   对于单个节点的 Broker，需要配置 Broker 参数，在收到消息后，将消息写入磁盘后再给 Producer 返回确认响应，这样即使发生宕机，由于消息已经被写入磁盘，就不会丢失消息，恢复后还可以继续消费。例如，在 RocketMQ 中，需要将刷盘方式 flushDiskType 配置为 SYNC_FLUSH 同步刷盘。

   如果是 Broker 是由多个节点组成的集群，需要将 Broker 集群配置成：至少将消息发送到 2 个以上的节点，再给客户端回复发送确认响应。这样当某个 Broker 宕机时，其他的 Broker 可以替代宕机的 Broker，也不会发生消息丢失。

3. 消费阶段

   消费阶段采用和生产阶段类似的确认机制来保证消息的可靠传递，客户端从 Broker 拉取消息后，执行用户的消费业务逻辑，成功后，才会给 Broker 发送消费确认响应。如果 Broker 没有收到消费确认响应，下次拉消息的时候还会返回同一条消息，确保消息不会在网络传输过程中丢失，也不会因为客户端在执行消费逻辑中出错导致丢失。

   你在编写消费代码时需要注意的是，不要在收到消息后就立即发送消费确认，而是应该在执行完所有消费业务逻辑之后，再发送消费确认。

对于kafka的相关使用可以参考之前的一篇文章【消息队列(一)-如何解决消息丢失】

解决消息重复问题

消息重复的情况必然存在

在 MQTT 协议中，给出了三种传递消息时能够提供的服务质量标准，这三种服务质量从低到高依次是：

• At most once: 至多一次。消息在传递时，最多会被送达一次。换一个说法就是，没什么消息可靠性保证，允许丢消息。一般都是一些对消息可靠性要求不太高的监控场景使用，比如每分钟上报一次机房温度数据，可以接受数据少量丢失。
• At least once: 至少一次。消息在传递时，至少会被送达一次。也就是说，不允许丢消息，但是允许有少量重复消息出现。
• Exactly once：恰好一次。消息在传递时，只会被送达一次，不允许丢失也不允许重复，这个是最高的等级。

这个服务质量标准不仅适用于 MQTT，对所有的消息队列都是适用的。我们现在常用的绝大部分消息队列提供的服务质量都是 At least once，包括 RocketMQ、RabbitMQ 和 Kafka 都是这样。也就是说，消息队列很难保证消息不重复。

利用幂等性解决重复消息问题

1. 利用数据库的唯一约束实现幂等

   不光是可以使用关系型数据库，只要是支持类似“INSERT IF NOT EXIST”语义的存储类系统都可以用于实现幂等，比如，你可以用 Redis 的 SETNX 命令来替代数据库中的唯一约束，来实现幂等消费。

2. 为更新的数据设置前置条件

   另外一种实现幂等的思路是，给数据变更设置一个前置条件，如果满足条件就更新数据，否则拒绝更新数据，在更新数据的时候，同时变更前置条件中需要判断的数据。这样，重复执行这个操作时，由于第一次更新数据的时候已经变更了前置条件中需要判断的数据，不满足前置条件，则不会重复执行更新数据操作。
   但是，如果我们要更新的数据不是数值，或者我们要做一个比较复杂的更新操作怎么办？用什么作为前置判断条件呢？更加通用的方法是，给你的数据增加一个版本号属性，每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒绝更新数据，更新数据的同时将版本号 +1，一样可以实现幂等更新。

3. 记录并检查操作
   如果上面提到的两种实现幂等方法都不能适用于你的场景，还有一种通用性最强，适用范围最广的实现幂等性方法：记录并检查操作，也称为“Token 机制或者 GUID（全局唯一 ID）机制”，实现的思路特别简单：在执行数据更新操作之前，先检查一下是否执行过这个更新操作。

具体的实现方法是，在发送消息时，给每条消息指定一个全局唯一的 ID，消费时，先根据这个 ID 检查这条消息是否有被消费过，如果没有消费过，才更新数据，然后将消费状态置为已消费。

原理和实现是不是很简单？其实一点儿都不简单，在分布式系统中，这个方法其实是非常难实现的，在“检查消费状态，然后更新数据并且设置消费状态”中，三个操作必须作为一组操作保证原子性，才能真正实现幂等，否则就会出现 Bug。

对于这个问题，当然我们可以用事务来实现，也可以用锁来实现，但是在分布式系统中，无论是分布式事务还是分布式锁都是比较难解决问题。

对于kafka的相关使用可以参考之前的一篇文章【消息队列(二)-消息幂等】

处理消息积压

优化消息收发性能，预防消息积压的方法有两种，增加批量或者是增加并发，在发送端这两种方法都可以使用，在消费端需要注意的是，增加并发需要同步扩容分区数量，否则是起不到效果的。

对于系统发生消息积压的情况，需要先解决积压，再分析原因，毕竟保证系统的可用性是首先要解决的问题。快速解决积压的方法就是通过水平扩容增加 Consumer 的实例数量。

以前的代码，用于收集当前方法的所有参数，放在map中方便调取

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70

import com.google.common.collect.ImmutableMap;
import lombok.Data;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.IntStream;

@Aspect
@Component
public class ArgumentsCollector {

    private static final ThreadLocal<Map<String, Object>> ARGUMENTS = ThreadLocal.withInitial(ImmutableMap::of);

    static Map<String, Object> getArgs() {
        return ARGUMENTS.get();
    }

    private Object[] args(Object[] args, int exceptLength) {
        if (exceptLength == args.length) {
            return args;
        }

        return Arrays.copyOf(args, exceptLength);
    }

    @Pointcut("@annotation(CollectArguments)")
    void collectArgumentsAnnotationPointCut() {
    }

    @Before("collectArgumentsAnnotationPointCut()")
    public void doAccessCheck(JoinPoint joinPoint) {
        final String[] parameterNames = ((MethodSignature) joinPoint.getSignature()).getParameterNames();
        final Object[] args = args(joinPoint.getArgs(), parameterNames.length);

        ARGUMENTS.set(Collections.unmodifiableMap((IntStream.range(0, parameterNames.length)
                .mapToObj(idx -> Tuple2.of(parameterNames[idx], args[idx]))
                .collect(HashMap::new, (m, t) -> m.put(t.getT1(), t.getT2()), HashMap::putAll))));
    }

    @After("collectArgumentsAnnotationPointCut()")
    public void remove() {
        ARGUMENTS.remove();
    }

    @Data
    private static class Tuple2<T1, T2> {

        private T1 t1;
        private T2 t2;

        Tuple2(T1 t1, T2 t2) {
            this.t1 = t1;
            this.t2 = t2;
        }

        public static <T1, T2> Tuple2<T1, T2> of(T1 t1, T2 t2) {
            return new Tuple2<>(t1, t2);
        }
    }
}

附送一段代码，用于将方法中收集的参数转换成Bean

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import org.springframework.beans.BeanUtils;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.validation.DataBinder;

public class BinderUtil {

    BinderUtil() {
    }

    @SuppressWarnings("unchecked")
    public static <T> T getTarget(Class<T> beanClazz) {
        final DataBinder binder = new DataBinder(BeanUtils.instantiate(beanClazz));
        binder.bind(new MutablePropertyValues(ArgumentsCollector.getArgs()));
        return (T) binder.getTarget();
    }
}

使用实例：

1
2
3
4
5
6
7
8
9
10

@Override
@CollectArguments
public List<PsJobSequenceVO> findJobSequence(
        String jobSeqGroupId,
        String jobSeqId,
        Integer state,
        Date endDate
) {
    return jobSequenceHandler.findJobSequence(BinderUtil.getTarget(PsJobSequenceFindRO.class)).getData();
}

在线脑图

先来看表结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

CREATE TABLE a (
  `id`bigint AUTO_INCREMENT ,
  `a` int,
  `b` int,
  PRIMARY KEY (`id`),
  KEY `idx_a_b` (`a`,`b`)
);

CREATE TABLE b (
	`id`bigint AUTO_INCREMENT ,
  `b` int,
  `c` int,
  PRIMARY KEY (`id`)
)

看一下join语句，因为b上没有索引，所以mysql用的BLNJ：

1
2
3
4

explain select * from a 
join b using(b)
where a = 1
order by a, b;

如果b表有索引的话：

1
2
3
4
5
6
7

CREATE TABLE b (
	`id`bigint AUTO_INCREMENT ,
  `b` int,
  `c` int,
  PRIMARY KEY (`id`),
  KEY `idx_b` (`b`)
)

可以发现a表idx_a_b有序性没有利用上，至于原因，先看一下BNLJ执行的流程图:

执行过程为：

1. 扫描表 t1，顺序读取数据行放入 join_buffer 中，直到 join_buffer 满了，继续第 2 步；
2. 扫描表 t2，把 t2 中的每一行取出来，跟 join_buffer 中的数据做对比，满足 join 条件的，作为结果集的一部分返回；
3. 清空 join_buffer；
4. 继续扫描表 t1，顺序读取之后数据放入 join_buffer 中，继续执行第 2 步，直到所有数据读取完毕。

其中隐含的问题在于第二步：即使t1表的数据是有序读取到join_buffer中的，由于是先扫描t2表再关联join_buffer数据，导致join_buffer中的有序性失效。

如果表b有索引idx_b,那么使用BKA算法第二步的关联顺序与BNLJ相反，是先扫描join_buffer后通过索引关联t2,则可以利用join_buffer中的有序数据。

为了便于说明问题，我们先使用一个小一点儿的表，建表和初始化语句如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

CREATE TABLE `t` (
  `id` int(11) NOT NULL,
  `c` int(11) DEFAULT NULL,
  `d` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `c` (`c`)
) ENGINE=InnoDB;
 
insert into t values
(0,0,0),
(5,5,5),
(10,10,10),
(15,15,15),
(20,20,20),
(25,25,25);

这个表除了主键 id 外，还有一个索引 c，初始化语句在表中插入了 6 行数据。

下面的语句序列，是怎么加锁的，加的锁又是什么时候释放的呢？

1

select * from t where d = 5 for update;

比较好理解的是，这个语句会命中 d = 5 的这一行，对应的主键 id = 5，因此在 select 语句执行完成后，会在id = 5 这一行主键上加一个写锁，而且由于两阶段锁协议，这个写锁会在执行 commit 语句的时候释放。

由于字段 d 上没有索引，因此这条查询语句会做全表扫描。那么，其他被扫描到的，但是不满足条件的 5 行记录上，会不会被加锁呢？

我们知道，InnoDB 的默认事务隔离级别是可重复读，所以本文接下来没有特殊说明的部分，都是设定在可重复读隔离级别下。

幻读是什么？

现在，我们就来分析一下，假设只在 id = 5 这一行加锁，而其他行的不加锁的话，会怎么样。

下面先来看一下这个场景（这个结果是建立在前面假设之上，实际上是错误的）：

假设只在 id = 5 这一行加行锁，可以看到，session A 里执行了三次查询，分别是 Q1、Q2 和 Q3。它们的 SQL 语句相同，都是 select * fom t where d=5 for update。我们来看一下这三条 SQL 语句，分别会返回什么结果。

1. Q1 只返回 id = 5 这一行；
2. 在 T2 时刻，session B 把 id = 0 这一行的 d 值改成了 5，因此 T3 时刻 Q2 查出来的是 id = 0 和 id = 5 这两行；
3. 在 T4 时刻，session C 又插入一行（1,1,5），因此 T5 时刻 Q3 查出来的是 id = 0、id = 1 和 id = 5 的这三行。

其中，Q3 读到 id = 1 这一行的现象，被称为“幻读”。也就是说，幻读指的是一个事务在前后两次查询同一个范围的时候，后一次查询看到了前一次查询没有看到的行。

这里，我需要对“幻读”做一个说明：

1. 在可重复读隔离级别下，普通的查询是快照读，是不会看到别的事务插入的数据的。因此，幻读在当前读下才会出现。
2. 上面 session B 的修改结果，被 session A 之后的 select 语句用当前读看到，不能称为幻读。幻读仅专指新插入的行。

如果只从我们学到的事务可见性规则来分析的话，上面这三条 SQL 语句的返回结果都没有问题。

因为这三个查询都是加了 for update，都是当前读。而当前读的规则，就是要能读到所有已经提交的记录的最新值。并且，session B 和 sessionC 的两条语句，执行后就会提交，所以 Q2 和 Q3 就是应该看到这两个事务的操作效果，而且也看到了，这跟事务的可见性规则并不矛盾。

幻读有什么问题？

**首先是语义上的。**session A 在 T1 时刻就声明了，“我要把所有 d=5 的行锁住，不准别的事务进行读写操作”。所以我们假设只锁了id = 5这一行的语义与select * from t where d = 5 for update 不同。

其次，是数据一致性的问题。 **这个数据不一致到底是怎么引入的？**肯定是前面的假设有问题。

我们把扫描过程中碰到的行，也都加上写锁，再来看看执行效果。

由于 session A 把所有的行都加了写锁，所以 session B 在执行第一个 update 语句的时候就被锁住了。需要等到 T6 时刻 session A 提交以后，session B 才能继续执行。

这样对于 id = 0 这一行，在数据库里的最终结果还是 (0,5,5)。在 binlog 里面，执行序列是这样的：

1
2
3
4
5
6
7

insert into t values(1,1,5); /*(1,1,5)*/
update t set c=5 where id=1; /*(1,5,5)*/
 
update t set d=100 where d=5;/* 所有 d=5 的行，d 改成 100*/
 
update t set d=5 where id=0; /*(0,0,5)*/
update t set c=5 where id=0; /*(0,5,5)*/

可以看到，按照日志顺序执行，id = 0 这一行的最终结果也是 (0,5,5)。所以，id = 0 这一行的问题解决了。

但同时你也可以看到，id = 1 这一行，在数据库里面的结果是 (1,5,5)，而根据 binlog 的执行结果是 (1,5,100)，也就是说幻读的问题还是没有解决。为什么我们已经这么“凶残”地，把所有的记录都上了锁，还是阻止不了 id = 1 这一行的插入和更新呢？

原因很简单。在 T3 时刻，我们给所有行加锁的时候，id = 1 这一行还不存在，不存在也就加不上锁。

**也就是说，即使把所有的记录都加上锁，还是阻止不了新插入的记录，**这也是为什么“幻读”会被单独拿出来解决的原因。

如何解决幻读？

现在你知道了，产生幻读的原因是，行锁只能锁住行，但是新插入记录这个动作，要更新的是记录之间的“间隙”。因此，为了解决幻读问题，InnoDB 只好引入新的锁，也就是间隙锁 (Gap Lock)。

顾名思义，间隙锁，锁的就是两个值之间的空隙。比如文章开头的表 t，初始化插入了 6 个记录，这就产生了 7 个间隙。

这样，当你执行 select * from t where d=5 for update 的时候，就不止是给数据库中已有的 6 个记录加上了行锁，还同时加了 7 个间隙锁。这样就确保了无法再插入新的记录。

也就是说这时候，在一行行扫描的过程中，不仅将给行加上了行锁，还给行两边的空隙，也加上了间隙锁。

现在你知道了，数据行是可以加上锁的实体，数据行之间的间隙，也是可以加上锁的实体。但是间隙锁跟我们之前碰到过的锁都不太一样。

比如行锁，分成读锁和写锁。下图就是这两种类型行锁的冲突关系。

也就是说，跟行锁有冲突关系的是“另外一个行锁”。

但是间隙锁不一样，**跟间隙锁存在冲突关系的，是“往这个间隙中插入一个记录”这个操作。**间隙锁之间都不存在冲突关系。

这句话不太好理解，我给你举个例子：

这里 session B 并不会被堵住。因为表 t 里并没有 c = 7 这个记录，因此 session A 加的是间隙锁 (5,10)。而 session B 也是在这个间隙加的间隙锁。它们有共同的目标，即：保护这个间隙，不允许插入值。但，它们之间是不冲突的。

间隙锁和行锁合称 next-key lock，每个 next-key lock 是前开后闭区间。也就是说，我们的表 t 初始化以后，如果用 select * from t for update 要把整个表所有记录锁起来，就形成了 7 个 next-key lock，分别是 (-∞,0]、(0,5]、(5,10]、(10,15]、(15,20]、(20, 25]、(25, +supremum]。

备注：这篇文章中，如果没有特别说明，我们把间隙锁记为开区间，把 next-key lock 记为前开后闭区间。

你可能会问说，这个 supremum 从哪儿来的呢？

这是因为 +∞是开区间。实现上，InnoDB 给每个索引加了一个不存在的最大值 supremum，这样才符合我们前面说的“都是前开后闭区间”。

间隙锁和 next-key lock 的引入，帮我们解决了幻读的问题，但同时也带来了一些“困扰”。

对应到我们这个例子的表来说，业务逻辑这样的：任意锁住一行，如果这一行不存在的话就插入，如果存在这一行就更新它的数据，代码如下：

1
2
3
4
5
6
7
8
9

begin;
select * from t where id=N for update;
 
/* 如果行不存在 */
insert into t values(N,N,N);
/* 如果行存在 */
update t set d=N set id=N;
 
commit;

这个逻辑一旦有并发，就会碰到死锁。你一定也觉得奇怪，这个逻辑每次操作前用 for update 锁起来，已经是最严格的模式了，怎么还会有死锁呢？

这里，我用两个 session 来模拟并发，并假设 N=9。

图 8 间隙锁导致的死锁

你看到了，其实都不需要用到后面的 update 语句，就已经形成死锁了。我们按语句执行顺序来分析一下：

1. session A 执行 select … for update 语句，由于 id = 9 这一行并不存在，因此会加上间隙锁 (5,10);
2. session B 执行 select … for update 语句，同样会加上间隙锁 (5,10)，间隙锁之间不会冲突，因此这个语句可以执行成功；
3. session B 试图插入一行 (9,9,9)，被 session A 的间隙锁挡住了，只好进入等待；
4. session A 试图插入一行 (9,9,9)，被 session B 的间隙锁挡住了。

至此，两个 session 进入互相等待状态，形成死锁。当然，InnoDB 的死锁检测马上就发现了这对死锁关系，让 session A 的 insert 语句报错返回了。

你现在知道了，间隙锁的引入，可能会导致同样的语句锁住更大的范围，这其实是影响了并发度的。

你可能会说，为了解决幻读的问题，我们引入了这么一大串内容，有没有更简单一点的处理方法呢。

我在文章一开始就说过，如果没有特别说明，今天和你分析的问题都是在可重复读隔离级别下的，间隙锁是在可重复读隔离级别下才会生效的。所以，你如果把隔离级别设置为读提交的话，就没有间隙锁了。但同时，你要解决可能出现的数据和日志不一致问题，需要把 binlog 格式设置为 row。这，也是现在不少公司使用的配置组合。

• 幸存者偏差：高估了成功的概率
• 看清楚自己
• 是否高估自己
• 是否陷入从众心理
• 是否与沉没成本难舍难分
• 是否是互惠互利陷阱
• 摆脱确认偏差，不要证明自己是正确的
• 是否屈服于权威
• 是否依据对比来判断（请不要怎么做）
• 现成偏见
• 是否故意营造一种危机气氛
• 明确判断他人的能力范围
• 警惕控制错觉
• 警惕激励过敏
• 是否是平均值会起作用？
• 警惕公地悲剧
• 切勿只以结果判断决定
• 面对选择，遵守自己的标准
• 排除对他人好感的影响
• 禀赋效应 let it go
• 无巧不成书，如果从不发生才令人感到意外
• 唱反调的人也许是最重要的人
• 0风险偏误
• 稀有性谬论
• 概率偏误-前车之鉴后事之师
• 独立事件-赌徒谬论
• 锚定效应，你的锚在哪里？
• 警惕归纳法，警惕黑天鹅事件
• 厌恶风险，但不要无视损失
• 社会性懈怠
• 赢家诅咒-你会得到什么、失去什么
• 是否高估了人的影响
• 相互关系不等同于因果关系
• 避免光环效应过度影响总体印象
• 合理不一定真实
• 不要被框架效应限制注意力
• 遇到不明情况，就会发生行动偏误
• 不作为偏误，如果不解决问题，你就是问题的一部分
• 自利偏误，为什么你从来不自责
• 避免长时间负面影响
• 自我选择偏误
• 联想偏误，不要欠拟合也不要过拟合
• 认知失调，是否在自我安慰
• 即使行乐，只限周末，不要为了眼前的利益破坏未来的利益

第四章 Schema设计

选择优化的数据类型

更小的通常更好，但是要确保没有低估需要存储的值的范围，因为在schema中的多个地方增加数据类型范围是一个十分耗时的操作。

简单就好

简单的数据类型的操作通常需要更少的cpu时间。

尽量避免NULL

可为NULL的列使得索引、索引比统计和值比较都更为复杂。

当然也有例外，InnoDB使用单独的位存储NULL值， 所以对于稀疏数据有很好的空间效率。

选择标识符

一旦选定一种类型，要确保在所有的关联表中都使用同样的类型。类型之间需要精确匹配，包括像UNSIGNED这样的属性。尽量只用整型定义标识符。

注意可变长字符串

其在临时表和排序时可能导致悲观的按最大长度分配内存

范式与反范式

范式是好的，但是反范式有时也是必须的，并且能带来好处。

第五章 创建高性能索引

B-Tree 索引的查询类型

• 全值匹配： 指的是和索引的所有列进行匹配
• 匹配最左前缀： 查找索引前几列进行匹配
• 匹配列前缀: 只匹配某一列的值的开头部分
• 匹配范围值： 查找索引某一范围的值
• 精确匹配某一列并范围匹配另外一列
• 只访问索引的查询：覆盖索引

B-Tree 索引的限制

• 如果不是按照索引的最左列开始查找，则无法使用索引

• 不能跳过索引中的列

• 如果查询中有某个列的范围查询，则其右边的所有列都无法使用索引：

  例如查询 索引为key(last_name, fisrt_name, dob)

  1	where last_name = 'a' and first_name like 'J%' and dob = '1877-12-23'

索引的优点

1. 大大减少服务器需要扫描的数据量
2. 帮助服务器避免排序和临时表
3. 将随机I/O变为顺序I/O

高性能索引-独立的列

如果查询中列不是独立的，则mysql不会使用索引

1
2

select actor_id from sakila.actor where actor_id + 1 = 5;

高性能索引-前缀索引和索引选择性

有时候需要索引很长的字符列，通常可以索引开始部分的字符，同时也会降低索引的选择性。

索引的选择性指的是，不重复的索引值和数据表的记录总数的比值。索引的选择性越高表示查询效率越高，因为选择性高的索引可以过滤掉更多的行。

前缀索引是一种能使索引更小，更快的有效方法，但是也有其缺点：前缀索引无法做order by 和 group by，也无法使用前缀索引做覆盖索引。

高性能索引-多列索引

最容易遇到的困惑是多列索引的顺序，正确的顺序依赖于使用索引的查询，同时需要考虑如何更好的满足排序和分组需要。对于如何选择索引的列顺序有一个经验法则：将选择性最高的列放在索引的最前列。只有不需要考虑排序和分组时，将选择性跟高的列放在最前面通常是最好的，但是考虑问题需要更全面，避免随机I/O和排序更加重要。

高性能索引-覆盖索引

如果一个索引包含所需要查询的字段的值，我们就可以称之为“覆盖索引”

覆盖索引的好处：

• 索引条目通常远小于数据行大小，所以如果只需要读取索引，那mysql就会极大的减少数据访问量。
• 因为索引是按照列值顺序存储的，所以对于I/O密集型范围查询回避随机从磁盘读取每一行数据的I/O要小的多
• 由于Innodb的聚簇索引，覆盖索引对Innodb表特别有用，可以避免对主键索引的二次查询。

覆盖索引的陷阱：

1

select * from products where actor = 'SEAN CARREY' and title like '%APOLLO%';

• 没有索引能够覆盖这个查询，因为查询从表中选择了所有的列
• mysql不能再索引中执行like操作，只能做最左前缀匹配

高性能索引-使用索引扫描来做排序

mysql有两种方式可以生成有序的结果：通过排序操作；或者使用索引顺序扫描。mysql可以使用同一个索引既满足排序，有用于查找行。

只有当索引的列顺序和ORDER BY子句的顺序完全一致，并且所有列的排序方向都一样时，mysql才能使用索引来对结果做排序。如果查询需要关联多张表，则只有当ORDER BY子句引用的字段全部为第一个表时，才能用索引做排序。ORDER BY子句和查询的限制是一样的：需要满足索引的最左前缀的要求。

有一种情况下ORDER BY子句可以不满足索引的最左前缀要求：

1
2

select rental_id, staff_id from sakila.rental where rental_date = '2005-05-25'
order by inventory_id, customer_id;

索引为key(rental_date, inventory_id, customer_id)，前导列为常量的时候，如果where子句或者join子句中对这些列指定了常量，就可以弥补ORDER BY的不足。

1
2

where rental_date > '2005-12-25' order by inventory_id, customer_id;
where rental_date = '2005-12-25' and inventory_id in (1, 2) order by cusomter_id;

对于索引上是范围查询，mysql无法使用之后的索引列

高性能索引-使用索引扫描减少锁

索引可以让查询锁定更少的行，如果你的查询从不访问那些不需要的行，那么就会锁定更少的行。但这只有当innoDB在存储引擎层能够过滤掉所有不需要的行是才有效。如果索引无法过滤掉无效的行，那么innoDB检索到数据并返回给服务器层后，innoDB已经锁定这些行了（mysql 5.6后没有这个问题）。

高性能索引-避免多个范围条件

下面的查询：

1

where last_online > date_sub(now(), interval 7 day) and age bwtween 18 and 25

这个查询有一个问题：它有两个范围条件,last_online和age列，mysql可以使用last_online的索引或者是age列的索引，但是无法同时使用它们。

高性能索引-延迟关联优化分页

如果一个查询匹配结果有上百万行的话会怎样？

1

select * from profiles where sex = 'm' order by rating limit 10;

即使有索引，如果用户界面需要翻页，并且翻页到比较靠后的地方也会非常慢，如：

1

select * from profiles where sex = 'm' order by rating limit 1000000, 10;

无论如何创建索引，这种查询都是个严重的问题，mysql需要花费大量时间来扫描需要丢弃的数据。其中一个解决的办法是限制能够翻页的数量。

优化这类索引的另一个比较好的策略是使用延迟关联，通过使用覆盖索引返回需要的主建，再根据这些主建回主表获得需要的行，这样可以减少mysql扫描需要丢弃的行数。

1
2
3
4
5
6

select * from profiles innner join 
(
    select id fomr profiles p where p.sex = 'm' order by rating limit 1000000, 10

) as t using (id);

第六章 慢查询优化

优化数据访问

查询性能低下最基本的原因是访问数据太多。某些查询可能不可避免的需要筛选大量数据，单这并不常见。大部分性能低下的查询都可以通过减少访问的数据量的方式进行优化。对于低效的查询，我们发现通过下面几个步骤来分析总是很有效：

1. 确认应用程序时候检索大量超过需要的数据。
2. 确认mysql服务层是否在分许大量超过需要的数据行。

第一种情况可以使用limit和选择需要的列来解决。在确定查询只返回需要的数据之后，接下来应该看看查询为了返回结果是否扫描了过多的数据，对于mysql有三个衡量查询开销的指标：

• 响应时间
• 扫描的行数
• 返回的行数

没有那个指标能完美地衡量查询的开销，但它们大致反映了mysql内部查询时需要访问多少数据，并可以大概推算出查询运行的时间。

在评估查询开销的时候，需要考虑一下从表中找到某一行数据的成本。mysql有好几种访问方式可以查找并返回一行结果。有些访问方式可能需要扫描很多行才能返回一行结果，也有些访问方式可能无需扫描就能返回结果。

在explain语句中的type列反应了访问类型。访问类型有很多种，从全表扫描到索引扫描、范围扫描、唯一索引查询、常数引用等。这里列的这些，速度是从慢到快，扫描行数也是从大到小。如果查询没有办法找到合适是访问类型，那么解决的最好办法通常是增加一个合适的索引。

一般mysql能够使用如下三种方式应用where条件，从好到坏依次为：

• 从索引中使用where条件吗来过滤不匹配的记录，这是在存储引擎层完成的。
• 使用索引覆盖扫描来返回记录（extra出现using index），直接从索引中过滤不需要的数据并返回命中结果。这是在mysql服务层完成的，但无需再回表查询记录。
• 从数据表中返回数据，然后过滤不满足条件的记录（extra出现using where）。这是在mysql服务器层完成，mysql需要从数据表中读出记录然后过滤。

如果发现查询需要扫描大量的数据但只返回少数的行，那么可以使用下面的技巧去优化它：

• 使用覆盖索引，把需要用的列都放到索引中
• 改变库表结构。例如使用单独的汇总表
• 重写复杂的查询， 让mysql优化器能够以更加高效的方式执行这个查询

重构查询方式-切分查询

有时候对于一个大查询我们需要分而治之，将大查询分成小查询。删除旧数据是一个很好的例子，如果用一个大的语句一次性完成，则可能一次需要锁住很多数据、占满整个事务日志、耗尽系统资源、阻塞恨到重要的查询。

重构查询方式-分解关联查询

例如下面这个查询：

1
2
3
4

select * from tag
join tag_post on tag_post.tag_id = tag.id
join post on tag_post.post_id = post.id
where tag.tag = 'mysql';

可以分解成下面这个查询来代替：

1
2
3
4
5

select * from tag where tag = 'mysql';

select * from tag_post where tag_id = 1234;

select * from post where post.id in (123,456,567);

用分解关联查询的方式重构查询有如下的优势：

• 让缓存效率更高。许多应用程序可以方便地缓存单表查询对应的结果对象。
• 将查询分解后，执行单个查询可以减少锁的竞争。
• 在应用层做关联，可以更容易对数据库进行拆分，更容易做到高性能和高扩展。
• 查询本身效率可能提升，使用in代替关联查询，可能比随机关联更高效。
• 减少冗余记录的查询
• 相当于在应用层实现了哈希关联

重构查询方式-优化关联查询

• 确保on或者using子句上的列上有索引。
• 确保任何group by和 order by 的表达式中只涉及到一个表中的列，这样mysql才有可能使用索引来优化这个过程

因为字符比较是从右往左比较的，所以第一层循环 needle.length + 1 <= i < haystack.length。

1
2
3
4
5
6
7
8
9
10
11
12
13

                  start=needle.length - 1                                             end=haystack.length - 1
                          +                                                                   +
                          |                                                                   |
                          |                                                                   |
                          v-------------------------------------------------------------------v->
  0   1   2   3   4   5   6   7   8   9  10  11  12  13  14  15  16  17  18  19  20  21  22  23
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
| t | h | i | s |   | i | s |   | a |   | s | i | m | p | l | e |   | e | x | a | m | p | l | e |
+-------------------------------+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
+---------------------------+
| e | x | a | m | p | l | e |
+---+---+---+---+---+---+---+

第二层循环中i变量表示坏字符的位置、j表示搜索坏字符开始位置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

                          i
                          +
                          |
                          v
  0   1   2   3   4   5   6   7   8   9  10  11  12  13  14  15  16  17  18  19  20  21  22  23
+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
| t | h | i | s |   | i | s |   | a |   | s | i | m | p | l | e |   | e | x | a | m | p | l | e |
+-------------------------------+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
+---------------------------+
| e | x | a | m | p | l | e |
+---+---+---+---+---+---+-+-+
                          ^
                          |
                          +
                          j

i,j指针向左搜索,如果完全匹配直接返回i即可

1
2
3
4
5

for (j = needle.length - 1; needle[j] == haystack[i]; --i, --j) {
    if (j == 0) {
        return i;
    }
}

坏字符规则和好字符规则

坏字符:从右向左第一个不匹配的字符
好字符:从坏字符下一个字符直到最后的字符

可以认为字符移动有两种策略：坏字符对齐和好字符对齐，然后选择字符移动距离大的策略即可

wiki实现

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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

/**
 * Returns the index within this string of the first occurrence of the
 * specified substring. If it is not a substring, return -1.
 *
 * There is no Galil because it only generates one match.
 *
 * @param haystack The string to be scanned
 * @param needle The target string to search
 * @return The start index of the substring
 */
public static int indexOf(char[] haystack, char[] needle) {
    if (needle.length == 0) {
        return 0;
    }
    int charTable[] = makeCharTable(needle);
    int offsetTable[] = makeOffsetTable(needle);
    for (int i = needle.length - 1, j; i < haystack.length;) {
        for (j = needle.length - 1; needle[j] == haystack[i]; --i, --j) {
            if (j == 0) {
                return i;
            }
        }
        // i += needle.length - j; // For naive method
        i += Math.max(offsetTable[needle.length - 1 - j], charTable[haystack[i]]);
    }
    return -1;
}

/**
 * Makes the jump table based on the mismatched character information.
 */
private static int[] makeCharTable(char[] needle) {
    final int ALPHABET_SIZE = Character.MAX_VALUE + 1; // 65536
    int[] table = new int[ALPHABET_SIZE];
    for (int i = 0; i < table.length; ++i) {
        table[i] = needle.length;
    }
    for (int i = 0; i < needle.length - 2; ++i) {
        table[needle[i]] = needle.length - 1 - i;
    }
    return table;
}

/**
 * Makes the jump table based on the scan offset which mismatch occurs.
 * (bad character rule).
 */
private static int[] makeOffsetTable(char[] needle) {
    int[] table = new int[needle.length];
    int lastPrefixPosition = needle.length;
    for (int i = needle.length; i > 0; --i) {
        if (isPrefix(needle, i)) {
            lastPrefixPosition = i;
        }
        table[needle.length - i] = lastPrefixPosition - i + needle.length;
    }
    for (int i = 0; i < needle.length - 1; ++i) {
        int slen = suffixLength(needle, i);
        table[slen] = needle.length - 1 - i + slen;
    }
    return table;
}

/**
 * Is needle[p:end] a prefix of needle?
 */
private static boolean isPrefix(char[] needle, int p) {
    for (int i = p, j = 0; i < needle.length; ++i, ++j) {
        if (needle[i] != needle[j]) {
            return false;
        }
    }
    return true;
}

/**
 * Returns the maximum length of the substring ends at p and is a suffix.
 * (good suffix rule)
 */
private static int suffixLength(char[] needle, int p) {
    int len = 0;
    for (int i = p, j = needle.length - 1;
             i >= 0 && needle[i] == needle[j]; --i, --j) {
        len += 1;
    }
    return len;
}