好像一切都有所安排。

周一在公司收到了上周日买的两本的书，《圆圈正义》和《杀死一只知更鸟》。

周三晚回家顺手把《杀死一只知更鸟》也带回了住处，想着说晚上也能看看。

周四一如既往，工作完充电器放公司电脑带回家。

好像一切都有所安排，晚上洗澡不小心滑了一下。这一滑，直接导致周五请假一天。到现在还清楚的记得周四下班和同事说羡慕周五能请假休息一天，没想到自己也”实现了周五休息一天”。

周五晚看着电脑所剩不多的电和同事说明天去拿充电器（在周四晚出事后提前说好了让同事下班帮忙带一下）

被告知周六要加班。现在是周五23点46分，外面下着阵雨，因为行动不便就没选择现在去拿充电器。

突然想起来周六咋过呢？看了一眼书桌，拿起了周三带回的书。

好像一切都有所安排。

对于 MQ 来说，其实不管是 RocketMQ、Kafka 还是其他消息队列，它们的本质都是：一发一存一消费。下面我们先以这个本质作为根，一起由浅入深地聊聊 MQ。

从MQ的本质说起

原始模型

将 MQ 掰开了揉碎了来看，都是一发一存一消费，再通俗点就是一个转发器。

生产者先将消息投递到一个叫做队列的容器中，消费者不断从这个容器中取出消息^1进行消费，仅此而已。

上图便是消息队列最原始的模型，它包含了两个关键词：消息和队列。

1、消息：就是要传输的数据，可以是最简单的字符串，也可以是任何自定义的复杂格式（只要能按预定格式解析（序列化/反序列化^2）出来即可）。
2、队列：一种先进先出数据结构。它是存放消息的容器，消息从队尾入队，从队头出队，入队即发消息的过程，出队即收消息的过程。

原始模型的演进

再看今天我们最常用的消息队列产品（RocketMQ、Kafka 等等），你会发现：它们都在最原始的消息模型上做了扩展，同时提出了一些新名词，比如：主题（topic）、分区^3（partition）、生产/消费组（producer/consumer group）等等。

要彻底理解这些新概念，先从消息模型的演进说起（道理好比：架构从来不是设计出来的，而是演进来的）。

队列模型

最初的消息队列就是上一节讲的原始模型，它是一个严格意义上的队列（Queue）。消息按照什么顺序写进去，就按照什么顺序读出来。不过，队列没有 “读” 这个操作，读就是出队，从队头中 “删除” 这个消息。

这便是队列模型：它允许多个生产者往同一个队列发送消息。但是，如果有多个消费者，实际上是竞争的关系，也就是一条消息只能被其中一个消费者接收到，消费完就不能再次被消费。

发布-订阅模型

如果需要将一份消息数据分发给多个消费者，并且每个消费者都要求收到全量的消息。很显然，队列模型无法满足这个需求。

一个可行的方案是：为每个消费者创建一个单独的队列，让生产者发送多份。这种做法比较笨，而且同一份数据会被复制多份，也很浪费空间。

为了解决这个问题，就演化出了另外一种消息模型：发布-订阅模型。

在发布-订阅模型中，存放消息的容器变成了 “主题”，消费者在接收消息之前需要先 “订阅主题”。最终，每个消费者都可以收到同一个主题的全量消息。

仔细对比下它和 ”队列模型” 的异同：发布消息的是生产者，主题即消息的容器，消费者即订阅者，无本质区别。唯一的不同点在于：一份消息数据是否可以被多次消费。

小结

上面两种模型说白了就是：单播和广播^4。并且当 发布-订阅模型 中只有一个订阅者时，它和队列模型就一样了。

这也解释了为什么现在主流的 RocketMQ、Kafka 都是基于发布-订阅模型实现的？至于各类产品中的生产/消费组^5、分区、消费模式、消息回溯、延迟消费等等衍生概念都是在上述原始模型的基础上，为了解决某些问题而产生的。

通过MQ本质看应用场景

回顾一下上面消息队列的原始模型，不难发现消息队列的这两个特性：

• 在没有消息队列时，通常生产者作为调用方、消费者作为被调方，消息通过一次RPC[^6]直接进行传达并被“消费”。引入消息队列后，一次RPC变成了两次RPC，生产者和消费者之间无需知道对方的存在。

• 多了一个中间节点（消息队列）转储消息，原本的同步操作变成了异步。

再回想起倒背如流的消息队列应用场景：系统解耦、异步通信、流量削峰、内容分发等等，无非是利用了上面两个特性。

那么引入消息队列后就没有坏处吗？

当然不。鱼和熊掌不可兼得，在某些方面获得便利的同时也引入了一些问题：

1、消息的及时性问题，通常而言业务对消息的及时性要求不高。

2、引入额外的组件，系统复杂度上升，稳定性下降。

3、因为网络波动和异步，会存在消息的顺序和重复问题。

4、生产者发送失败和消费者消费失败的处理带来的一致性的问题。

纵观目前主流的消息队列，也并没有完全解决这些问题，根据侧重不同平衡利弊，所以才有了现在诸多的消息队列。那目前主流的消息队列是如何平衡和解决这些问题，内部实现又是如何呢？下面我们就以RocketMQ为例，探究一下内部设计和实现细节。

RocketMQ

原始队列模型毕竟简单，无法满足真实的应用场景，所以目前主流的消息队列都引入了一些额外的概念来解决这些泛问题，描述的名词可能不同但作用总体上相差无几。所以先来了解一下RocketMQ中的概念[^7]，以便更好的理解后文内容。

RocketMQ中的概念

RocketMQ核心由四部分组成，分别是NameServer、Broker、Producer以及Consumer。

通过上图可以看到，这四部分都可集群部署，这是RocketMQ吞吐量大、可用性高的原因之一。Broker的集群部署模式也有很多种：支持多Master模式、多Master多Slave异步复制模式、多Master多Slave同步复制模式。

NameServer

NameServer是一个功能齐全的服务，其角色类似Dubbo中的Zookeeper，但NameServer与Zookeeper相比更轻量。主要是因为NameServer被设计成几乎无状态的，节点之间相互无通信，通过部署多个NameServer并在客户端做可用性轮训，实现了一个伪集群的高可用。

NameServer压力不会太大，平时主要开销是在维持心跳和提供Topic-Broker的关系数据。

Broker

Broker是具体提供业务的服务，负责存储消息，转发消息。每个Broker启动时都会向NameServer注册自己的信息，并与NameServer保持长链接及心跳，定时将Topic信息更新到NameServer。

有一点需要注意，Broker向NameServer发送心跳时， 会带上当前自己所负责的所有Topic信息，如果Topic个数太多（万级别），会导致一次心跳中Topic的数据就几十M，网络情况差的话， 网络传输失败，心跳失败，导致NameServer误认为Broker心跳失败。

Producer

Producer是指具体生产消息的服务，即生产者。生产者在启动时会。从支持三种消息发送方式：

• 单向发送：消息发出去后，可以继续发送下一条消息或执行业务代码，不需要等待服务器回应，且没有回调函数[^8]。
• 异步发送：消息发出去后，可以继续发送下一条消息或执行业务代码，不等待服务器回应，有回调函数。
• 同步发送：消息发出去后，等待服务器响应成功或失败，才能继续后面的操作。

Consumer

Consumer是指具体消费消息的服务，即消费者。消费者在启动时会从NameServer拉取Topic及Queue信息，根据负载均衡策略选择消费的Queue，支持集群消费和广播消费，可提供近实时的消息订阅机制。RocketMQ中提供了两张消费API，即Pull和Push两种：

• Pull，拉取型消费。消费者主动从Broker中拉去消息进行消费，只要拉取到消息，用户应用就会启动消费过程，所以 Pull 也称主动消费类型。

• Push，推送式消费。RocketMQ Client封装了消息的拉取、消费进度和其他的内部维护工作，将消息到达后执行的回调接口留给用户应用程序来实现。所以 Push 称为被动消费类型，但从实现上看还是基于 Pull 模式，不同于 Pull 的是 Push 首先要注册回调函数，当拉取到消息后触发回调函数开始消费消息。

RocketMQ关键特性及实现原理

分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件，需要具有高吞吐量、高可用等特点。而谈到消息系统的设计，就回避不了两个问题：

1. 消息的顺序问题
2. 消息的重复问题

RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件，它是怎样来解决这两个问题的？

顺序消息

消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了 3 条消息，分别是订单创建、订单付款、订单完成。消费时，要按照这个顺序消费才有意义。但同时多个订单之间同类的消息又是可以并行消费的。

假如生产者产生了2条消息：M1、M2，要保证这两条消息的顺序，应该怎样做？你脑中想到的可能是这样：

M1发送到S1后，M2发送到S2，如果要保证M1先于M2被消费，那么需要M1到达消费端后，通知S2，然后S2再将M2发送到消费端。

这个模型存在的问题是，如果M1和M2分别发送到两台Broker上，就不能保证M1一定先达到，也就不能保证M1被先消费，那么就需要在MQ Broker集群维护消息的顺序。那么如何解决？一种简单的方式就是将M1、M2发送到同一个Broker上：

这样可以保证M1先于M2到达MQ Broker（客户端等待M1成功后再发送M2），根据队列先进先出的原则，M1会先于M2被消费，这样就保证了消息的顺序。

这个模型，理论上可以保证消息的顺序，但在实际运用中你应该会遇到下面的问题：

网络延迟问题。只要将消息从一台服务器发往另一台服务器，就会存在网络延迟问题。如上图所示，如果发送M1耗时大于发送M2的耗时，那么M2就先被消费，仍然不能保证消息的顺序。即使M1和M2同时到达消费端，由于不清楚消费端1和消费端2的负载情况，仍然有可能出现M2先于M1被消费。如何解决这个问题？将M1和M2发往同一个消费者即可，且发送M1后，需要消费端响应成功后才能发送M2。

但又会引入另外一个问题，如果发送M1后，消费端1没有响应，那是继续发送M2呢，还是重新发送M1？一般为了保证消息至少被消费一次[^9]，肯定会选择重发M1到另外一个消费端2，就如下图所示。

这样的模型就严格保证消息的顺序，细心的你仍然会发现问题，消费端1没有响应Server时有两种情况，一种是M1确实没有到达，另外一种情况是消费端1已经响应，但是Server端没有收到。如果是第二种情况，重发M1，就会造成M1被重复消费。也就是我们后面要说的第二个问题，消息重复问题。

回过头来看消息顺序问题，严格的顺序消息非常容易理解，而且处理问题也比较容易，要实现严格的顺序消息，简单且可行的办法就是：

保证 生产者 - MQBroker - 消费者 是一对一对一的关系。

但是这样设计，消息系统的吞吐量就大打折扣，也会导致更多的异常处理，比如：只要消费端出现问题，就会导致当前流程阻塞，又不得不花费更多的精力来解决阻塞问题。但消息队列需要做到高吞吐量、高容错性。这似乎是一对不可调和的矛盾，那么RocketMQ是如何解决这个问题的呢？

首先可以明确的是RocketMQ并没有实现严格意义上的顺序消息。有些问题，看起来很重要，但实际上我们可以通过合理的设计或者将问题分解来规避。如果硬要把时间花在解决它们身上，实际上是浪费的，效率低下的。从这个角度来看消息的顺序问题，我们可以得出两个结论：

1、不关注乱序的应用实际大量存在
2、队列无序并不意味着消息无序

最后我们从源码角度分析RocketMQ怎么实现发送顺序消息。

1
2
3
4
5
6
7
8
9

// RocketMQ默认提供了两种MessageQueueSelector实现：随机/Hash
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    }
}, orderId);

在获取到路由信息以后，会根据MessageQueueSelector实现的算法来选择一个队列，同一个OrderId获取到的队列是同一个队列。

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

private SendResult send()  {
    // 获取topic路由信息
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    if (topicPublishInfo != null && topicPublishInfo.ok()) {
        MessageQueue mq = null;
        // 根据我们的算法，选择一个发送队列
        // 这里的arg = orderId
        mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
        if (mq != null) {
            return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
        }
    }
}

消息重复

上面在解决消息顺序问题时，引入了一个新的问题，就是消息重复。那么RocketMQ是怎样解决消息重复的问题呢？还是“恰好”不解决。

造成消息的重复的根本原因是：网络不可达。只要通过网络交换数据，就无法避免这个问题。所以解决这个问题的办法就是不解决，转而绕过这个问题。那么问题就变成了：如果消费端收到两条一样的消息，应该怎样处理？

1、消费端处理消息的业务逻辑保持幂等性
2、保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现

第1条很好理解，只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID，如果新到的消息ID已经在日志表中，那么就不再处理这条消息。

我们可以看到第1条的解决方式，很明显应该在消费端实现，不属于消息系统要实现的功能。第2条可以消息系统实现，也可以业务端实现。正常情况下出现重复消息的概率不一定大，且由消息系统实现的话，肯定会对消息系统的吞吐量和高可用有影响，所以最好还是由业务端自己处理消息重复的问题，这也是RocketMQ不解决消息重复的问题的原因。

RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重。

发送消息

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡，如下图所示：

首先分析一下RocketMQ的客户端发送消息的源码：

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

// 构造Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 初始化Producer，整个应用生命周期内，只需要初始化1次
producer.start();
// 构造Message
Message msg = new Message("TopicTest1",// topic
                        "TagA",// tag：给消息打标签,用于区分一类消息，可为null
                        "OrderID188",// key：自定义Key，可以用于去重，可为null
                        ("Hello MetaQ").getBytes());// body：消息内容
// 发送消息并返回结果
SendResult sendResult = producer.send(msg);
// 清理资源，关闭网络连接，注销自己
producer.shutdown();

在整个应用生命周期内，生产者需要调用一次start方法来初始化，初始化主要完成的任务有：

1. 如果没有指定namesrv地址，将会自动寻址
2. 启动定时任务：更新namesrv地址、从namsrv更新topic路由信息、清理已经挂掉的broker、向所有broker发送心跳…
3. 启动负载均衡的服务

初始化完成后，开始发送消息，发送消息的主要代码如下：

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

private SendResult sendDefaultImpl(Message msg,......) {
    // 检查Producer的状态是否是RUNNING
    this.makeSureStateOK();
    // 检查msg是否合法：是否为null、topic,body是否为空、body是否超长
    Validators.checkMessage(msg, this.defaultMQProducer);
    // 获取topic路由信息
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    // 从路由信息中选择一个消息队列
    MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
    // 将消息发送到该队列上去
    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}

代码中需要关注的两个方法tryToFindTopicPublishInfo和selectOneMessageQueue。前面说过在producer初始化时，会启动定时任务获取路由信息并更新到本地缓存，所以tryToFindTopicPublishInfo会首先从缓存中获取topic路由信息，如果没有获取到，则会自己去namesrv获取路由信息。selectOneMessageQueue方法通过轮询的方式，返回一个队列，以达到负载均衡的目的。

如果Producer发送消息失败，会自动重试，重试的策略：

1. 重试次数 < retryTimesWhenSendFailed（可配置）
2. 总的耗时（包含重试n次的耗时） < sendMsgTimeout（发送消息时传入的参数）
3. 同时满足上面两个条件后，Producer会选择另外一个队列发送消息

消息存储

RocketMQ的消息存储是由consume queue和commit log配合完成的。

Consume Queue

consume queue是消息的逻辑队列，相当于字典的目录，用来指定消息在物理文件commit log上的位置。

我们可以在配置中指定consume queue与commitlog存储的目录
每个topic下的每个queue都有一个对应的consumequeue文件，比如：

1

${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}

Consume Queue文件组织，如图所示：

1. 根据topic和queueId来组织文件，图中TopicA有两个队列0,1，那么TopicA和QueueId=0组成一个ConsumeQueue，TopicA和QueueId=1组成另一个ConsumeQueue。
2. 按照消费端的GroupName来分组重试队列，如果消费端消费失败，消息将被发往重试队列中，比如图中的%RETRY%ConsumerGroupA。
3. 按照消费端的GroupName来分组死信队列，如果消费端消费失败，并重试指定次数后，仍然失败，则发往死信队列，比如图中的%DLQ%ConsumerGroupA。

死信队列（Dead Letter Queue）一般用于存放由于某种原因无法传递的消息，比如处理失败或者已经过期的消息。

Consume Queue中存储单元是一个20字节定长的二进制数据，顺序写顺序读，如下图所示：

1. CommitLog Offset是指这条消息在Commit Log文件中的实际偏移量
2. Size存储中消息的大小
3. Message Tag HashCode存储消息的Tag的哈希值：主要用于订阅时消息过滤（订阅时如果指定了Tag，会根据HashCode来快速查找到订阅的消息）

Commit Log

CommitLog：消息存放的物理文件，每台broker上的commitlog被本机所有的queue共享，不做任何区分。
文件的默认位置如下，仍然可通过配置文件修改：

1

${user.home}\store\${commitlog}\${fileName}

CommitLog的消息存储单元长度不固定，文件顺序写，随机读。消息的存储结构如下表所示，按照编号顺序以及编号对应的内容依次存储。

消息存储实现

消息存储实现，比较复杂，也值得大家深入了解，这小节只以代码说明一下具体的流程。

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

// Set the storage time
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
synchronized (this) {
    long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
    // Here settings are stored timestamp, in order to ensure an orderly global
    msg.setStoreTimestamp(beginLockTimestamp);
    // MapedFile：操作物理文件在内存中的映射以及将内存数据持久化到物理文件中
    MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();
    // 将Message追加到文件commitlog
    result = mapedFile.appendMessage(msg, this.appendMessageCallback);
    switch (result.getStatus()) {
    case PUT_OK:break;
    case END_OF_FILE:
         // Create a new file, re-write the message
         mapedFile = this.mapedFileQueue.getLastMapedFile();
         result = mapedFile.appendMessage(msg, this.appendMessageCallback);
     break;
     DispatchRequest dispatchRequest = new DispatchRequest(
                topic,// 1
                queueId,// 2
                result.getWroteOffset(),// 3
                result.getWroteBytes(),// 4
                tagsCode,// 5
                msg.getStoreTimestamp(),// 6
                result.getLogicsOffset(),// 7
                msg.getKeys(),// 8
                /**
                 * Transaction
                 */
                msg.getSysFlag(),// 9
                msg.getPreparedTransactionOffset());// 10
    // 1.分发消息位置到ConsumeQueue
    // 2.分发到IndexService建立索引
    this.defaultMessageStore.putDispatchRequest(dispatchRequest);
}

消息的索引文件

如果一个消息包含key值的话，会使用IndexFile存储消息索引，文件的内容结构如图：

索引文件主要用于根据key来查询消息的，流程主要是：

1. 根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目，例如图中所示 slotNum=5000000)
2. 根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)
3. 遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)

消息订阅

RocketMQ消息订阅有两种模式，一种是Push模式，即MQServer主动向消费端推送；另外一种是Pull模式，即消费端在需要时，主动到MQServer拉取。但在具体实现时，Push和Pull模式都是采用消费端主动拉取的方式。

首先看下消费端的负载均衡：

消费端会通过RebalanceService线程，10秒钟做一次基于topic下的所有队列负载：

1. 遍历Consumer下的所有topic，然后根据topic订阅所有的消息
2. 获取同一topic和Consumer Group下的所有Consumer
3. 然后根据具体的分配策略来分配消费队列，分配的策略包含：平均分配、消费端配置等

如同上图所示：如果有 5 个队列，2 个 consumer，那么第一个 Consumer 消费 3 个队列，第二 consumer 消费 2 个队列。这里采用的就是平均分配策略，它类似于我们的分页，TOPIC下面的所有queue就是记录，Consumer的个数就相当于总的页数，那么每页有多少条记录，就类似于某个Consumer会消费哪些队列。

通过这样的策略来达到大体上的平均消费，这样的设计也可以很方面的水平扩展Consumer来提高消费能力。

消费端的Push模式是通过长轮询的模式来实现的，就如同下图：

Consumer端每隔一段时间主动向broker发送拉消息请求，broker在收到Pull请求后，如果有消息就立即返回数据，Consumer端收到返回的消息后，再回调消费者设置的Listener方法。如果broker在收到Pull请求时，消息队列里没有数据，broker端会阻塞请求直到有数据传递或超时才返回。

当然，Consumer端是通过一个线程将阻塞队列LinkedBlockingQueue<PullRequest>中的PullRequest发送到broker拉取消息，以防止Consumer一致被阻塞。而Broker端，在接收到Consumer的PullRequest时，如果发现没有消息，就会把PullRequest扔到ConcurrentHashMap中缓存起来。broker在启动时，会启动一个线程不停的从ConcurrentHashMap取出PullRequest检查，直到有数据返回。

未完待续…

[^6]: RPC(Remote Procedure Call)，远程过程调用。可以通俗的理解为不同系统间的接口调用。
[^7]: 此处的概念并不是指RocketMQ独有，部分名词属于消息领域模型中通用的概念。
[^8]: 回调函数多用于异步操作完成后的通知或处理，在这里用于确认消息是否发送成功。
[^9]: 几乎所有的MQ都做到了至少一次（At least once），三种消息交付方式之间的区别详见此处。

一天晚上，一位朋友发了个统计学领域一些定理及定义的截图给我，逐字逐句看了很久，完全没看懂在说啥，不由得想：为什么数学领域要把一些「能用通俗语言描述」的东西描述的晦涩难懂？

因为自己平时在给人解答一些技术问题时，都会尽量用通俗的话来描述，就算与实际含义有点偏差，也至少能让人家觉得：「噢，大概知道了」，而不是讲些大而美的名词在外行面前显得「专业」。

直到一次偶然机会，了解到了一个名词：「概念工程学」。

搭建一个概念体系和盖楼造汽车是一样的，这是一种工程，只要每个部件有统一标准，不管谁造出来都能拼成一个整体。数学工作者努力的把「能通俗」的事情翻译成严谨的数学语言，就是让这些概念标准化，标准化就有利于分工合作和深入研究。

首先能确定的一点是：任何领域的研究到达一定深度时就一定只是为少数人准备的。

客观世界的复杂性使我们需要更精确的工具去描述，才能更好发展我们的理论而不至于混乱、不自洽。对于类似数学领域这种严谨的「基础理论科学」来说更是如此。

总之，通俗的语言有助于科普，严谨的数学语言有助于深入，两者都很重要。

如果将交往看成一个找最优解问题，普通朋友能做到局部最优，但男女朋友可以做到全局最优，因为从局部最优搜索到全局最优需要不断考验对方的边界，而普通朋友看起来关系和谐，却是没法打破这一点的。

越陌生的朋友，越容易「在每一件事上都让彼此都舒服」。而恋人之间应该追求的是「总体上来看，我们在一起比不在一起，让彼此更舒服了」

比如你有个同学，这同学和你不算熟，你俩只有一个共同爱好，就是打dota。你们的生活除了游戏没有交集，只有晚上打dota的时候一起开黑，你们共度的每一段时光，都是彼此都开心的。你会发现，和这个同学一起玩的时候，「彼此都舒服」。至于他其他时候做什么，你一点都不关心。他爱干啥干啥，都不会令你感到不适。

男女朋友就是完全另一个系统了。

你认为，是否让彼此都舒服，是以「时间」来计算的。会有一个合理的「时间」，让男女双方都感到满意。这显然是一个过于简化的模型。

同样是两个小时，让女生陪男生看球，和让男生陪女生逛街，显然给双方带来的是截然不同的感受。因此时间并不是最本质的，做的事情才比较重要。

在成为恋人之后，你会发现，由于你们的生活被多多少少捆绑在一起，你们很多时候根本没有所谓「共赢」的选择。在某一件事情上，根本不可能让彼此「都舒服」。

就比如说看五月天演唱会。可能你女朋友是个五月天狂热粉，你对五月天毫无兴趣。这时候如果你陪你女朋友去看，你女朋友高兴，你不高兴。但是你不陪你女朋友去看，她觉得你作为男朋友竟然不喜欢她的爱豆，变成了你高兴她不高兴。也有可能，你不陪她，她就要和别的男生看，这时候你就更不高兴。或者说没有女生陪她看，你又担心她在这个比较复杂场合的安全，你又不高兴。再或者你说那咱们别去看五月天了，这时候变成了她不高兴。

你看，因为你们现在是男女朋友关系，而不是普通朋友了，就会出现很多别的想法，责任、义务、占有欲。这些想法就会让「彼此都舒服」成为一件不可能的事情。

而这些问题，根本不是所谓增加或者减少两个人「黏」在一起的时间可以解决的。这段时间在不在一起，都会引起其中一个人的不舒适。

男女朋友之间应该做的，是追求所谓「两个人加起来更开心」。

比如一件事，如果压根没这件事，你们俩心情不变，都是（0，0）。但是如果这件事女孩子做了，她很开心，你小不开心，变成了（10，-3）。这件事就是件可以做的事情。可这件事要是女孩子小开心，你大不开心，变成了（2，-20），那这件事就不应该做。

我之前说，喜欢一个人，是愿意和她做让彼此都开心的事情，但爱一个人，应该是愿意和她做让她开心可不让自己开心的事情。大概意思就是这样。

如果两个人都有这样的心态，那日积月累，你们每个人的开心值期望都应该是上升的，在一个地方让对方一下下，对方在未来就会用更好的方式弥补你。你陪我去看五月天，我就陪你去看邓紫棋。这就是所谓比较健康的恋情。

当然，并不是每一段恋情都是健康的。

如果有一个人的心态是「你应该事事站在我的立场考虑，只要我开心的事情就让我去做，我不开心的事情就不能做」，那另一个人的开心值很可能就会一直下降，直到他无法继续容忍。这个很好理解。

但是还有一种情况，即便有一个人的心态是「我应该事事站在你的立场考虑，只要你开心的事情我就愿意做，你不开心我就不做」，也会导致同样的结果。在模型上，他们是等价的。你什么时候看到过一段感情，只有一方玩命付出，努力牺牲，不断退让，最后还能修成正果的？

所以说，无论是男孩子还是女孩子，当你们感觉到另一半一直在让着自己，每件事都以自己开心为第一使命的时候，不要开心，这可能是分手的前奏。

然而，量化开心度并不是一件容易的事情。人们往往更容易感知到自己「是否开心」，而非「多开心」或者「多不开心」。更重要的是，人和人的价值观是有很大差异的。如果两个差别很大的人在一起，就会经常遇到这种取舍。动不动就是我很开心你很不开心。这样的感情就会导致两个人情绪波动太大，矛盾太多。而两个价值观很相似的人在一起，或许生活中大部分事情都是两个人都开心或者都不开心，他们几乎不太用做这种取舍。

理论固然有其可取之处，不过找到了对的人，这些就都不算是太大的问题了。

1、一个字符的String.length()是多少？

Java中字符采用Unicode（UTF16）编码，即16位（2字节）一个单元，UTF16可以包含一个单元或两个单元，对于U+0000-U+FFFF范围内的字符采用2字节编码，大于U+FFFF的采用四字节编码。例如一些emoji或者生僻字，采用四字节编码及两单元，String#length()将返回2，如需返回字符数可使用String#codePointCount(0,length)。

2、SpringSecurity使用了什么模式？

构建者模式、责任链模式。

3、SpringSecurity使用构建者模式最后构建的是一个什么？

是一个Filter、而且是FilterChainProxy，其内部包含多个SecurityFilterChain，根据请求匹配一个SecurityFilterChain并应用其中的Filter。

4、简单说一下Synchronized和AQS有什么不同？

• 前者是JVM层面实现、后者是语言层面实现。
• 前者根据不同锁实现方式不一样、后者是基于队列+CAS实现。
• 后者相比前者提供了更灵活的API、比如是否响应中断、超时时间设置、可绑定多个条件队列等等
• 前者只能是非公平锁、后者可实现公平锁，即先等待的线程先获取到锁。

5、偏向锁、轻量级锁、重量级锁的升级过程简单说一下？

在不存在锁竞争时是偏向锁、此时只使用了一次CAS设置对象头中的偏向线程ID；存在资源竞争时、在全局安全点撤销偏向锁升级为轻量级锁，轻量级锁使用CAS来修改对象头中持有锁的线程ID；在轻量级锁CAS竞争锁失败一定次数（默认10）后升级为重量级锁；重量级锁基于操作系统的Mutex Lock实现。

6、三种锁是怎样确保线程安全的？

使用Synchronized关键字时，会在字节码中插入monitorenter和monitorexit指令，当执行 monitorenter 指令时，线程试图获取锁也就是获取 对象监视器 monitor 的持有权，如果获取失败则根据锁类型自旋或者阻塞，从而确保线程安全。

7、AQS的原理简单说一下？内部使用了什么数据结构与设计模式？

AQS内部使用双向循环队列+CAS实现，使用了模版方法模式。

8、ThreadLocal存不存在内存泄漏问题?

ThreadLocalMap 中使用的 key 为 ThreadLocal 的弱引用,而 value 是强引用。所以，如果 ThreadLocal 没有被外部强引用的情况下，在垃圾回收的时候，key 会被清理掉，而 value 不会被清理掉。这样一来，ThreadLocalMap 中就会出现 key 为 null 的 Entry。假如我们不做任何措施的话，value 永远无法被 GC 回收，这个时候就可能会产生内存泄露。ThreadLocalMap 实现中已经考虑了这种情况，在调用 set()、get()、remove() 方法的时候，会清理掉 key 为 null 的记录。但在使用完 ThreadLocal方法后 最好手动调用remove()方法以便及时回收资源。

服务端

父EventLoopGroup中的Channel

Registered -> Bind -> Active -> [Read -> Read Complete] -> Close -> Inactive -> Unregistered

子EventLoopGroup中的Channel

Registered -> Active -> [(Read/Write -> Read Complete/Flush)、Event Triggered、Exception Caught] -> Inactive ->Unregistered

客户端

Registered -> Connect -> Active(Lazy Active) -> [(Read/Write -> Read Complete/Flush)、Event Triggered、Exception Caught、Disconnect] -> Close -> Inactive -> Unregistered

什么是AOP？

AOP（Aspect-oriented Programming），称为面向切面编程，作为面向对象的一种补充，用于处理系统中分布于各个模块的横切关注点，比如权限控制、事务管理、日志、缓存等等。

AOP代理主要分为静态代理和动态代理，静态代理的代表为AspectJ；而动态代理则以Spring AOP为代表。静态代理一般在编译期或者加载期实现，动态代理是运行期实现。

静态代理

AspectJ的静态代理，不是基于代理的。它使用修改字节码的技术将通知织入到目标类，所以，在运行时的类已经不是原本的那个类，是经过修改的了。通常来说，你写的AOP相关的代码，和对象的结合过程叫做织入(weave)。AspectJ的织入过程，有可能发生在三个阶段：

• 编译时织入(Compile-time weaving) 使用AspectJ的编译器ajc，在项目编译的阶段就将代码织入目标类
• 编译后织入(Post-compile weaving) 使用AspectJ的编译器ajc，向javac编译出来的的class织入代码
• 加载时织入(Load-time weaving) 使用特定的类加载器在类加载时期修改类字节码后加载进JVM

从上面可以看出，在程序的运行阶段，我们使用的类都是已经织入了通知代码的（并不会创建出多余的对象），而不像Spring AOP，需要在运行的时候去生成动态代理或者生成类。所以说，AspectJ的运行效率相比动态代理生成的类要高一点。

动态代理

动态代理则不会修改字节码，而是在内存中生成一个代理对象，这个AOP对象包含了目标对象的全部方法，并且在特定的切点做了增强处理，并回调原对象的方法。所以动态代理是在方法层级上进行代理，代理细度上不如静态代理。

生成AOP代理的入口

Spring所有的代理AopProxy的创建最后都是ProxyCreatorSupport#createAopProxy这个方法，而这个方法如下：

1
2
3
4
5
6

protected final synchronized AopProxy createAopProxy() {
if (!this.active) {
activate();
}
return getAopProxyFactory().createAopProxy(this);
}

显然它又是调用了AopProxyFactory#createAopProxy方法，它的唯一实现为DefaultAopProxyFactory。它做了一个简单的逻辑判断：若实现类接口，使用JdkDynamicAopProxy去创建，否则使用ObjenesisCglibAopProxy。

最终拿到AopProxy后，调用AopProxy#getProxy()就可以拿到代理对象，从而进行相应的工作了。

我们基本有一共共识就是：默认情况下，若我们实现了接口，就实用JDK动态代理，若没有就实用CGLIB。那么接下来，就来看看代理对象的创建、执行的具体过程是如何的。

AopProxy

它是一个AOP代理的抽象接口。提供了两个方法，让我们可以获取对应 配置的AOP对象的代理：

1
2
3
4

public interface AopProxy {
Object getProxy();
Object getProxy(@Nullable ClassLoader classLoader);
}

它的实现类只有三个：

• JdkDynamicAopProxy 基于jdk动态代理生成代理对象
• CglibAopProxy 基于cglib生成一个代理子类
• ObjenesisCglibAopProxy 基于Objenesis扩展CglibAopProxy，提供不调用类的构造函数来创建代理对象。

JdkDynamicAopProxy

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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

final class JdkDynamicAopProxy implements AopProxy, InvocationHandler, Serializable {

private static final long serialVersionUID = 5531744639992436476L;
private static final Log logger = LogFactory.getLog(JdkDynamicAopProxy.class);

// 这里就保存这个AOP代理所有的配置信息，包括所有的增强器等等
private final AdvisedSupport advised;
    // 用于记录代理接口中是否定义了equals和hashCode方法
private boolean equalsDefined;
private boolean hashCodeDefined;

public JdkDynamicAopProxy(AdvisedSupport config) throws AopConfigException {
Assert.notNull(config, "AdvisedSupport must not be null");
        // 至少需要一个增强器和目标实例才行
if (config.getAdvisors().length == 0 && config.getTargetSource() == AdvisedSupport.EMPTY_TARGET_SOURCE) {
throw new AopConfigException("No advisors and no TargetSource specified");
}
this.advised = config;
}

@Override
public Object getProxy() {
return getProxy(ClassUtils.getDefaultClassLoader());
}

    // 创建代理对象
@Override
public Object getProxy(@Nullable ClassLoader classLoader) {
if (logger.isTraceEnabled()) {
logger.trace("Creating JDK dynamic proxy: " + this.advised.getTargetSource());
}
        // 这部很重要，就是去找接口 我们看到最终代理的接口就是这里返回的所有接口们（除了我们自己的接口，还有Spring默认的一些接口）  大致过程如下：
//1、获取目标对象自己实现的接口(最终肯定都会被代理的)
//2、是否添加`SpringProxy`这个接口：目标对象实现对就不添加了，没实现过就添加
//3、是否新增`Adviced`接口，注意不是Advice通知接口。 实现过就不实现了，没实现过并且advised.isOpaque()为false就添加（默认会添加的）
//4、是否新增DecoratingProxy接口。传入的第二个参数decoratingProxy为true，并且没实现过就添加（显然这里，首次进来是会添加的）
//5、代理类的接口一共是目标对象的接口+上面三个接口SpringProxy、Advised、DecoratingProxy（SpringProxy是个标记接口而已，其余的接口都有对应的方法的）
//DecoratingProxy 这个接口Spring4.3后才提供
Class<?>[] proxiedInterfaces = AopProxyUtils.completeProxiedInterfaces(this.advised, true);
findDefinedEqualsAndHashCodeMethods(proxiedInterfaces);
        // 创建jdk动态代理对象，InvocationHandler传的this
return Proxy.newProxyInstance(classLoader, proxiedInterfaces, this);
}

// 找找看接口里有没有自己定义equals方法和hashCode方法，然后标记一下
// 注意此处用的是getDeclaredMethods，只会找当前类中的
private void findDefinedEqualsAndHashCodeMethods(Class<?>[] proxiedInterfaces) {
for (Class<?> proxiedInterface : proxiedInterfaces) {
Method[] methods = proxiedInterface.getDeclaredMethods();
for (Method method : methods) {
if (AopUtils.isEqualsMethod(method)) {
this.equalsDefined = true;
}
if (AopUtils.isHashCodeMethod(method)) {
this.hashCodeDefined = true;
}
if (this.equalsDefined && this.hashCodeDefined) {
return;
}
}
}
}

@Override
@Nullable
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Object oldProxy = null;
boolean setProxyContext = false;

TargetSource targetSource = this.advised.targetSource;
Object target = null;

try {
if (!this.equalsDefined && AopUtils.isEqualsMethod(method)) {
// The target does not implement the equals(Object) method itself.
return equals(args[0]);
}
else if (!this.hashCodeDefined && AopUtils.isHashCodeMethod(method)) {
// The target does not implement the hashCode() method itself.
return hashCode();
}
else if (method.getDeclaringClass() == DecoratingProxy.class) {
// There is only getDecoratedClass() declared -> dispatch to proxy config.
return AopProxyUtils.ultimateTargetClass(this.advised);
}
else if (!this.advised.opaque && method.getDeclaringClass().isInterface() &&
method.getDeclaringClass().isAssignableFrom(Advised.class)) {
// Service invocations on ProxyConfig with the proxy config...
return AopUtils.invokeJoinpointUsingReflection(this.advised, method, args);
}

Object retVal;

if (this.advised.exposeProxy) {
// Make invocation available if necessary.
oldProxy = AopContext.setCurrentProxy(proxy);
setProxyContext = true;
}

// Get as late as possible to minimize the time we "own" the target,
// in case it comes from a pool.
target = targetSource.getTarget();
Class<?> targetClass = (target != null ? target.getClass() : null);

// Get the interception chain for this method.
List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);

// Check whether we have any advice. If we don't, we can fallback on direct
// reflective invocation of the target, and avoid creating a MethodInvocation.
if (chain.isEmpty()) {
// We can skip creating a MethodInvocation: just invoke the target directly
// Note that the final invoker must be an InvokerInterceptor so we know it does
// nothing but a reflective operation on the target, and no hot swapping or fancy proxying.
Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);
retVal = AopUtils.invokeJoinpointUsingReflection(target, method, argsToUse);
}
else {
// We need to create a method invocation...
MethodInvocation invocation =
new ReflectiveMethodInvocation(proxy, target, method, args, targetClass, chain);
// Proceed to the joinpoint through the interceptor chain.
retVal = invocation.proceed();
}

// Massage return value if necessary.
Class<?> returnType = method.getReturnType();
if (retVal != null && retVal == target &&
returnType != Object.class && returnType.isInstance(proxy) &&
!RawTargetAccess.class.isAssignableFrom(method.getDeclaringClass())) {
// Special case: it returned "this" and the return type of the method
// is type-compatible. Note that we can't help if the target sets
// a reference to itself in another returned object.
retVal = proxy;
}
else if (retVal == null && returnType != Void.TYPE && returnType.isPrimitive()) {
throw new AopInvocationException(
"Null return value from advice does not match primitive return type for: " + method);
}
return retVal;
}
finally {
if (target != null && !targetSource.isStatic()) {
// Must have come from TargetSource.
targetSource.releaseTarget(target);
}
if (setProxyContext) {
// Restore old proxy.
AopContext.setCurrentProxy(oldProxy);
}
}
}

@Override
public boolean equals(@Nullable Object other) {
if (other == this) {
return true;
}
if (other == null) {
return false;
}

JdkDynamicAopProxy otherProxy;
if (other instanceof JdkDynamicAopProxy) {
otherProxy = (JdkDynamicAopProxy) other;
}
else if (Proxy.isProxyClass(other.getClass())) {
InvocationHandler ih = Proxy.getInvocationHandler(other);
if (!(ih instanceof JdkDynamicAopProxy)) {
return false;
}
otherProxy = (JdkDynamicAopProxy) ih;
}
else {
// Not a valid comparison...
return false;
}

// If we get here, otherProxy is the other AopProxy.
return AopProxyUtils.equalsInProxy(this.advised, otherProxy.advised);
}

@Override
public int hashCode() {
return JdkDynamicAopProxy.class.hashCode() * 13 + this.advised.getTargetSource().hashCode();
}

}

CglibAopProxy

1. 异步调用的方法中抛出异常
2. Around切面在调用ProceedingJoinPoint#proceed时catch并且处理了异常

前言

Redis5新增了一个Stream的数据类型，这个类型作为消息队列来使用时弥补了List和Pub/Sub的不足并且提供了更强大的功能，比如ack机制以及消费者组等概念，在有轻量消息队列使用需求时，使用这个新类型那是再好不过了。对于这个类型，在这里就不赘述了，想了解的话可以看一下这篇文章，在这里，我们就具体来讲一下在SpringBoot中的实践与踩坑。

注意，SpringBoot版本需要大于2.2（即spring-data-redis需要大于2.2）。

新API和类

在开始正文之前，我们先简单了解一下在2.2引入的和Stream操作相关的方法和类。

消息和消息ID的对象

消息（或者称为记录）和消息ID在Spring-Data-Redis中使用Record和RecordId来表示。

一个Record包含三部分内容：

• stream表示这个消息要发往那个Stream，也就是Stream的key
• recordId表示这个消息的ID，一般Redis服务器自动生成，也可以指定
• value表示消息内容

SpringBoot为我们提供了五种消息类型的抽象：MapRecord、ObjectRecord、ByteRecord、ByteBufferRecord、StringRecord，以及一个消息ID类型：RecordId。

这里另外说一下：其实除开ObjectRecord，其他几个Record都是通过继承MapRecord扩展而来的。StringRecord中的消息内容也并非仅仅是一个字符串，而是一个键值都为字符串类型的Map（ByteRecord、ByteBufferRecord同理）。而ObjectRecord最后也会使用HashMapper转换成MapRecord。为什么最后都是操作Map类型？这是因为Stream中的内容是以多个key-value这种键值对的形式存储的。

那么我们怎样去创建一个消息对象呢？

一般来说我们使用前两个消息类型比较多，所以Spring-Data-Redis很贴心的在Record这个顶级接口中提供了两个静态方法用于直接构造MapRecord和ObjectRecord：

1
2
3
4
5
6
7
8

static <S, K, V> MapRecord<S, K, V> of(Map<K, V> map)
    Assert.notNull(map, "Map must not be null!");
    return StreamRecords.mapBacked(map);
}
static <S, V> ObjectRecord<S, V> of(V value) {
    Assert.notNull(value, "Value must not be null!");
    return StreamRecords.objectBacked(value);
}

我们可以看到，这两个方法实际上是调用了StreamRecords中提供的静态方法来创建，StreamRecords这个类提供了下面这些方法用于创建五种Record：

1
2
3
4
5
6

ByteRecord rawBytes(Map<byte[], byte[]> raw) 
ByteBufferRecord rawBuffer(Map<ByteBuffer, ByteBuffer> raw) 
StringRecord string(Map<String, String> raw)
<S, K, V> MapRecord<S, K, V> mapBacked(Map<K, V> map)
<S, V> ObjectRecord<S, V> objectBacked(V value)
RecordBuilder<?> newRecord()  // 通过builder方式创建

当然，我们还可以通过使用某个具体的Record类型的create静态方法来创建，下面是几个示例：

1
2
3
4
5
6
7
8
9
10

String streamKey = "channel:stream:key1";//stream key
MailInfo mailInfo = new MailInfo("554205726@qq.com", "sendmail");//定义一个Object类型的消息内容
Map<String, String> map = new HashMap<String, String>() {{
    put("receiver", "534619360@qq.com");
}}; //定义一个Map类型的消息内容
Record.of(mailInfo).withStreamKey(streamKey);
Record.of(map).withStreamKey(streamKey).withId(RecordId.of("123"));//指定id
StreamRecords.objectBacked(mailInfo).withStreamKey(streamKey);
StreamRecords.mapBacked(map).withStreamKey(streamKey).withId(RecordId.autoGenerate());//指定id
ObjectRecord.create(streamKey, mailInfo); //使用ObjectRecord的create静态方法创建

如果我们不通过withId方法显示调用去指定id，那么默认的情况下就是使用RecordId.autoGenerate()自动生成。还有一个需要注意的地方就是在使用StreamRecords的方法来构建Record时一定要记住用withStreamKey方法来指定Stream Key。

不管是消息或是消息ID，这些类基本都提供了扁平化的api来构造，使用起来还是很简单的。那么在构造了一个Record后怎么将其持久化到Redis的Stream类型中呢？

向Stream添加消息（Record)

使用RedisTemplate操作Stream：

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

String streamKey = "channel:stream:key1";//stream key
MailInfo mailInfo = new MailInfo("554205726@qq.com", "sendmail");

ObjectRecord<String, MailInfo> record = ObjectRecord.create(streamKey, mailInfo);

RecordId id = record.getId(); //构造Record时使用的RecordId
RecordId recordId = redisTemplate.opsForStream().add(record); //返回的RecordId

id.getTimestamp(); //null
id.getSequence(); //null
id.shouldBeAutoGenerated(); //true

recordId.getTimestamp(); //not null
recordId.getSequence(); //not null
recordId.shouldBeAutoGenerated(); //false

使用add方法来添加记录，该方法执行成功后返回添加的记录的id信息。注意下面的结果，这里有两个问题需要注意：

1. 为什么我们构造Record时使用的RecordId和添加记录返回的RecordId不同？

   这个问题很好理解，因为构造Record时不指定id时虽然是自动生成，但是这个自动生成并不是在构造时就自动生成好了的，而是在执行Redis命令持久化时Redis服务器来自动生成的，所以前者在获取时间戳和序号的时返回null。

2. 为什么添加记录返回的RecordId调用shouldBeAutoGenerated方法返回false呢，不是自动生成了吗？

   其实也很好理解，因为在持久化一条Stream的记录时，我们可以指定id，也可以选择让Redis来自动生成，那么这也就导致add方法执行成功获取到Redis返回的id信息后在构造RecordId时并不知道返回的这个id是我们之前指定的还是Redis自动生成的，所以说前者返回true，后者返回false并不难理解。

说到这里，其实你去看一下Record构造时默认自动生成id是如何做到的就很好理解了。在这里稍微提一下：在构造Record时默认使用RecordId.autoGenerate()作为RecordId，而这个方法返回了一个匿名对象，这个匿名对象重写了上面那三个方法，前两个方法重写直接返回null，后者也就是shouldBeAutoGenerated方法返回true。

实现消息队列

在基本了解了SpringBoot2.2新增的几个Stream操作api和相关类之后，也就到了我们Stream实现消息队列的实践部分了。为了方便，下面我会以发送邮件为例来讲一下如何使用Strean来实现消息队列。

为了方便后续的讲解，先构造一个简单的邮件信息类作为我们的消息内容：

1
2
3
4
5
6
7
8
9

@Data
@AllArgsConstructor
@NoArgsConstructor
public class MailInfo {

    private String receiver;

    private String description;
}

构造StreamMessageListenerContainer

在使用Pub/Sub模式时，我们都是先创建一个RedisMessageListenerContainer容器，向这个容器注册监听器然后在onMessage方法中处理业务逻辑即可。那么使用Stream类型的话有没有提供一个类似的容器呢？答案是肯定的。在SpringBoot2.2提供了StreamMessageListenerContainer这个Stream类型专有的消息监听容器，而唯一的实现也就是DefaultStreamMessageListenerContainer。

StreamMessageListenerContainer的构造函数相比RedisMessageListenerContainer多了一个StreamMessageListenerContainerOptions，这个对象是使用builder方式来创建的：

1
2
3
4
5
6
7
8
9

StreamMessageListenerContainerOptions<String, ObjectRecord<String, MailInfo>> options =
        StreamMessageListenerContainerOptions.builder()
                .batchSize(100) //一批次拉取的最大count数
                .executor(executor)  //线程池
                .pollTimeout(Duration.ZERO) //阻塞式轮询
                .targetType(MailInfo.class) //目标类型（消息内容的类型）
                .build();
StreamMessageListenerContainer<String, ObjectRecord<String, MailInfo>> container =
    StreamMessageListenerContainer.create(redisConnectionFactory, options);

在构造StreamMessageListenerContainerOptions时最关键的就是targetType、objectMapper以及设置序列化器这几个方法，这些参数的设置会直接影响到后续接收到消息后能否反序列化为java对象！由于这部分内容涉及源码过多，在后面一部分我们再针对这几个方法进行详细的探查。

在构造完StreamMessageListenerContainer之后，现在该怎么注册消息监听器呢？我们接着往下看。

注册StreamListener

在Pub/Sub模式中我们使用addMessageListener(MessageListener, Topic)方法添加一个MessageListener到指定的Topic，那么在使用Stream消息的监听容器时，我们是使用receive方法。

Spring-data-redis提供了三个方法用于注册StreamListener：

1
2
3
4
5
6
7
8
9

default Subscription receive(StreamOffset<K> streamOffset, StreamListener<K, V> listener) {
    return register(StreamReadRequest.builder(streamOffset).build(), listener);
}
default Subscription receive(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K, V> listener) {
    return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(false).build(), listener);
}
default Subscription receiveAutoAck(Consumer consumer, StreamOffset<K> streamOffset, StreamListener<K, V> listener) {
    return register(StreamReadRequest.builder(streamOffset).consumer(consumer).autoAcknowledge(true).build(), listener);
}

如果你发现你哪里receive方法和我的不太一样？那么你使用的版本应该是2.2，这个版本的问题比较多，比如上面这个地方，使用第二个方法注册StreamListener，在消息被消费之后会自动ack，因为ConsumerStreamReadRequestBuilder的autoAck属性默认就是true（除非使用第一个方法指定StreamReadRequest），这个问题在2.3修复了，感兴趣可以去看看这部分源码，修补提交在这里。

我个人最初就是使用的SpringBoot2.2.2，使用过程中发现问题真的有点很多。比如还有一处序列化器泛型类型错误导致StreamMessageListenerContainerOptions构造混乱的问题【修补提交】，所以说如果你想尝鲜，那么强烈建议使用SpringBoot最新发布的版本。

Consumer和StreamOffset

可以看到receive方法另外还需要Consumer和StreamOffset两个参数，

Consumer

1

Consumer.from("group name", "consumer name")

Consumer表示消费者组中的某个消费者，这个东西只会在消费者组模式中用到。我们一般通过上面这种方式来创建，第一个参数表示消费者组，第二个参数表示消费者。

StreamOffset

StreamOffset用于表示在某个Stream上的偏移量，它包含两部分内容，一个是stream的key，另一个是ReadOffset用于表示读取偏移量。前者应该不需要过多的解释，那么ReadOffset这个读取偏移量是干嘛用的呢？

要搞清楚ReadOffset，我们首先要知道Stream中偏移量的含义，在Stream中偏移量既可以表示消费记录时的偏移量，又可以表示消费者组在Stream上的偏移量。还记得Redis中我们怎么读取Stream中的记录吗？

通过xread命令也就是非消费者组模式直接读取，或者使用xreadgroup命令在消费者组中命令一个消费者去消费一条记录，这个时候，我们可以通过0、>、$分别表示第一条记录、最后一次未被消费的记录和最新一条记录，这也就是ReadOffset的用途之一：用于表示直接读取或消费者组中消费者读取记录时的偏移量。

那么还有另外的用途吗？

当然了，还记得怎样创建消费者组吗？一般我们使用xgroup create命令创建一个消费者组时可以选择从Stream的第一条消息开始，或者Stream的中间某个记录开始，又或者从Stream的最新一条记录开始。也就分别代表了0、$。这也就是ReadOffset的用途之二：用于表示创建消费者组时该消费者组在Stream上的偏移量。

理解ReadOffset最快最简单的方法就是在Redis-cli中用Redis命令操作一番。这其中还有一些值得注意的问题，比如创建消费者组时不能使用>表示最后一次未被消费的记录；比如0表示从第一条开始并且包括第一条；$表示从最新一条开始但并不是指当前Stream的最后一条记录，所以使用$时最新一条也就是表示下一个xadd添加的那一条记录，所以说$在非消费者组模式的阻塞读取下才有意义！

实现StreamListener

同样的用Pub/Sub来类比，在Pub/Sub模式下我们实现的消息监听器是一般是MessageListener或者使用MessageListenerAdapter反射调用处理方法，在Strean消息队列的实现中必然也需要一个监听器用于处理真正的业务逻辑，这个类目前只有一个，也就是StreamListener：

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

public class StreamMessageListener implements StreamListener<String, ObjectRecord<String, MailInfo>> {
    
    private final Logger logger = LoggerFactory.getLogger(StreamMessageListener.class);
    private final StringRedisTemplate stringRedisTemplate;

    public StreamMessageListener(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public void onMessage(ObjectRecord<String, MailInfo> message) {
        RecordId id = message.getId();
        MailInfo messageValue = message.getValue();
        logger.info("消费stream:{}中的信息:{}, 消息id:{}", message.getStream(), messageValue, id);
        // 发邮件...
        stringRedisTemplate.opsForStream().acknowledge(MAIL_GROUP, message); //手动ack
    }
}

StreamListener和MessageListener差不多，只需要实现onMessage方法，只不过多了个泛型参数罢了。在实现消息监听器后也就可以使用receive方法进行注册了：

1
2
3

container.receive(Consumer.from(MAIL_GROUP, "consumer-1"),
        StreamOffset.create(MAIL_CHANNEL, ReadOffset.lastConsumed()),
        new StreamMessageListener(stringRedisTemplate));

注册完成之后启动StreamMessageListenerContainer容器：

1

container.start();

完整代码：

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

@Component
public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {

    public static final String MAIL_CHANNEL = "channel:stream:mail";
    public static final String MAIL_GROUP = "group:mail";

    private final ThreadPoolTaskExecutor executor;
    private final RedisConnectionFactory redisConnectionFactory;
    private final StringRedisTemplate stringRedisTemplate;

    private StreamMessageListenerContainer<String, ObjectRecord<String, MailInfo>> container;

    public StreamConsumerRunner(ThreadPoolTaskExecutor executor, RedisConnectionFactory redisConnectionFactory, StringRedisTemplate stringRedisTemplate) {
        this.executor = executor;
        this.redisConnectionFactory = redisConnectionFactory;
        this.stringRedisTemplate = stringRedisTemplate;
    }


    @Override
    public void run(ApplicationArguments args) {
        StreamMessageListenerContainerOptions<String, ObjectRecord<String, MailInfo>> options =
                StreamMessageListenerContainerOptions.builder()
                        .batchSize(10)
                        .executor(executor)
                        .pollTimeout(Duration.ZERO)
                        .targetType(MailInfo.class)
                        .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, MailInfo>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        prepareChannelAndGroup(stringRedisTemplate.opsForStream(), MAIL_CHANNEL, MAIL_GROUP);

        container.receive(Consumer.from(MAIL_GROUP, "consumer-1"),
                StreamOffset.create(MAIL_CHANNEL, ReadOffset.lastConsumed()),
                new StreamMessageListener(stringRedisTemplate));

        this.container = container;
        this.container.start();
    }

    private void prepareChannelAndGroup(StreamOperations<String, ?, ?> ops, String channel, String group) {
        String status = "OK";
        try {
            StreamInfo.XInfoGroups groups = ops.groups(channel);
            if (groups.stream().noneMatch(xInfoGroup -> group.equals(xInfoGroup.groupName()))) {
                status = ops.createGroup(channel, group);
            }
        } catch (Exception exception) {
            RecordId initialRecord = ops.add(ObjectRecord.create(channel, "Initial Record"));
            Assert.notNull(initialRecord, "Cannot initialize stream with key '" + channel + "'");
            status = ops.createGroup(channel, ReadOffset.from(initialRecord), group);
        } finally {
            Assert.isTrue("OK".equals(status), "Cannot create group with name '" + group + "'");
        }
    }

    @Override
    public void destroy() {
        this.container.stop();
    }

    public static class StreamMessageListener implements StreamListener<String, ObjectRecord<String, MailInfo>> {

        private final Logger logger = LoggerFactory.getLogger(StreamMessageListener.class);
        private final StringRedisTemplate stringRedisTemplate;

        public StreamMessageListener(StringRedisTemplate stringRedisTemplate) {
            this.stringRedisTemplate = stringRedisTemplate;
        }

        @Override
        public void onMessage(ObjectRecord<String, MailInfo> message) {
            RecordId id = message.getId();
            MailInfo messageValue = message.getValue();

            logger.info("消费stream:{}中的信息:{}, 消息id:{}", message.getStream(), messageValue, id);

            stringRedisTemplate.opsForStream().acknowledge(MAIL_GROUP, message);
        }
    }
}

注意prepareChannelAndGroup方法，在初始化容器时，如果key对应的stream或者group不存在时会抛出异常，所以我们需要提前检查并且初始化。

测试

添加一个测试接口：

1
2
3
4
5
6
7

@GetMapping("/sendMail")
public ResponseEntity<RecordId> sendMail(String receiver, String description) {
    MailInfo mailInfo = new MailInfo(receiver, description);
    ObjectRecord<String, MailInfo> record = Record.of(mailInfo).withStreamKey(channel);
    RecordId recordId = redisTemplate.opsForStream().add(record);
    return ResponseEntity.ok(recordId);
}

访问进行测试：

1

2020-06-28 19:26:17.870  INFO 21900 --- [         task-1] reamConsumerRunner$StreamMessageListener : 消费stream:channel:stream:mail中的信息:MailInfo(receiver=534619360@qq.com, description=发送邮件), 消息id:1593343576237-0

控制台输出日志，如果在redis-cli中使用xpending命令检查ack信息会发现也是0，因为我们虽然使用receive方法注册，但是在onMessage中手动提交了确认，当然，你也可以使用receiveAutoAck方法添加。

实践中踩到的坑

自动ack和泛型类型错误

这两个问题在前面已经提到了并且在2.3已经修复，这里不多说。还是那句话，如果想尝鲜，那么强烈推荐使用SpringBoot最新发布的版本。

RedisTemplate序列化器使用错误导致容器无法反序列化

RedisTemplate的hashvalue的序列化器最初使用的json序列化器，导致容器监听到新消息反序列化时抛出异常：

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

java.lang.IllegalArgumentException: Value must not be null!
    at org.springframework.util.Assert.notNull(Assert.java:198)
    at org.springframework.data.redis.connection.stream.Record.of(Record.java:81)
    at org.springframework.data.redis.connection.stream.MapRecord.toObjectRecord(MapRecord.java:147)
    at org.springframework.data.redis.core.StreamObjectMapper.toObjectRecord(StreamObjectMapper.java:132)
    at org.springframework.data.redis.core.StreamObjectMapper.map(StreamObjectMapper.java:158)
    at org.springframework.data.redis.core.StreamOperations.read(StreamOperations.java:458)
    at org.springframework.data.redis.stream.DefaultStreamMessageListenerContainer.lambda$getReadFunction$2(DefaultStreamMessageListenerContainer.java:232)
    at org.springframework.data.redis.stream.StreamPollTask.doLoop(StreamPollTask.java:138)
    at org.springframework.data.redis.stream.StreamPollTask.run(StreamPollTask.java:123)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

这是为什么呢？我们知道Redis中Stream中存的是键值对并且DefaultStreamOperations中操作的都是byte[]，也就是说我们虽然添加的是ObjectRecord，但是会先转换成MapRecord，然后再被转换成ByteRecord，最后进行序列化。来看一下add方法：

1
2
3
4
5
6

public RecordId add(Record<K, ?> record) {
    Assert.notNull(record, "Record must not be null");
    MapRecord<K, HK, HV> input = StreamObjectMapper.toMapRecord(this, record); //转换成MapRecord
    ByteRecord binaryRecord = input.serialize(keySerializer(), hashKeySerializer(), hashValueSerializer()); //再使用序列化器转换成ByteRecord
    return execute(connection -> connection.xAdd(binaryRecord), true);
}

通过上面这个方法我们可以发现stream序列化时和其他类型不一样，我们在使用json序列化一个对象时都是直接进行的，而这里分了两步并且序列化器是用于第二部转换，那么ObjectRecord是怎么转换成MapRecord的呢？点进StreamObjectMapper.toMapRecord方法可以看到其实是通过ObjectRecord#toMapRecord方法完成的，这个方法需要一个HashMapper用于将对象的属性/属性值映射构造成Map类型，你会发现opsForStream方法重载了一个默认无参的方法，而这个方法默认使用的是ObjectHashMapper，在我们构造StreamMessageListenerContainerOptionsBuilder时调用targetType时默认使用的也是ObjectHashMapper。而这个ObjectHashMapper会将对象中的属性和属性值转换成byte[]形式，所以在第一步之后这个MapRecord中的值的类型已经是byte[]了，那么也就导致第二步在使用json序列化器转换为ByteRecord时出现这种情况：objectMapper.writeValueAsBytes(byte[])，这是一个测试实例：

1
2
3
4
5
6
7
8

@Test
void test() throws JsonProcessingException {
    ObjectMapper mapper = new ObjectMapper();
    byte[] value = "534619360@qq.com".getBytes();
    byte[] bytes = mapper.writeValueAsBytes(value);
    System.out.println(new String(bytes));
    //输出"NTM0NjE5MzYwQHFxLmNvbQ=="
}

反应到stream中值就变成了\“NTM0NjE5MzYwQHFxLmNvbQ==\“（引号需要转义）。为了便于理解，我们可以使用设置使用json序列化器的RedisTemplate进行add断点debug测试看一下转换后的两个Record中的内容：

1
2
3
4
5
6

@Test
void test() {
    stringRedisTemplate.setHashValueSerializer(RedisSerializer.json());
    MailInfo mailInfo = new MailInfo("534619360@qq.com", "send mail");
    stringRedisTemplate.opsForStream().add(Record.of(mailInfo).withStreamKey(channel));
}

运行测试并断点查看MapRecord和ByteRecord：

使用Redis Desktop Manager查看值：

测试结束终端抛出上面提到的异常。这个问题解决办法就是使用String序列化器也就是使用StringRedisTemplate，因为这个序列化器不能序列化byte[]类型的对象，使用这个序列化器在序列化时如果已经是byte[]，那么就会直接返回原byte[]：

更具体的细节可以跟着add方法debug一遍。

ReadOffset使用错误导致group中消费者消费失败

异常：

1

RedisCommandExecutionException: ERR The $ ID is meaningless in the context of XREADGROUP: you want to read the history of this consumer by specifying a proper ID, or use the > ID to get new messages. The $ ID would just return an empty result set.

上面StreamOffset中也提到了，这涉及到0、>、$的使用场景和范围，如果出现这个异常，很有可能你在消费者组模式下设置消费者读取的offset时使用了ReadOffset.latest()，而这个对应着$，也就是最新一条记录。如果不明白那么你可能对这三个标识符的使用还不是很理解，最好的解决办法就是使用redis命令先完整的操作一遍。

stream或者group不存在导致启动抛出异常

同样在上面提到了，在构造StreamMessageListenerContainer时需要stream和group存在才可以。解决方法就是提前检查并初始化，上面已给出代码。

总结

其实在实践之初，我在网上也搜了很多相关的资料，但是发现这些资料基本都是使用redis-cli进行命令上的操作，并没有SpringBoot中实现。这次实践可谓是艰辛，由于目前该支持的迭代次数比较少，不乏一些bug或者小问题（2.3已经比较稳定），并且只有lettuce提供了stream类型的操作实现，而lettuce本身又有些小毛病，这些因素结合在一起也就导致这个过程花费了我整整两天时间，而这篇博客又花了整整一下午的时间才算完成，其中可能有些内容因为涉及东西比较多只能粗略提一下，并且语言组织上不太好可能不好去理解。话说回来，这篇文章也算是我自己实践后的一个个人总结吧，这个过程其实学到的东西还是很多的，也不枉费花了这么多时间。如果你发现文章有什么地方有问题或者有什么地方不理解，也欢迎在评论区留言一起交流~

写在前面

还是NLog问题，虽然NLog上手简单，配置容易，但是在NetCore3下还是有一些坑的，虽然这不是NLog的问题。因为这个问题百度一圈都无果，所以在这里记录一下，希望可以帮助到百度的同学（虽然这个站点没有提交到百度收录？逃~

问题由来

好，那么现在回到主题。前一阵子在用WPF来做一个First、Follow集算法模拟的工具，项目中使用NLog记录日志，在NetCore 3.1下作为单文件发布。发布后File类型的日志target没有将日志如期写入到文件中。并且仅在单文件发布并且fileName使用相对路径或者${basedir}指定基础目录时，才会出现这个问题，其实这个问题的解决方法很简单，只需稍微改一下就可以解决。

解决

使用${basedir:fixtempdir=true}指定基础目录。

这个问题主要原因其实是NetCore3单文件发布时AppDomain.BaseDirectory无法正确引用基础路径的问题，并且微软并没有打算在NetCore3修复这个问题，或许NetCore5会修复这个问题吧，文末会放几个链接，感兴趣可以去看看。

写在最后

又水了一篇文章，最近文章质量有点差，上个月还鸽掉了一篇Aop详解的文章（其实一直都在draft中），感觉有点罪恶感 :D

参考：

• PublishSingleFile excluding appsettings not working as expected
• Single file publish: AppContext.BaseDirectory doesn’t point to apphost directory
• When .net core app published as single file - ${basedir} is wrong
• Why cannot write log to files when WPF application publishing as a single file
• Basedir layout renderer

问题由来

在一次将Asp.net Core默认日志换成NLog时，发现NLog配置文件中的设置不生效？具体的来说就是在NLog文件中设置的路由以及对应的日志级别只有在Info或者以上时才生效，而Debug、Trace级别则不会有日志输出。比如我的NLog配置：

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

<?xml version="1.0" encoding="utf-8" ?>
<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://www.nlog-project.org/schemas/NLog.xsd NLog.xsd"
      autoReload="true" throwExceptions="false" throwConfigExceptions="true"
      internalLogLevel="Warn" internalLogFile="${basedir}/logs/internal.log">

  <variable name="logDirectory" value="${basedir}/logs"/>

  <extensions>
    <add assembly="NLog.Web.AspNetCore"/>
  </extensions>

  <targets>
    <default-wrapper xsi:type="BufferingWrapper" bufferSize="100"/>
    <target xsi:type="File" name="file" fileName="${logDirectory}/${shortdate}-${level}.log" encoding="utf-8"
            layout="${longdate}|${uppercase:${level}}|${logger}|${message} ${exception:format=tostring}" />
  </targets>
  <targets>
    <default-wrapper xsi:type="AsyncWrapper">
      <wrapper-target xsi:type="RetryingWrapper"/>
    </default-wrapper>
    <target xsi:type="ColoredConsole" name="console" detectConsoleAvailable="true"
            layout="${longdate}|${uppercase:${level}}|${logger}|${message}"/>
  </targets>

  <rules>
    <logger name="Microsoft.*" minlevel="Debug" writeTo="console"/>
    <logger name="WebApplication.*" minlevel="Trace" writeTo="console"/>
    <logger name="*" minlevel="Info" writeTo="file" final="true"/>
  </rules>

</nlog>

按照rules中的设置，Microsoft命名空间下Debug级别的日志应该都可以输出在控制台中的，但是启动后控制台的输出只有：

1
2
3
4
5

2020-06-16 20:41:34.9142|INFO|Microsoft.Hosting.Lifetime|Now listening on: https://localhost:5001
2020-06-16 20:41:34.9353|INFO|Microsoft.Hosting.Lifetime|Now listening on: http://localhost:5000
2020-06-16 20:41:34.9353|INFO|Microsoft.Hosting.Lifetime|Application started. Press Ctrl+C to shut down.
2020-06-16 20:41:34.9353|INFO|Microsoft.Hosting.Lifetime|Hosting environment: Development
2020-06-16 20:41:34.9353|INFO|Microsoft.Hosting.Lifetime|Content root path: D:\workspace\csharp\WebApplication\WebApplication

注意这里的格式已经应用了NLog中设置的布局，这说明配置文件被加载了并没有问题，但为什么日志级别的设置不生效呢？

原因所在

其实是因为appsettings.{env}.json中日志级别的设置覆盖了Nlog中的设置。在我们创建一个Asp.Net Core项目时，一般会帮我们创建好两个appsettings.json文件：

1
2
3
4
5
6
7
8
9
10

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  },
  "AllowedHosts": "*"
}

1
2
3
4
5
6
7
8
9

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft": "Warning",
      "Microsoft.Hosting.Lifetime": "Information"
    }
  }
}

注意这里使用.net core3.1创建的模板，2.1创建的项目命名空间以及对应日志级别会有所不同。

从这两个文件很容易发现默认的日志级别为Information并且Microsoft命名空间下的日志级别被设置成了Warning，当然也就覆盖了NLog中的设置，为什么会覆盖呢？可以思考一下。并且这里的设置能够影响到NLog中的设置还有一个前提：将NLog添加到容器，并且是以DI的方式来获取ILogger：

1
2
3
4
5
6
7
8
9
10

private static IHostBuilder CreateHostBuilder(string[] args)
{
    return Host.CreateDefaultBuilder(args)
        .ConfigureLogging(logBuilder =>
        {
            logBuilder.ClearProviders()
            .AddNLog("NLog.config");
        })
        .ConfigureWebHostDefaults(webBuilder => webBuilder.UseStartup<Startup>());
}

如果你是这样添加NLog，然后使用DI在需要记录日志的类中注入ILogger的话，那很可能就中招了。

1

private static readonly NLog.Logger _logger = NLog.LogManager.GetCurrentClassLogger();

如果你不使用DI来注入而是使用上面这个方式来获取logger，那么就不存在这个问题。但是我们使用容器主要的目的不就是容器可以更方便的管理对象之间的依赖吗？更具体地说那就是DI。那这个问题该怎样解决呢？

解决方法

解决的方法其实很简单：

首先删除appsettings.{env}.json中日志相关的设置，然后在注册NLog时设置最低日志级别为你所用到的最低级别（因为就算去掉了这些设置，默认的日志级别也是Information），这样一来，就相当于把日志级别控制和输出都交给NLog来管理。比如我上面的文件中给自己项目的命名空间设置了Trace并且已是最低级别，那么在删除appsettings.{env}.json中日志相关的设置后可以这样来添加NLog：

1
2
3
4
5
6
7
8
9
10
11

private static IHostBuilder CreateHostBuilder(string[] args)
{
    return Host.CreateDefaultBuilder(args)
        .ConfigureLogging(logBuilder =>
        {
            logBuilder.ClearProviders()
            .SetMinimumLevel(LogLevel.Trace) // 最低为Trace
            .AddNLog(_configFileRelativePath);
        })
        .ConfigureWebHostDefaults(webBuilder => webBuilder.UseStartup<Startup>());
}

或者你也可以使用AddFilter来细分使得日志输出可以“交到”NLog手中。当然，如果你不想删除原有的设置并且不想通过SetMinimumLevel(LogLevel)方法来设置最低级别，你也可以在appsettings.{env}.json文件中设置命名空间和相应的日志级别来保证NLog中的设置不会被覆盖（因为这个文件在启动时会自动被加载）。总之，只需要保证NLog中日志级别的设置不会因为该文件或者默认的Information级别所覆盖即可。如果不使用DI来获取的话那就当我没说。（逃

在一次部署Asp.Net Core应用时遇到了这么几个问题：

1. 启动应用后，从IConfiguration中获取不到连接字符串。
2. 使用nginx反代后，Identity框架页面跳转后域名被改写成了localhost或者主机名。

问题一：

应用使用sqlite数据库，在程序启动后创建并初始化数据库，本地开发启动并没有报错，但是部署到服务器却获取不到连接字符串，报错：

百度和谷歌都无果，最初我还以为是程序中某个地方可能出错了，但是在WSL上经过多次测试，发现如果没有在执行文件的目录下启动了项目才会报这样的错误，这个问题很奇怪，猜测是Directory.GetCurrentDirectory()获取路径导致，并未求证。知道了问题所在那么解决的办法就很简单了，cd到执行文件目录启动即可。

问题二：

这个问题是宝塔面板导致的，在设置反代的发送域名后，并没有反应到配置文件中，也就是说在面板上设置proxy_set_header Host $host;后文件中并没有改过来，依旧是localhost或者$hostname，也就导致了页面跳转时域名变成了localhost或者主机名。解决的办法很简单，只需要手动修改nginx配置文件中的反代发送域名即可。

在这之前，其实还遇到了一个和nginx反代相关的问题：如果使用nginx来代理目录，从而实现一个域名下部署多个项目，这样做之后就导致Asp.net Core应用全部404，网上搜索一番，有些类似的问题解答提到使用Microsoft.AspNetCore.HttpOverrides包，但是我照着操作一番后依旧没用，目前比较笨的解决办法就是mvc的路由前手动加前缀，更好的办法目前没有找到，如果你知道，还请评论告知~

JetBrains产品对于WSL提供了一定的支持，但是其只支持Microsoft Store中发行的WSL，对于类似ArchWSL这种非商店发行版，在配置工具链时却不能够被发现。下面给出两个方法来解决这个问题。

由于我目前只使用过ArchWSL这个非商店发行版的WSL，所以下面以ArchWSL为例，对于其他版本的WSL，理论上也行得通。

1. 第一种方法很简单，只需要复制执行文件到指定目录

   只需将Arch.exe文件拷贝一份到C:\Users\lolico\AppData\Local\Microsoft\WindowsApps目录下即可。

   注意，如果你安装了多个Arch，在拷贝时需要将每个版本对应的执行文件都拷贝到WindowsApps目录下。

2. 第二种方法是网上搜索到的一个方法，并且在RubyMine官方文档中提到了这个方法

   以RubyMine为例，修改wsl.distributions.xml文件：

   1 2 3 4 5 6

   <descriptor> <id>Arch</id> <microsoft-id>Arch</microsoft-id> <executable-path>Arch.exe</executable-path> <presentable-name>Arch Linux</presentable-name> </descriptor>

   • executable-path使用执行文件的绝对路径。
   • wsl.distributions.xml文件位于产品对应的配置文件夹下，对于不同版本，配置文件夹位置可能不一样。

   旧版本软件的配置文件夹在%HOMEPATH%下，此时该文件位于配置文件夹下的config\options目录中。在2020.1版本，配置文件夹迁移到了%HOMEPATH%\AppData\Roaming\JetBrains下，此时该文件位于配置文件夹下的options目录中。由于我一直使用ToolBox来安装并管理这些软件，所以说如果你并不是从ToolBox中安装的软件，新版本的配置文件夹可能仍然在%HOMEPATH%下。

最后，还想说一句，在使用商店发行版时我们可以用类似\\wsl$\Ubuntu这种路径来访问WSL的根目录，在使用非商店发行版时却不行，解决的办法还是上面提到第一个方法，拷贝文件到WindowsApps之后即可使用\\wsl$\Arch来访问Arch的根目录。

参考：

^ Custom WSL distributions
^ Support custom WSL distributive (not from Microsoft Store)

在一次偶然情况下发现SpringBoot开发的应用在没有使用@EnableAspectJAutoProxy注解的情况下，AOP还是可以正常工作。不经引起了我的注意，随后便猜测SpringBoot中是否存在一个自动配置类在不使用注解的情况下完成自动配置并开启了AOP。

在Idea中尝试搜索AopAutoConfiguration，果然不出所料，有这么一个自动配置类：

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

@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(prefix = "spring.aop", name = "auto", havingValue = "true", matchIfMissing = true)
public class AopAutoConfiguration {

    @Configuration(proxyBeanMethods = false)
    @ConditionalOnClass(Advice.class)
    static class AspectJAutoProxyingConfiguration {

        @Configuration(proxyBeanMethods = false)
        @EnableAspectJAutoProxy(proxyTargetClass = false)
        @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false",
                matchIfMissing = false)
        static class jdkDynamicAutoProxyConfiguration {

        }

        @Configuration(proxyBeanMethods = false)
        @EnableAspectJAutoProxy(proxyTargetClass = true)
        @ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true",
                matchIfMissing = true)
        static class CglibAutoProxyConfiguration {

        }

    }
    // ...
}

可以看到@ConditionalOnProperty注解中matchIfMissing属性的设置，也正是这个属性的设置使得我们在不使用@EnableAspectJAutoProxy注解开启AOP的情况下，AOP也可以正常工作。

从这个自动配置类中还能够看出AOP默认使用Cglib代理。那么如果我们使用@EnableAspectJAutoProxy注解去指定默认不使用Cglib代理，有没有用呢？

经过测试发现，使用@EnableAspectJAutoProxy注解显示指定是没有用的。那么该如何设置默认使用jdk动态代理呢？根据自动配置类能知道，我们只需要在application.properties中指定spring.aop.proxy-target-class=false或者设置spring.aop.auto=false关闭aop自动配置后自行使用@EnableAspectJAutoProxy注解进行设置。

并且如果查看SpringBoot1.5.x中的这个自动配置类会发现默认是使用jdk动态代理，那么为什么在SpringBoot2.x中改为默认使用Cglib了呢？

使用jdk动态代理会存在一个代理问题：目标对象一定要实现接口。因为jdk动态代理是基于接口的，并且意味着当使用@Autowired注入一个使用jdk代理生成的对象时必须使用接口。

就像这样：

1
2

@Autowired
UserService userService;

一旦使用实现类的方式进行注入就会失败：

1
2

@Autowired
UserServiceImpl userService;

使用Cglib代理就不存在这个问题，原因就在于Cglib是使用子类的方式进行代理。

当查看类似@EnableCaching、@EnableAsync、@EnableTransactionManagement注解中proxyTargetClass属性的注释，我们还会发现注释中说到，这个属性设置为true后会影响Spring管理的所有Bean使用的代理方式。但是我经过测试发现其他注解的这个属性设置为true时并不会影响@EnableAsync的代理方式。但是@EnableAsync的这个属性设置为true后会影响其他注解下对象的代理方式。(SpringBoot2.2.2下测试)

注意：proxyTargetClass属性默认为false时并不意味着如果目标对象没有实现接口，生成代理类就会失败，这种情况下Spring会使用Cglib代理。

参考：

^ Use @EnableTransactionManagement(proxyTargetClass = true)
^ @EnableTransactionManagement proxyTargetClass not control by spring.aop.proxyTargetClass

写在前面

在SpringSecurity中配置跨域，我相信所有用过SpringSecurity的人应该都知道，因为实在是太简单了。那我为什么还要写这篇文章呢？写这篇文章的目的当然不是去解释如何配置跨域，而是通过分析Spring对跨域支持的源码来感受设计中的优雅。

先声明一下开发环境：SpringBoot：2.2.2

正文

既然说到SpringSecurity配置跨域，那么我们就先简单复习一下如何配置跨域。

配置跨域

我们都知道集成SpringSecurity后配置跨域我们只需要在继承WebSecurityConfigurerAdapter类，重写configure(HttpSecurity http)方法，开启cors并提供一个跨域配置源即可。下面是一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@Override
protected void configure(HttpSecurity http) throws Exception {
    http.sessionManagement()
            .sessionCreationPolicy(SessionCreationPolicy.STATELESS) // 前后端分离
            .and()
            .csrf().disable() // 禁用csrf
            .cors(); // 跨域
    // ... 省略其他配置
}
// 提供一个CorsConfigurationSource
// 这里直接注册成Bean即可，注意方法名必须是corsConfigurationSource，后面会解释
// 也可以cors().configurationSource(corsConfigurationSource())指定
@Bean
public CorsConfigurationSource corsConfigurationSource() {
    CorsConfiguration configuration = new CorsConfiguration();
    configuration.addAllowedOrigin("*"); // 根据实际的需要去设置
    configuration.addAllowedMethod("*"); // 同上
    configuration.addAllowedHeader("*"); 
    configuration.setMaxAge(3600L);
    configuration.setAllowCredentials(true);
    UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
    source.registerCorsConfiguration("/**", configuration);
    return source;
}

就如前面所说，只需要开启cors再提供一个跨域配置源即可。方法很简单，但这里有个坑需要注意一下。

暴露公共接口时跨域的一个坑

如果我们还重写了configure(WebSecurity web)方法，使用web.ignoring().antMatchers(ignorePaths)去暴露一个公共接口’/pub’那么上面的跨域配置对这个接口来说就没用，也就是说这个接口会出现跨域问题。然而我们原本就是为了提供公共接口’/pub’，但现在却有跨域问题，那怎么能行！！！（一般来说这个方法是对静态资源设置直接放行，而不是公共接口！）

那这到底是为什么呢？

因为SpringSecurity配置跨域支持，是通过CorsFilter过滤器来实现的，我们web.ignoring()中设置后对应的接口请求就不会经过CorsFilter来处理，这个接口当然就存在跨域问题了！之所以说这个方法是对静态资源设置直接放行，而不是公共接口也是这个原因，那正确的方法是什么呢？还是configure(HttpSecurity http)方法：

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

@Override
protected void configure(HttpSecurity http) throws Exception {
    http.sessionManagement()
            .sessionCreationPolicy(SessionCreationPolicy.STATELESS) // 前后端分离
            .and()
            .csrf().disable() // 禁用csrf
            .cors() // 跨域
            .and()
            .authorizeRequests()
            .antMatchers("/pub/**").permitAll() // 匿名通过认证
            .anyRequest().authenticated() //剩下的任何请求都需要认证
}

cors方法

现在我们来看一下cors()方法，点进这个方法看看，其实很简单，就是应用了一个CorsConfigurer配置类。如果看过SpringSecurity自动配置，对形如xxxConfigurer的类名应该不陌生。
这个Configurer其实就是在”FilterChain”上添加了一个过滤器，即CorsFilter

我们都知道CorsFilter的构造方法需要一个CorsConfigurationSource，在请求到来时，使用CorsProcessor根据提供的CorsConfiguration去对请求进行处理（在CorsFilter中默认是DefaultCorsProcessor）而CorsConfiguration是通过CorsConfigurationSource#getCorsConfiguration方法获得的，所以说怎么获得CorsConfigurationSource至关重要。

还记得上面在配置CorsConfigurationSource时，我们直接注册Bean而不是通过configurationSource()方法指定吗？这种方法为什么是可行的呢？来看一下CorsConfigurer是如何获得CorsConfigurationSource并构造CorsFilter的：

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

private CorsFilter getCorsFilter(ApplicationContext context) {
    //如果指定了CorsConfigurationSource，那么用指定的
    if (this.configurationSource != null) {
        return new CorsFilter(this.configurationSource);
    }
    boolean containsCorsFilter = context.containsBeanDefinition(CORS_FILTER_BEAN_NAME);
    //如果容器中已经有名字是’corsFilter‘的bean，则用已经有的
    if (containsCorsFilter) { 
        return context.getBean(CORS_FILTER_BEAN_NAME, CorsFilter.class);
    }
    boolean containsCorsSource = context.containsBean(CORS_CONFIGURATION_SOURCE_BEAN_NAME);
    //如果既没有指定，容器中也不存在名字是’corsFilter‘的CorsFilter
    //那么看一下容器中有没有名字是’corsConfigurationSource‘的CorsConfigurationSource
    //如果有，取出来作为CorsConfigurationSource
    if (containsCorsSource) {
        CorsConfigurationSource configurationSource = context.getBean(CORS_CONFIGURATION_SOURCE_BEAN_NAME, CorsConfigurationSource.class);
        return new CorsFilter(configurationSource);
    }
    //如果也没有corsConfigurationSource，看看类路径下存不存在HandlerMappingIntrospector这个类
    boolean mvcPresent = ClassUtils.isPresent(HANDLER_MAPPING_INTROSPECTOR,context.getClassLoader());
    if (mvcPresent) { //如果存在
        return MvcCorsFilter.getMvcCorsFilter(context);
    }
    return null;
}
static class MvcCorsFilter { //内部类
    private static final String HANDLER_MAPPING_INTROSPECTOR_BEAN_NAME = "mvcHandlerMappingIntrospector";
    private static CorsFilter getMvcCorsFilter(ApplicationContext context) {
        if (!context.containsBean(HANDLER_MAPPING_INTROSPECTOR_BEAN_NAME)) {
            throw new NoSuchBeanDefinitionException(HANDLER_MAPPING_INTROSPECTOR_BEAN_NAME, "A Bean named " + 
            HANDLER_MAPPING_INTROSPECTOR_BEAN_NAME +" of type " + HandlerMappingIntrospector.class.getName()
            + " is required to use MvcRequestMatcher. Please ensure Spring Security & Spring MVC are configured in a shared ApplicationContext.");
        }
        // 从容器中取出HandlerMappingIntrospector作为CorsConfigurationSource
        HandlerMappingIntrospector mappingIntrospector = context.getBean(HANDLER_MAPPING_INTROSPECTOR_BEAN_NAME, HandlerMappingIntrospector.class);
        return new CorsFilter(mappingIntrospector);
    }   
}

获取CorsConfigurationSource并构造CorsFilter的步骤注释里写的很清楚了，正常来说我们配置跨域配置源不管是直接指定也好，还是注册成Bean也好（注意Bean名字的要求），都是可以被获取到的。一般情况下，我们也的确是这样做的（直接提供一个CorsConfigurationSource）。但为什么最后有MvcCorsFilter.getMvcCorsFilter(context)这样一个调用？通过这个方法里抛出的异常信息不难猜测到是SpringSecurity为了兼容SpringMVC中配置跨域的方式。

还记得不使用SpringSecurity时如何在SpringMVC中配置支持跨域吗？

两种方式：

• 将@CrossOrigin注解标注在支持跨域的接口上
• 重写WebMvcConfigurer#addCorsMappings方法进行全局配置

看到这里你可能会猜测：是不是HandlerMappingIntrospector实现了CorsConfigurationSource，并且是根据上面两种方式的配置来返回跨域配置的呢？

事实上，的确是这样的。为了便于理解后面给出的代码，先来看看CorsFilter类和CorsConfigurationSource接口：

CorsConfigurationSource

1
2
3
4

public interface CorsConfigurationSource {
    @Nullable
    CorsConfiguration getCorsConfiguration(HttpServletRequest request);
}

跨域配置源，实现类要实现getCorsConfiguration方法返回一个CorsConfiguration跨域配置，其中包含允许那些域、请求方法、请求头，是否允许携带凭证，缓存时间是多久，允许携带的头属性等信息。

CorsConfigurationSource有五个实现类：

• CorsInterceptor
• HandlerMappingIntrospector
• PreFlightHandler
• ResourceHttpRequestHandler
• UrlBasedCorsConfigurationSource

CorsFilter

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

public class CorsFilter extends OncePerRequestFilter {
    private final CorsConfigurationSource configSource;
    private CorsProcessor processor = new DefaultCorsProcessor();

    public CorsFilter(CorsConfigurationSource configSource) {
        Assert.notNull(configSource, "CorsConfigurationSource must not be null");
        this.configSource = configSource;
    }

    public void setCorsProcessor(CorsProcessor processor) {
        Assert.notNull(processor, "CorsProcessor must not be null");
        this.processor = processor;
    }

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response,
            FilterChain filterChain) throws ServletException, IOException {
        // 获取CorsConfiguration
        CorsConfiguration corsConfiguration = this.configSource.getCorsConfiguration(request);
        // 根据CorsConfiguration处理请求
        boolean isValid = this.processor.processRequest(corsConfiguration, request, response);
        if (!isValid || CorsUtils.isPreFlightRequest(request)) {
            return;
        }
        filterChain.doFilter(request, response);
    }
}

DefaultCorsProcessor#processRequest中根据请求是否跨域，是否是预检请求以及CorsConfiguration等信息来对请求进行处理和在响应头中写入一些信息。具体的源码就不分析了，还是比较好理解的。前提是需要对CORS有一定的了解，可以看下HTTP访问控制（CORS）这篇文章。

HandlerMappingIntrospector

HandlerMappingIntrospector比较特别，不要认为这是个拦截器，”Introspector”翻译成中文是内省者的意思。

这个类在初始化后会调用afterPropertiesSet方法，将容器中所有的HandlerMapping添加到该类的handlerMappings这个List中。

来看一下官方对于这个类的解释：

1
2
3
4

Helper class to get information from the HandlerMapping that would serve a specific request.
Provides the following methods:
    - getMatchableHandlerMapping(javax.servlet.http.HttpServletRequest) — obtain a HandlerMapping to check request-matching criteria against.
    - getCorsConfiguration(javax.servlet.http.HttpServletRequest) — obtain the CORS configuration for the request.

这个类是一个帮助类，用于从HandlerMapping中获取请求的特定信息，提供了两个方法。第一个方法用于获取一个MatchableHandlerMapping来检查请求匹配条件，第二个方法用于获取适用于这个请求的CorsConfiguration跨域配置。

我们重点关注第二个方法：

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

public CorsConfiguration getCorsConfiguration(HttpServletRequest request) {
    Assert.notNull(this.handlerMappings, "Handler mappings not initialized");
    HttpServletRequest wrapper = new RequestAttributeChangeIgnoringWrapper(request);
    for (HandlerMapping handlerMapping : this.handlerMappings) {
        HandlerExecutionChain handler = null;
        try {
            handler = handlerMapping.getHandler(wrapper); // 获取处理执行链
        }
        catch (Exception ex) {
            // Ignore
        }
        if (handler == null) {
            continue;
        }
        if (handler.getInterceptors() != null) {
            //遍历拦截器，如果拦截器同时实现了CorsConfigurationSource则用这个拦截器作为跨域配置源
            for (HandlerInterceptor interceptor : handler.getInterceptors()) {
                if (interceptor instanceof CorsConfigurationSource) {
                    return ((CorsConfigurationSource) interceptor).getCorsConfiguration(wrapper);
                }
            }
        }
        //从执行链获取处理器，如果处理器本身也实现了CorsConfigurationSource，则用处理器作为跨域配置源
        if (handler.getHandler() instanceof CorsConfigurationSource) {
            return ((CorsConfigurationSource) handler.getHandler()).getCorsConfiguration(wrapper);
        }
    }
    return null;
}

这个CorsConfigurationSource实现类根据请求从HandlerMapping中获取获取HandlerExecutionChain执行链，再依次从执行链的拦截器和处理器中获取CorsConfigurationSource，如果获取到了再调用其HandlerMappingIntrospector#getCorsConfiguration方法返回跨域配置。具体来说就是那两个if判断。

所以这么说来的话，HandlerMappingIntrospector虽然实现了CorsConfigurationSource但其本质有点像一个委托类？它检查请求对应的执行链上的拦截器和处理器有没有实现CorsConfigurationSource，如果有，再委托给这个CorsConfigurationSource来获取CorsConfiguration。所以说如果我们在一个Controller的接口上标注了@CrossOrigin注解，那么对应的，在拦截器中获取不到CorsConfiguration，就会从这个Handler上获取到CorsConfiguration，也就是将@CrossOrigin注解中提供的信息封装成了CorsConfiguration。那为什么还会先检查执行链中的拦截器呢？

因为SpringMVC中还有第二种方法配置跨域支持，也就是上面提到的重写WebMvcConfigurer#addCorsMappings方法进行全局配置。那为什么重写这个方法添加跨域配置最后会注册成拦截器呢？（一个实现了CorsConfigurationSource的拦截器）

这就要说到SpringBoot在WebMvc的自动配置、WebMvcConfigurer和HandlerMapping了。

如果你有仔细看过SpringBoot在SpringMVC的自动配置方面的源码，你一定知道WebMvcConfigurationSupport这个最主要的配置类在注册HandlerMapping的时候会从一个CorsRegisty中获取跨域配置：（这里以RequestMappingHandlerMapping为例）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

@Bean
public RequestMappingHandlerMapping requestMappingHandlerMapping(
        @Qualifier("mvcContentNegotiationManager") ContentNegotiationManager contentNegotiationManager,
        @Qualifier("mvcConversionService") FormattingConversionService conversionService,
        @Qualifier("mvcResourceUrlProvider") ResourceUrlProvider resourceUrlProvider) {
            
    RequestMappingHandlerMapping mapping = createRequestMappingHandlerMapping();
    mapping.setOrder(0);
    mapping.setInterceptors(getInterceptors(conversionService, resourceUrlProvider));
    mapping.setContentNegotiationManager(contentNegotiationManager);
    mapping.setCorsConfigurations(getCorsConfigurations()); //设置跨域配置
    // ...省略一大段set
    return mapping;
}
protected final Map<String, CorsConfiguration> getCorsConfigurations() {
    if (this.corsConfigurations == null) {
        CorsRegistry registry = new CorsRegistry();
        addCorsMappings(registry);  //向CorsRegistry中添加跨域映射
        this.corsConfigurations = registry.getCorsConfigurations(); //获取跨域配置
    }
    return this.corsConfigurations;
}

addCorsMappings()方法是个空方法，并且只有DelegatingWebMvcConfiguration类重写了这个方法。实际上WebMvcConfigurationSupport这个类中用@Bean这个可传递的注解标注了很多方法但该类上并没有标注@Configuration，那么为什么还会起到配置类的作用呢？其实真正的配置类是DelegatingWebMvcConfiguration。

DelegatingWebMvcConfiguration

在DelegatingWebMvcConfiguration这个类上有个@Configuration注解，并且继承自WebMvcConfigurationSupport，实际上它就是个委托类。

可以说这个类才是真正的配置类，去看看DelegatingWebMvcConfiguration这个类，相信你一定会发现什么！！！

DelegatingWebMvcConfiguration中有WebMvcConfigurerComposite这么一个对象，并且将容器中所有WebMvcConfigurer注入进来：

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

@Configuration(proxyBeanMethods = false)
public class DelegatingWebMvcConfiguration extends WebMvcConfigurationSupport {
    // WebMvcConfigurer复合类
    private final WebMvcConfigurerComposite configurers = new WebMvcConfigurerComposite();
    // 将容器中所有WebMvcConfigurer添加到WebMvcConfigurerComposite
    @Autowired(required = false)
    public void setConfigurers(List<WebMvcConfigurer> configurers) {
        if (!CollectionUtils.isEmpty(configurers)) {
            this.configurers.addWebMvcConfigurers(configurers);
        }
    }
    // ...省略
}

如果你去看了一下这个类的源码，你就会发现WebMvcConfigurer中有的方法这个类都有，并且这个委托类仅仅是将请求委托给configurers，来看看重写的addCorsMappings方法：

1
2
3
4

@Override
protected void addCorsMappings(CorsRegistry registry) {
    this.configurers.addCorsMappings(registry);
}

调用WebMvcConfigurerComposite#addCorsMappings，显而易见WebMvcConfigurerComposite是个复合的WebMvcConfigurer，他也实现了WebMvcConfigurer并且内部维护了一个List<WebMvcConfigurer> delegates列表，实现的所有方法会依次调用列表中WebMvcConfigurer对应的方法。（并且你还能发现WebMvcConfigurer中的方法都是作为回调方法并且大部分是返回void的）

说到这里，不得不说一个题外话。如果看Spring源码比较多的话，就会发现Spring中类的命名都有规律可循并且某些后缀都是有特定意义的，比如xxxComposite、xxxConfigurer、Delegatingxxx、xxxDelegator等等，这样我们看到这个类名就立马能猜到它的作用。

我们平时对WebMvc进行一些配置都是实现WebMvcConfigurer类，重写其中的方法。下面是一个例子：

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

@Configuration
public class WebMvcConfig implements WebMvcConfigurer {

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/**")
                .allowedOrigins("*")
                .allowedHeaders("*")
                .allowedMethods("*")
                .allowCredentials(true)
                .maxAge(3600);
    }
    // ...
}

说到这，也就是相当于WebMvcConfigurationSupport#getCorsConfigurations方法会回调容器中所有WebMvcConfigurer实现类的addCorsMappings()方法，向CorsRegistry中添加跨域映射，然后再取出CorsConfiguration返回：

1
2
3
4
5
6
7
8

protected final Map<String, CorsConfiguration> getCorsConfigurations() {
    if (this.corsConfigurations == null) {
        CorsRegistry registry = new CorsRegistry();
        addCorsMappings(registry); //向CorsRegistry中添加跨域映射
        this.corsConfigurations = registry.getCorsConfigurations(); //获取跨域配置
    }
    return this.corsConfigurations;
}

最后反应到AbstractHandlerMapping中的就是使用CorsConfiguration注册一个CorsInterceptor拦截器，这个拦截器是AbstractHandlerMapping中的一个内部类，继承自HandlerInterceptorAdapter，并且实现了CorsConfigurationSource。

看到这里，如果没有了解过HandlerMapping，可能会一头雾水，可以看看我的这篇文章源码角度分析Spring容器启动阶段注册Controller处理器的流程，虽然不是讲HandlerMapping，但是相信在看完后，会对HandlerMapping有一个理解。

CorsInterceptor

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

private class CorsInterceptor extends HandlerInterceptorAdapter implements CorsConfigurationSource {
    @Nullable
    private final CorsConfiguration config;
    
    public CorsInterceptor(@Nullable CorsConfiguration config) {
        this.config = config;
    }

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)throws Exception {
        return corsProcessor.processRequest(this.config, request, response);
    }

    @Override
    @Nullable
    public CorsConfiguration getCorsConfiguration(HttpServletRequest request) {
        return this.config;
    }
}

这个类只重写了拦截器的preHandle方法，其他方法都是空方法。而且你能发现这个preHandle方法中的内容和CorsFilter#doFilterInternal方法基本是一模一样的，都是根据CorsConfiguration使用跨域处理器处理请求。

看到这里，现在应该知道关于HandlerMappingIntrospector的猜测是没错的，并且知道了HandlerMappingIntrospector是如何与SpringMVC两种支持跨域的配置方式联系起来的，这里再次总结一下：

1. 首先获取请求对应的执行链上的拦截器，判断拦截器有没有实现CorsConfigurationSource（CorsInterceptor类），如果有则调用getCorsConfiguration获取CorsConfiguration后返回
2. 如果拦截器上获取失败，则判断处理器有没有实现CorsConfigurationSource（PreFlightHandler类），如果有则调用getCorsConfiguration获取CorsConfiguration后返回

从而实现了兼容SpringMVC中两种配置跨域的方式。

这其中最关键的几点就在于CorsConfigurer获取CorsConfigurationSource并且构造CorsFilter的步骤、HandlerMappingIntrospector获取CorsConfiguration的步骤，还有Spring回调WebMvcConfigurer对HandlerMapping进行设置跨域配置等信息的步骤

其中还涉及到了SpringMVC中HandlerMapping、HandlerExecutionChain、Handler、Interceptor等相关知识。

根据这次的分析，能得到几个结论：

• SpringMVC支持跨域两种方式一个是基于处理器实现，另一个是基于拦截器实现。
• SpringSecurity跨域是基于过滤器，并且兼容了SpringMVC的两种配置（使用HandlerMappingIntrospector“桥接”）。
• SpringSecurity中的CorsConfigurer使用HandlerMappingIntrospector来兼容SpringMVC跨域两种方式。
• HandlerMappingIntrospector获取CorsConfiguration时的优先级是先拦截器，再处理器。
• SpringBoot注册HandlerMapping或者说通过WebMvcAutoConfiguration自动配置来对WebMvc必要的组件进行装配和注入。
• WebMvcConfigurer是DelegatingWebMvcConfiguration类驱动WebMvcConfigurerComposite来进行回调的。

并且经过这次的源码阅读，也是足足感受到Spring设计上的优雅。

写在最后

文章写的有点乱，并且有点跳跃。仅仅是跟着文章来看可能不大能看懂，最好在电脑上根据源码来阅读。这篇文章也仅仅是作为我个人在一次踩坑后好奇心大法，阅读源码后的一段总结以及感悟吧，自己能看懂并且以后还能看懂也就满意了。如果这篇文章有幸被你刷到并且你能够看懂我想表达的那我自然是更高兴。其实这个博客存在的理由也仅仅是为了记录自己学习过程中的感悟和总结，便于自己以后回顾，毕竟我比较健忘。所以需要记录下有必要的，并且在个人看来，这篇文章干货还是足足的，所以说更加有必要记录。其实在写文章之初我也不想写这么一篇文章，因为实在是太难写明白了，并且由于涉及到的东西比较分散很难进行组织，也可能我表达能力差的原因吧，但最终还是花了一下午加一晚上，在不断修改下产出了这么一篇很长很长很长的文章，可能是写过的字数最多的文章了吧😥。文章中可能有错别字也可能有错误的内容，如果你发现文章有什么错误的地方或者没表述清楚的内容，欢迎在评论中交流。

写在前面

原本是打算写一个VueJS学习笔记专栏，用于记录vue学习过程中的一些感想，但后来想想，如此做实在是太麻烦，并且每篇文章的篇幅也会比较短，所以现在考虑直接在这一篇博文中进行总结。将学习过程中的感谢或者踩得一些坑直接记录在此一篇文章中，尽量做到每个点都短小精悍。该篇博文并非最终稿，内容会随着文章的更新不断丰富。

官方中文文档：

• Guide：https://cn.vuejs.org/v2/guide/
• API：https://cn.vuejs.org/v2/api/
• Style Guide：https://cn.vuejs.org/v2/style-guide/
• Examples：https://cn.vuejs.org/v2/examples/

下面大部分内容都可以从官方文档中找到。这篇总结，仅将我个人认为有必要记录的进行汇总，以方便后续查找和回顾。

正文

1. 在选项属性或者回调方法使用箭头函数会导致在方法内使用this获取不到vue实例

1
2

created: () => console.log(this.a)
vm.$watch('a', newValue => this.myMethod())

2. 事件总线机制下，监听总线事件的回调方法要使用箭头函数，否则this指代总线实例。

1
2
3
4
5

mounted(){ // 挂载后执行
  EventBus.$on('event-name',data => { // 这里要用箭头函数，否则this指代EventBus
this.data = data;
  })
}

3. v-bind、v-on、v-slot指令使用动态指令参数（2.6.0+）

1
2
3

<a :[attr]="value">...</a>
<a @[event]="value">...</a>
<template #[solt]>...</template>

attr、event和solt会作为一个javascript表达式进行求值

4. 在DOM中使用模板时，避免使用大写字符来命名键名，因为浏览器会把attribute名全部强制转换为小写。

5. 数据一般是从父组件流向子组件的，prop的值会因父组件中数据的改变而改变。

6. 不要修改prop的值，如果一定要这么做，请使用计算属性或者data。

7. 计算属性默认只有getter，但是我们也可以提供一个setter。

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

computed: {
  fullName: {
    // getter
    get: function () {
        return this.firstName + ' ' + this.lastName
    },
    // setter
    set: function (newValue) {
        var names = newValue.split(' ')
        this.firstName = names[0]
        this.lastName = names[names.length - 1]
    }
  }
}

8. 父组件中的data可以是一个对象也可以是一个方法，但是子组件的data属性必须是一个方法。

9. 由于JavaScript的限制，Vue不能检测数组和对象的变化。也就是说对对象的属性或数组进行修改添加删除不会反应到视图中（但值的却改变了），可以使用vm.$set实例方法（全局方法的别名），或Vue.set全局方法。

1
2
3
4
5
6
7
8
9

var vm = new Vue({
  data:{
    lisi:{
      age: 16    
    }
  }
})
vm.lisi.age = 18; // 直接对对象属性进行修改是不会直接响应到视图中
vm.$set(this.lisi,'age',18); // 响应式

10. v-if和v-show的异同：

• v-if指令是直接销毁和重建DOM达到让元素显示和隐藏的效果
• v-show指令通过修改元素的display属性让其显示或者隐藏

11. 使用v-cloak指令解决页面加载时闪烁的问题

12. 事件修饰符、按键修饰符以及系统修饰键

13. 使用ref属性在父组件中调用子组件的方法。

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

<div id="app">
    <button @click="add"> Click this button {{count}} times.</button>
    <children ref="child"></children>
</div>
<template id="btn">
    <div>
        <button @click="add"> Click this button {{count}} times.</button>
    </div>
</template>
<script>
    var btn = {
        template: '#btn',
        data() {
            return {
                count: 0
            }
        },
        methods: {
            add() {
                this.count++;
            }
        }
    }
    const vm = new Vue({
        el: '#app',
        data:{
            count: 0
        },
        methods: {
            add() {
                this.count++;
                this.$refs.child.add(); // 调用子组件方法
            }
        },
        components: {
            'children': btn
        }
    })
</script>

正式将评论插件由valine更改为gitalk，提高了评论的门槛，但考虑到文章受众以及博客类型，还是可以接受的。之前的评论丢失，后续也可能会考虑回到valine。

不要在选项属性或回调上使用箭头函数，比如created: () => console.log(this.a)或vm.$watch('a', newValue => this.myMethod())。因为箭头函数并没有this，this会作为变量一直向上级词法作用域查找，直至找到为止，经常导致Uncaught TypeError: Cannot read property of undefined或Uncaught TypeError: this.myMethod is not a function之类的错误。

错误示范：

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

<el-button type="primay" icon="el-icon-plus" @click="add"></el-button>
<p>{{count}}</p>
<script>
    const app = new Vue({
        // ...
        data: {
            count: 1
        },
        methods: {
            add: () => this.count++
        }
    });
</script>

正确示范：

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

<script>
    const app = new Vue({
        // ...
        methods: {
            add: function () {
                this.count++
            },
            minus() {
                this.count--
            }
        }
    });
</script>

前言

用ssh连接GitHub，需要在GitHub账号上传唯一的公钥。当我们需要连接两个或多个GitHub账号，上传同一个公钥是不允许的，那么该如何设置才能在一台电脑上连接多个GitHub账号呢？

创建密钥并上传到GitHub

假设我们已经有一对默认的密钥id_rsa、id_rsa.pub关联了a账号，现在我们来为b账号创建一个密钥：

1

ssh-keygen -t rsa -f ~/.ssh/id_rsa_b -C "youmail@example.com"

注意文件名不要与其他的密钥重复，否则会覆盖之前的密钥。
windows下使用绝对路径：C:\Users\xxx\.ssh\id_rsa_b

接下来我们就可以在用户目录下的.ssh文件夹中找到id_rsa_b和id_rsa_b.pub两个文件，将id_rsa_b.pub文件中的内容保存到b账号的ssh keys中。

配置身份验证使用的密钥

接下来我们要做的就是对不同账号配置使用不同的密钥去连接github。在.ssh文件夹中创建一个config文件（无扩展名），填入以下内容：

1
2
3
4
5
6
7
8
9
10
11

# default
Host github.com
    HostName github.com
    PreferredAuthentications publickey
    IdentityFile ~/.ssh/id_rsa

# b
Host b.github.com
    HostName github.com
    PreferredAuthentications publickey
    IdentityFile ~/.ssh/id_rsa_b

在这部分配置中我们使用不同的主机标识去映射同一个HostName即github.com，但是我们使用不同的密钥文件去进行身份验证。

配置中的Host在不和其他Host重复的情况下可以随意填写，但是建议使用一眼就能看懂的标识（并且这个标识在后面是需要用到的）

到这里，其实所有的工作都已经完成。在进行下一步操作前我们先测试ssh是否能连接到github：

1
2

ssh -T git@github.com
ssh -T git@b.github.com

不出意外，我们将看到验证成功的提示。如果提示 CreateProcessW failed error:2，先将config文件中的代理设置注释便可，实际上并不影响后续对仓库的推送。

对b账号的仓库单独设置用户名和邮箱

现在我们要做的就是对b账号下的仓库进行设置，因为在原本只有一个账号时，我们应该都用过下面这个命令来对git设置一个全局的用户名和邮箱：

1
2

git config --global user.name "a"
git config --global user.email "a@example.com"

现在我们当然不能对b账号下的仓库也使用全局的用户名和邮箱。如果使用全局的邮箱将b的仓库push到github，那么在github提交记录中看到的提交者将会是a，所以说我们需要对b账号下的仓库单独配置用户名和邮箱，进入项目文件夹：

1
2

git config user.name "b"
git config user.email "b@example.com"

邮箱要使用b账号注册github时使用的邮箱，因为github提交记录中的提交者是根据这个邮箱来查找的，这也是不能使用全局邮箱后提交者是a的原因。

修改远程仓库的地址

接下来我们还要重新设置远程仓库的地址，因为克隆仓库或者创建github仓库添加远程仓库地址时使用的主机标识默认是github.com，然而在config文件中对这个Host是使用~/.ssh/id_rsa密钥去验证，所以当我们本地已有b账号下的仓库或者未来克隆b账号下的仓库时要修改默认的远程仓库地址，两种方式：

先删除再添加

1
2

git remote rm origin
git remote add origin git@b.github.com:b/repo.git

注意不要照抄，根据config文件中配置的Host，github账号以及仓库名去设置‘@’符号后面的东西。

所有的工作都已经完成，push测试一下：

1

git push origin master

如果我们单纯的使用git push可能会报错，因为虽然添加了远程仓库，但是并没有设置本地分支具体跟踪哪个上游分支。使用-u或--set-upstream-to选项运行git branch来显式地设置：

1

git branch -u origin/master master

直接修改远程仓库地址

1、Git查看远程仓库名称

1

git remote -v

2、修改地址

1

git remote set-url origin git@b.github.com:b/repo.git

总结

总的来说，我们就是使用了不同的主机标识去映射github.com，并且对不同的标识在连接github时使用相应的密钥去验证身份。我们设置多个标识，相应的，我们也要修改远程仓库使用的Host标识（因为克隆下来的仓库的远程地址Host标识默认使用github.com）。虽然上面我们是拿两个账号来举例子，但是对于多个账号，设置的方法还是一模一样的。

python3编译安装dlib库十分麻烦，这里提供已经编译好的whl文件，使用pip直接安装即可。

• python3.6
  下载：dlib-19.8.1-cp36-cp36m-win_amd64.whl
  md5：9a704406fbb4036f70b5f174fec9db1f
  sha1：e2e49b82b8dee6a9483362713239558c42c38fea

• python3.7
  下载：dlib-19.17.99-cp37-cp37m-win_amd64.whl
  md5：b8c35e6e0098a7ece3e1a6935bbf1ae8
  sha1：886f510821c3033649edcf86ea80ec1157f0b1a7

• python3.8
  下载：dlib-19.19.0-cp38-cp38-win_amd64.whl
  md5：7e00b04c3cf468e102b338d6837f4d5d
  sha1：129d0142f1232d60c6571dc07eacdcd392b92f7b

安装：

1

$ pip install dlib-xx.xx.xx-cpxx-cpxx-win_amd64.whl  #安装相应的文件