什么是 Partition

kafka 将 一个Topic 中的消息分成”多段”，分别存储在不同的 Broker 里，这每一段消息被 kafka 称为 Partition

有人可能会问 kafka 为什么要将 Topic 中的消息分段储存在不同的Broker 里？因为一个 Topic 中的消息可能非常多，一台机器可能存不下，因此需要拆分成多段存储在不同的机器里，而且这样做还能提高读写性能

什么是 Replication

Partition 只存一份肯定不安全，为了数据的存储安全，以及实现高可用，可以让 kafka 将 Partition 存储多份存到不同的 Broker 中，这每一份 Partition 被 kafka 称为 Replication

Replication 逻辑上是一个 Partition 的副本，本质上跟 Partition 一样，还是一段 Topic 消息。

我们说一个文件有三个副本，那么一共应该有4个文件，但是在 kafka 中，我们说一个 Partition 下面有3个 Replication 时，那么总共只有3分数据。这点和我们正常理解的有些不一样，因为在 kafka 中 Replication 才是真正对外提供服务的实例！

因此，当一个 Partition 中只有一个 Replication 时，它俩可以理解成一个东西，当一个 Partition 中有多个 Replication 时，他两的关系更像是类和实例之间的关系，Partition 是分区这个类，而 Replication 是分区这个类的实例

出于节省存储空间的考虑，kafka 默认策略是只有一个Replication

Replication 的两种角色

虽然 Partition 有多个副本，但 kafka 规定只有一个能对外提供服务，这个副本被 kafka 称为 Leader Replication，而其他的副本被称为 Follower Replication

Replication 之间的数据同步

Producer 和 Consumer 只会与 Leader Replication 交互，而其它 Follower Replication 则从 Leader Replication 中同步数据。

Follower 与 Leader 的同步是有延时的，会受到下面两个参数的影响：

• replica.lag.time.max.ms：同步副本滞后与 leader 副本的时间
• zookeeper.session.timeout.ms：borker 与 zookeeper 会话超时时间

in-sync Replica

Follower Replication 必须及时从 Leader 中同步消息，不能 “落后太多”。这里的 “落后太多” 指 Follower Replication 落后于 Leader 消息的条数超过阈值或者 Follower 超过一定时间未向 Leader 发送 fetch 请求。

Leader 会跟踪与其保持一定程度同步的 Replication 列表，该列表称为 in-sync Replica(ISR)。如果一个 Follower 宕机，或者落后太多，Leader 将把它从 ISR 中移除。

Leader 中的一条消息只有被 ISR 里的所有 Follower 都接受到才会被认为已提交, 只有已提交消息，才会被 consumer 消费到。但是，为了写入性能， Follower 在接收到数据后就立马向 Leader 返回 ACK，而非等到数据写入 Log 中。因此，对于已经 commit 的消息，Kafka 只能保证它被保存在 ISR 里的所有 Follower 的内存中，而不能保证它们被持久化到磁盘中

in-sync Replica 的伸缩

Kafka 中有个名为 isr-expiration 任务会周期性的检测每个分区是否需要缩减其 ISR 集合。这个周期为 replica.lag.time.max.ms 参数值的一半。当 Follower 超过 replica.lag.time.max.ms 毫秒没有发起 fetch 请求或者，消息落后Leader rerplica.lag.max.messages 条时，就会被移出 ISR 列表

有缩减就会有补充，随着同步的进行，当 out-sync Replica(OSR) 集合中的某个 Follower Replication 的 LEO 大于等于该分区的 High Watermark(LW) 时，会被加入到 in-sync Replica(ISR) 集合中

in-sync Replica 相关的配置

1
2

ini复制代码# Follower 最大请求间隔
rerplica.lag.time.max.ms=10000

如果 Leader 发现 Follower 超过10秒没有向它发起 fetch 请求, 那么 Leader 会将其移出 ISR 列表

1
2

ini复制代码# Follower 消息落后的最大条数，超出该条数 Follower 会被移出 ISR 列表
rerplica.lag.max.messages=4000

1
2

ini复制代码# ISR 中至少要有多少个 Replication
min.insync.replicas=1

Out-Sync Relipca

与 in-sync Replica(ISR) 对应的还有 Out-Sync Relipcas(OSR)，即非同步副本集、也可以称作落后的副本列表

Assigned Repllica

Assigned Relipcas(AR) 指的是分区中的所有副本，AR = ISR + OSR

Leader 的几种选举

Leader Replication 如果挂掉就需要从 Follower Replication 中选出一个新的 Leader

同步列表选举

kafka 会优先从 ISR 中选取，因为 ISR 中的 Follower Replication 和 Leader 是保持同步的，因此数据一致性最高。

注意，是数据一致性最高，而不是数据一致,因为首先 Follower 与 Leader 的同步是有延时的，其次数据的复制是异步完成的。所以数据可能会丢失，但是概率和数量都很小

始终保证拥有足够数量的同步副本是非常重要的。要将 follower 提升为 Leader，它必须存在于同步副本列表中。每个分区都有一个同步副本列表，该列表由 Leader 分区和 Controller 进行更新。

从 ISR 中选取 leader 的过程称为 clean leader election

非同步列表选举

由于 ISR 是动态调整的，所以会存在 ISR 列表为空的情况，这时候可以从 ISR 之外选取。

通常来说，非同步副本落后 Leader 太多，因此，如果选择这些副本作为新 Leader，极有可能会出现数据的丢失。

在非同步副本中选取 leader 的过程称之为 unclean leader election

因此是否开启 unclean leader election，可以通过 Broker 端参数 unclean.leader.election.enable 来控制，当然也要根据你的具体业务去权衡！

开启 unclean leader election 很可能会造成数据丢失，但好处是，它使得 Leader 一直存在，不至于对外停止服务，因此提升了高可用性。反之，禁用 unclean leader election 的好处在于维护了数据的一致性，避免了消息丢失，但代价是牺牲了高可用性。分布式系统的 CAP 理论说的就是这种情况。

Leader 、follower 角色互换

除了上面提到的两种选举会导致 Leader 角色发生移动之外，还有一种情况也会造成 Leader 移动。

当 Controller Broker 从 Zookeeper 上感知到有新的 Broker 已加入集群时，它将使用该 Broker ID 来检查该 Broker 上是否存在Replication，如果存在，则 Controller 通知新加入的 Broker 同步现有对应 leader 的消息到自己的 follower 分区。之后，为了保证负载均衡，Controller 会下线当前 leader 分区，然后将新加入的 Broker 上的 follower 分区选举为 新leader 分区。

这就是 Leader 、follower 的角色互换，一般发生在有新的 Broker 加入集群时，目的是为了实现 Partition 的负载均衡

换一次 Leader 是有代价的，原 Leader 分区请求都会打到新的 Leader 分区。如果你不想发生 Leader 、follower 角色互换，可以在配置文件中将这个参数设置成 false

Leader Replication 负载均衡

由于 KafKa 客户端读写 topic 的分区时，只会跟 Leader Replication 交互，如果一个 Broker 中存在过多的 Leader Replication，就会导致该 Broker 压力过大。

Partition 配置

配置 topic 默认的副本数

1

ini复制代码default.replication.factor=3

如果创建 topic 时，没有显示指定副本数，则使用默认的副本数

创建 topic 时，可以通过 --replication-factor 显示指定副本数

1

arduino复制代码bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

一、前言

「本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！」

在上篇 精读 RocketMQ 源码系列（1）— NameServer 中我们有一个遗留问题：

由于从 broker 宕机到 NameServer 路由删除有120秒的间隔，会导致生产者可能会向一个已经宕机的 broker 发送消息，这种情况 RocketMQ 是如何处理的呢？

本文会对这个问题做一个解释。同时本文会从源码角度着重分析两个问题：

1. Producer 的启动流程是怎样的？
2. Producer 是如何把消息发送到 broker 上的？

需要强调的是，本文不会详细讲和 Producer、消息相关的一些概念，对这一块不太熟悉的同学可以在 精读 RocketMQ 源码系列（0）—开篇词 这篇中找到官方的中文文档，进行了解。

二、启动流程

RocketMQ 中生产者的核心类是 DefaultMQProducer，启动流程的源码入口是 DefaultMQProducer#start()。流程图如下：

大家可以对照流程图阅读源码，这里做个简单总结，启动流程步骤大致如下：

1. 检查生产者组是否合法
2. 获取 MQClientInstance
3. 将当前生产者注册到 MQClientInstance（注册可以理解为将 producer set 到 MQClientInstance）
4. 启动 MQClientInstance 客户端

接下来对一些小细节分析下

2.1 MQClientManager

字面意思上理解，MQ 客户端管理者。

整个 JVM 中只存在一个 MQClientManager 实例，为什么呢？

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

java复制代码public class MQClientManager {
    private final static InternalLogger log = ClientLogger.getLog();
    // MQClientManager 实例，对外暴露
    private static MQClientManager instance = new MQClientManager();
    private AtomicInteger factoryIndexGenerator = new AtomicInteger();
    // MQClientInstance 缓存表
    private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
        new ConcurrentHashMap<String, MQClientInstance>();

    private MQClientManager() {

    }

    public static MQClientManager getInstance() {
        return instance;
    }
...
}

对外暴露的instance是一个静态变量，只有在类初次加载的时候会被初始化，相关信息存储在 JVM 中，具有唯一性。

它维护一个 MQClientInstance 缓存表：ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable。同一个 clientId 只会创建一个 MQClientInstance。所以总结来说，在一个 JVM 实例中，只会有一个 MQClientManager 存在，但如果运行了多个应用程序（客户端），就会存在多个 MQClientInstance。

我们可以看一下 clientId 是怎么生成的：

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

java复制代码    public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClientIP());

        sb.append("@");
        sb.append(this.getInstanceName());
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }

        return sb.toString();
    }

clientId是由：ip地址、实例名、unitName（可选）拼接而成的。那这就有问题了，同一实例中 ip 地址和实例名都一样啊。

其实这里实例名已经被修改了，可以看到这里：已经把实例名改成了进程id

1
2
3

java复制代码   if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                    this.defaultMQProducer.changeInstanceNameToPID();
      }

2.2 MQClientInstance

MQClientInstance 中封装了 RocketMQ 网络处理的 API，是消费者和生产者与 NameServer、Broker 通信的网络通道。

1
2
3
4

java复制代码  if (consumerEmpty) {
      if (id != MixAll.MASTER_ID)
         continue;
  }

以上代码片段位于MQClientInstance的sendHeartbeatToAllBroker()方法，表明了生产者只会向 Master 的 broker 发送心跳

创建 MQClientInstance 的源码：

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

java复制代码    public MQClientInstance getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
        String clientId = clientConfig.buildMQClientId();
        MQClientInstance instance = this.factoryTable.get(clientId);
        if (null == instance) {
            instance =
                new MQClientInstance(clientConfig.cloneClientConfig(),
                    this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
            MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
            if (prev != null) {
                instance = prev;
                log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
            } else {
                log.info("Created new MQClientInstance for clientId:[{}]", clientId);
            }
        }

        return instance;
    }

利用 ConcurrentMap 来保证并发时不会出错，同时通过双重校验，保证多线程场景下，返回的实例是同一个。

2.3 心跳机制

在 MQClientInstance 启动之后，还有一行代码很重要：

1

java复制代码this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

注意，这里的 AllBroker 当然不是集群中所有的 Broker，而是与当前客户端相关的 Broker。

三、发送消息

启动完成之后，Producer 就可以开始发送消息了。可以看到在 DefaultMQProducer 中发送消息的方法非常多，大致可进行如下分类：

根据消息类型：

• 普通消息：没有什么特殊的地方，就是普通消息
• 延迟消息：延时消息在投递时，需要设置指定的延时级别，即等到特定的时间间隔后消息才会被消费者消费。mq服务端 ScheduleMessageService中，为每一个延迟级别单独设置一个定时器，定时(每隔1秒)拉取对应延迟级别的消费队列。目前RocketMQ不支持任意时间间隔的延时消息，只支持特定级别的延时消息，即 “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”
• 顺序消息：对于指定的一个 Topic，Producer保证消息顺序的发到一个队列中，消费的时候需要保证队列的数据只有一个线程消费。
• 事务消息：通过两阶段提交、状态定时回查来保证消息一定发到broker。具体流程见下图

根据发送方式：

• 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
• 可靠异步发送：异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。 MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，进行第二条消息发送。发送方通过回调接口接收服务器响应，并对响应结果进行处理。
• 单向（Oneway）发送：特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。 此方式发送消息的过程耗时非常短，一般在微秒级别。

这里，我们选择分析发送同步消息，发送消息的流程图如下：

简单总结为以下几个主要步骤：

1. 校验消息：主题和消息体的校验
2. 查找主题路由信息：注意这里查找出来的信息是以 MessageQueue 维度存储的
3. 选择消息队列：第2步中会返回待发送消息对应主题下的所有 Broker 的 MessageQueue 的信息，这一步就是在这些 MessageQueue 中选择一个进行发送
4. 执行具体发送消息的动作

3.1 选择 MessageQueue — 默认方案

MQFaultStrategy#selectOneMessageQueue

1
2
3
4
5
6
7
8

java复制代码    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        if (this.sendLatencyFaultEnable) {
            ...... // 业务逻辑
            return tpInfo.selectOneMessageQueue();
        }

        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }

根据参数 sendLatencyFaultEnable ，我们有两种方案，一种称之为默认选择方案，另一种为启用故障延迟后的方案。可以看到启用故障延迟后的方案实际调用了默认的方案，我们先看看默认方案是如何做的？

TopicPublishInfo#selectOneMessageQueue

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

java复制代码    public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
      // lastBrokerName 表示上次发送消息给了哪个 broker
        if (lastBrokerName == null) {
            return selectOneMessageQueue();
        } else {
            for (int i = 0; i < this.messageQueueList.size(); i++) {
              // sendWhichQueue 记录了一个 index，可自增，使用到了 ThreadLocal
                int index = this.sendWhichQueue.getAndIncrement();
                int pos = Math.abs(index) % this.messageQueueList.size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = this.messageQueueList.get(pos);
                if (!mq.getBrokerName().equals(lastBrokerName)) {
                    return mq;
                }
            }
            return selectOneMessageQueue();
        }
    }

可以看到，选择 MessageQueue 的方案其实很简单：

• 维护一个可自增的值 sendWhichQueue 每次将其与总的 MessageQueue 的数量取模获得新的 MessageQueue 的下标；
• 当选择的新的 MessageQueue 属于上次的 Broker 时，重新选择。

这么做可以使得相邻两次发送的消息不会发送到同一个 broker 上，实现负载均衡；同时当其中一个 broker 宕机时，可以最大限度减少消息发送到宕机的 broker 上。

3.2 选择 MessageQueue — 故障延迟方案

当 sendLatencyFaultEnable 开启时，我们会执行以下逻辑分支：

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

java复制代码    if (this.sendLatencyFaultEnable) {
            try {
                int index = tpInfo.getSendWhichQueue().getAndIncrement();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
                        return mq;
                }

                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                    }
                    return mq;
                } else {
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }

            return tpInfo.selectOneMessageQueue();
        }

整体逻辑，总结如下：

1. 选择出一个 MessageQueue，方法与默认方案的方法相同
2. 校验该 MessageQueue 是否可用，可用则直接返回
3. 不可用则在尝试从规避的 Broker 中选择一个可用的 broker，如果选出来的 broker 有写队列则返回
4. 如果无可写队列则最后再用默认方案选出一个队列返回

故障延迟机制的核心是使用了

1

java复制代码ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);    class FaultItem implements Comparable<FaultItem> {        private final String name;  // broker name        private volatile long currentLatency;  // 消息发送故障延迟时间        private volatile long startTimestamp;  // 不可用时间戳，当前时间不超过这个时间戳表示该需要规避该 broker				...    }

每次选择出一个队列之后，需要通过内存的一张表faultItemTable去判断当前这个Broker是否在其中，如果不在证明可用，直接返回即可；如果在，证明可能不可用，需要再判断一下

1
2
3

go复制代码   public boolean isAvailable() {            
       return (System.currentTimeMillis() - startTimestamp) >= 0;   
       }

该表是每次发送消息的时候都会更新。

再之后就是调用发送消息的核心方案 sendKernelImpl，进行消息的组装和发送。感兴趣的同学可对照流程图读一下源码

四、前言中提到的几个问题

启动流程和消息发送分别都已经在第二和第三节中叙述了。现在来看下生产者是如何应对 broker 宕机的问题的，同时这也是上一篇文章中遗留的一个问题。

4.1 生产者是如何应对 broker 宕机

我们来看看发送消息的方法 sendDefaultImpl 中可以看到有这么一段 for 循环

1

java复制代码  for (; times < timesTotal; times++) {    ...// 消息发送             sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);       endTimestamp = System.currentTimeMillis();       this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);       switch (communicationMode) {           case ASYNC:           return null;           case ONEWAY:           return null;           case SYNC:           if (sendResult.getSendStatus() != SendStatus.SEND_OK) {               // 开启了消息重试开关，则进行消息重试               if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {                   continue;                }           }           // 未开启，则直接返回结果，结束本次消息发送            return sendResult;            default:                break;        }  }

timesTotal 为重试次数+1，也就是说，开启了消息重试开关，生产者会进行消息重试。

结合刚刚我们讲的选择 MessageQueue 的方案，不论是默认方案还是故障延迟方案，在重新选择时，都会规避上一次的 broker。因此消息重试时是不会再选择到导致本次消息发送失败的 broker 的。

总结来说，RocketMQ 通过 消息重试+broker规避 实现了消息发送的高可用

4.2 生产环境下为什么不能自动创建Topic？

很多时候我们会被告知，生产环境下不要将 autoCreateTopicEnable 设置为 true，因为这会使得：自动新建的Topic只会存在于一台Broker上，后续所有对该Topic的请求都会局限在单台Broker上，造成单点压力。

但是为什么会这样呢？我们接下来来分析下

1、broker 启动时，会加载在本地创建的 topic

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

java复制代码 public BrokerController(
        final BrokerConfig brokerConfig,
        final NettyServerConfig nettyServerConfig,
        final NettyClientConfig nettyClientConfig,
        final MessageStoreConfig messageStoreConfig
    ) {
        this.brokerConfig = brokerConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.nettyClientConfig = nettyClientConfig;
        this.messageStoreConfig = messageStoreConfig;
        this.consumerOffsetManager = new ConsumerOffsetManager(this);
   			// 加载 topic 配置
        this.topicConfigManager = new TopicConfigManager(this);
        this.pullMessageProcessor = new PullMessageProcessor(this);
   			...
 }

在 TopicConfigManager的构造函数中，会判断 autoCreateTopicEnable ，然后对默认主题进行加载：

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

java复制代码    if (this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()) {
                String topic = TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC;
                TopicConfig topicConfig = new TopicConfig(topic);
                TopicValidator.addSystemTopic(topic);
                topicConfig.setReadQueueNums(this.brokerController.getBrokerConfig()
                    .getDefaultTopicQueueNums());
                topicConfig.setWriteQueueNums(this.brokerController.getBrokerConfig()
                    .getDefaultTopicQueueNums());
                int perm = PermName.PERM_INHERIT | PermName.PERM_READ | PermName.PERM_WRITE;
                topicConfig.setPerm(perm);
                this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
            }

可以看到这里创建了一个名为 AUTO_CREATE_TOPIC_KEY_TOPIC，读写队列都为 8 的主题信息。

接着该信息会被同步到 NameServer 上。

这里要注意的是，每一个开启了autoCreateTopicEnable 的broker 都会在启动时去加载默认主题信息并上报至 NameServer。那么在 NameServer 处存储的关于默认主题就会有多个 broker 信息

2、生产者发送消息，查询 topic 信息

生产者发送消息时，首先会使用 tryToFindTopicPublishInfo 去查询主题信息：

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

java复制代码    private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
        }

        if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
            return topicPublishInfo;
        } else {
            // 新创建的主题走这个分支
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
            return topicPublishInfo;
        }
    }

当然，现在主题的消息只在生产者这端存在，所以肯定找不到，最后只能走最下面的这个分支，来到 updateTopicRouteInfoFromNameServer 并执行以下逻辑：

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

java复制代码TopicRouteData topicRouteData;
if (isDefault && defaultMQProducer != null) {
   // 获取到 broker 创建的默认的主题信息
		topicRouteData = 				  this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
                            1000 * 3);
		if (topicRouteData != null) {
      // 将默认主题信息的读写队列数目更改为4
    		for (QueueData data : topicRouteData.getQueueDatas()) {
        		int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
            data.setReadQueueNums(queueNums);
            data.setWriteQueueNums(queueNums);
         }
     }
} 
....
TopicRouteData old = this.topicRouteTable.get(topic);
boolean changed = topicRouteDataIsChange(old, topicRouteData);

接着生产者会选择一个 MessageQueue，并将消息封装进行发送。注意，这里发送出去的消息，并不是默认主题的，而是消息本身的主题：

1

java复制代码// 代码位置：sendKernelImpl requestHeader.setTopic(msg.getTopic());

然后 broker 接收到消息后首先会对消息进行校验：AbstractSendMessageProcessor#msgCheck

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

java复制代码// 查询消息头的 topic 是否存在
        TopicConfig topicConfig =
            this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
        if (null == topicConfig) {
            int topicSysFlag = 0;
            if (requestHeader.isUnitMode()) {
                if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
                } else {
                    topicSysFlag = TopicSysFlag.buildSysFlag(true, false);
                }
            }

            log.warn("the topic {} not exist, producer: {}", requestHeader.getTopic(), ctx.channel().remoteAddress());
            // 不存在则根据消息中的相关信息进行主题的创建
            topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(
                requestHeader.getTopic(),
                requestHeader.getDefaultTopic(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                requestHeader.getDefaultTopicQueueNums(), topicSysFlag);

那么此时，新主题的信息就在这台 broker 上有了。

3、接下来的步骤

接下来事情的走向就是：

• broker 通过心跳机制上报主题消息，包括刚刚新创建的主题
• NameServer 接收来自 broker 的主题信息并更新路由信息
• 生产者再次往刚创建的新主题发消息时，发现新主题在 NameServer 端有路由，那就取到路由信息，按照路由信息进行发送

发现问题没有？到目前为止，虽然新主题的路由信息已经在 NameServer 存在了，但是只有一个 broker，并且不会再有更新。

以上。

那有没有方法解决这个问题呢？有！

方法一：

autoCreateTopicEnable 置为 false，所以生产环境需要用命令行工具手动创建Topic，可以用集群模式去创建（这样集群里面每个broker的queue的数量相同），也可以用单个broker模式去创建（这样每个broker的queue数量可以不一致）。

方法二：

连续快速地发送9条消息以上（单个 broker 的写队列默认是8）。

因为上面的关键点在于，当第一条消息发送出去之后，接收到消息的 Broker 便会在本地创建 topic，然后通过心跳机制同步到 NameServer，这个时间间隔最多只有30s。如果我们在最快时间内发送9条消息以上，那么消息就会被多个 broker 接收到，最后 NameServer 上的路由信息也将是多个 broker。

但这个方式太不可控，因此生产上我们还是使用方法一。

参考资料：

• RocketMQ 4.8.0 源码
• github.com/apache/rock…
• github.com/DillonDong/…
• 《RocketMQ 技术内幕》

最后

• 如果觉得有收获，三连支持下；
• 文章若有错误，欢迎评论留言指出，也欢迎转载，转载请注明出处；
• 文章源码 github 地址：github.com/CleverZiv/r… （带中文注释）
• 个人vx：Listener27, 交流技术、面试、学习资料、帮助一线互联网大厂内推等

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

「本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！」

题目如下：

给定一个字符串 s ，请你找出其中不含有重复字符的 最长子串 的长度。

题目要求找出给定字符串中不含重复字符的最长子串，我们可以采用暴力穷举的方式，得到字符串中的所有子串，然后一一判断不重复子串的长度，最后返回最长子串的长度即可，比如：

对于这样的一个字符串，我们首先从头开始进行遍历，将a取出：

然后取出下一个字符b，查看该字符是否重复，若不重复，继续放入新的字符串中：

下一个字符c也是如此：

紧接着下一个字符是a，此时发现新字符串中已经有了字符a，发生了重复，所以现在记录一下新字符串的长度，为3，然后从原字符串的第二个字符开始继续进行遍历：

再看下一个字符c，仍然放入新字符串：

直至遇到字符b，又产生了重复：

此时仍然记录当前新字符串的长度，并从原字符串的第三个字符开始遍历：

以此类推，就得到了一个无重复字符子串的长度表：

此时只需取出长度表中的最大值，即为字符串中无重复字符的最长子串长度。

清楚了算法思想之后，就可以写出代码：

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

java复制代码public static int lengthOfLongestSubstring(String s) {
        // 使用List集合来判断字符是否重复出现
        List<Character> list = new ArrayList<>();
        // 存储所有无重复子串的长度
        List<Integer> lenList = new ArrayList<>();
        // 记录子串长度
        int count = 0;
        char[] charArray = s.toCharArray();
        // 穷举所有子串
        for (int i = 0; i < charArray.length; ++i) {
            for (int j = i; j < charArray.length; ++j) {
                // 判断字符是否重复出现
                if (list.contains(charArray[j])) {
                    // 若出现，则记录当前子串的长度
                    lenList.add(count);
                    // 集合清空，count归0
                    list.clear();
                    count = 0;
                    // 结束本次循环
                    break;
                } else {
                    // 若未出现，则添加
                    list.add(charArray[j]);
                    count++;
                }
            }
        }
        // 对存放无重复子串长度的集合进行由大到小的排序
        lenList.sort((o1, o2) -> o2 - o1);
        // 排序后的首个元素即为集合中的最大值
        return lenList.get(0);
    }

将代码进行提交，结果出错：

原来我们没有考虑什么都不输入的情况，若无输入，则直接返回长度0即可，对于长度为1的输入，我们也得单独考虑，因为刚才的程序只有在出现重复字符的时候才会记录当前子串的长度，而如果输入字符串的长度为1，就没有重复的情况了，所以单独处理这两种情况即可，修改代码：

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

java复制代码public static int lengthOfLongestSubstring(String s) {
        // 若字符串长度为1，则直接返回1
        if (s.length() == 1) {
            return 1;
        }
        // 若无输入，则直接返回0
        if (s.length() == 0) {
            return 0;
        }
        // 使用List集合来判断字符是否重复出现
        List<Character> list = new ArrayList<>();
        // 存储所有无重复子串的长度
        List<Integer> lenList = new ArrayList<>();
        // 记录子串长度
        int count = 0;
        char[] charArray = s.toCharArray();
        for (int i = 0; i < charArray.length; ++i) {
            for (int j = i; j < charArray.length; ++j) {
                // 判断字符是否重复出现
                if (list.contains(charArray[j])) {
                    // 若出现，则记录当前子串的长度
                    lenList.add(count);
                    // 集合清空，count归0
                    list.clear();
                    count = 0;
                    // 结束本次循环
                    break;
                } else {
                    // 若未出现，则添加
                    list.add(charArray[j]);
                    count++;
                }
            }
        }
        // 对存放无重复子串长度的集合进行由大到小的排序
        lenList.sort((o1, o2) -> o2 - o1);
        // 排序后的首个元素即为集合中的最大值
        return lenList.get(0);
    }

测试通过：

暴力穷举虽然解决了题目需求，但执行效率非常低，为此，这里再介绍另外一种解决方案：滑动窗口。

对于这样的一个字符串：

我们设置一个滑动窗口，该窗口内的子串就是无重复字符的最长子串，定义两个指针用于划分窗口的左边界和右边界，并指定此时最长子串长度为1：

让right指针右移，扩大滑动窗口范围，此时最长子串长度为2：

继续右移right指针，最长子串长度为3：

当再次右移right指针时，发现字符a已经在滑动窗口中出现：

此时我们需要缩小滑动窗口，使其不再与当前字符a重复，让left指针右移：

当滑动窗口已不再与字符a重复后，扩大滑动窗口，right右移，此时最长子串长度仍为3：

此时又发现字符b与窗口中的字符重复，继续缩小滑动窗口：

无重复后，扩大滑动窗口，right指针右移：

以此类推，直到遍历结束。

代码如下：

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

java复制代码public static int lengthOfLongestSubstring(String s) {
        // 若无输入，则直接返回0
        if (s.length() == 0) {
            return 0;
        }
        // 定义滑动窗口左边界
        int left = 0;
        // 定义滑动窗口右边界
        int right = 0;
        // 无重复子串的最大长度
        int maxLen = 1;
        // 模拟滑动窗口
        Set<Character> set = new HashSet<>();
        // 遍历字符串
        while(right < s.length()){
            // 若滑动窗口中的字符与当前字符重复，则缩小滑动窗口，直至无重复为止
            while (set.contains(s.charAt(right))) {
                set.remove(s.charAt(left));
                left++;
            }
            // 计算当前滑动窗口长度，并与maxLen比较，取最大值作为新的无重复子串长度
            maxLen = Math.max(maxLen, right - left + 1);
            // 扩大滑动窗口
            set.add(s.charAt(right));
            right++;
        }
        return maxLen;
    }

测试通过：

该算法仍然有可以改进的地方，比如：

对于这样的一个字符串，当滑动窗口遇到重复字符：

此时缩小滑动窗口，left要一直右移，直至将字符w删除：

那么有没有办法能够让left直接移动到重复字符的下一个字符呢？我们可以改用HashMap来模拟滑动窗口，因为HashMap可以存储一个值，我们就让它存储字符的索引即可。
所以当遇到重复字符w时，直接从HashMap中取出滑动窗口中w的索引3，然后直接让left指针跳转至下一个索引4的位置即可。

代码如下：

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

java复制代码public static int lengthOfLongestSubstring(String s) {
        // 若无输入，则直接返回0
        if (s.length() == 0) {
            return 0;
        }
        // 定义滑动窗口左边界
        int left = 0;
        // 定义滑动窗口右边界
        int right = 0;
        // 无重复子串的最大长度
        int maxLen = 1;
        // 模拟滑动窗口
        Map<Character, Integer> map = new HashMap<>();
        // 遍历字符串
        while (right < s.length()) {
            // 若滑动窗口中的字符与当前字符重复，则缩小滑动窗口
            int index = map.getOrDefault(s.charAt(right), -1);
            // 直接让left指针跳转至滑动窗口中重复字符的下一个字符
            left = Math.max(left, index + 1);
            // 计算当前滑动窗口长度，并与maxLen比较，取最大值作为新的无重复子串长度
            maxLen = Math.max(maxLen, right - left + 1);
            // 扩大滑动窗口
            map.put(s.charAt(right), right);
            right++;
        }
        return maxLen;
    }

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

前言：Nodejs很火，但是好像工作中很少见到大面积的招聘岗位。我也用了nodejs写web服务很多年。分享一套我用了很久的代码，我觉得挺好。有需要的朋友拿去看看，我们不去纠结到底nodejs好不好，适不适合做Server，这些问题，吵翻天有何意义。能解决问题就好了。

开发快也是一种优势，现在中小型创业公司，底子差，业务要求快。加上云服务商的api加持，为何我们还要维护一个笨重的服务体系。当我需要AI服务的时候，我就接入AI服务商的接口就好了。我需要cdn服务的时候我接云服务就好了，我需要支付的时候我调用支付服务API就好了。那么一个小巧，快速的业务方案就会让创业型企业快速铺开市场，2-3个月就可以快速上线。尤其是在小程序盛行的今天，都不用笨重的APP了，云开发也是不错的选择，但是企业总要有一定的数据和业务能力，不然也会被云开发商掣肘。

今天分享的就是一套代码，会放到github上，github.com/vincent-li/… 大家下载还需要根据项目调整配置文件。不敢说是非常牛逼，我从来都是土狗一个，看看各位如果没有思路的朋友，是不是可以拿去直接先把业务干起来。废话不多说。代码走起。

思路比较简单待我一步一步拆解。我把nodejs定位就是web服务程序，不要跟我讨论底层稳定性的问题，因为我利用云服务商全部解决了。服务框架用的是koa，数据库是mongodb，那么我们来讲讲一个web服务到底需要什么。

• web服务最基本的是接收请求，查询数据库，返回结果给客户端。此类功能在controllers下面完成。
• 上传文件，利用一个插件multer，www.npmjs.com/package/mul…
• 静态文件服务，我会提到，但是还是不要写在服务器上面，讲过了，nodejs还是能力有限。利用cdn接口。
• 登录验证逻辑，利用session来做。有人说咋不用token，一个意思，只是放的位置不一样。
• log分析，利用守护进程pm2的log能力，具体后面讲。

讲了一堆估计没做过的人看懵了，不急，看看代码就明白了。入口index.js文件

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

php复制代码const Koa = require('koa')
const koaBodyParser = require('koa-bodyparser')
const koaViews = require('koa-views')
const koaStatic = require('koa-static')
const koaSession = require('koa-session')
const mongoStore = require('koa-session-mongo2')
const koaLogger = require('koa-logger')

const _config = require('../config')
const router = require('./controllers')

global._ = require('lodash')
global.model = require('./models')

const app = new Koa()

// 服务器运行log
app.use(koaLogger())

// 指定静态目录
app.use(koaStatic(_config.static))

// 指定views路径
app.use(
    koaViews(_config.views, {
        map: {
            html: 'lodash',
        },
    })
)

// 解析报文的body
app.use(
    koaBodyParser({
        enableTypes: ['json', 'form', 'text'],
    })
)

// session初始化
app.use(
    koaSession(
        {
            key: 'makefuture_sess',
            store: new mongoStore(_config.sessionURL),
            signed: false,
            // cookie过期时间，由浏览器负责到时清除，单位毫秒
            maxAge: 5 * 24 * 60 * 60 * 1000,
        },
        app
    )
)

// 使用router
app.use(router.routes(), router.allowedMethods())

console.log('启动端口：', _config.port)

app.listen(_config.port)

入口的js就是引入koa和一堆koa的插件，然后根据api装配起来就OK了。

由于项目都有各种环境，所以要根据启动环境把不同的配置项加载进来。const _config = require(‘../config’)，配置项中包括一些敏感的数据库连接账号和密码，我都替换掉了。所以各位在开发的时候，不建议把production的配置上传到gitlab，最好是运维持有并配置。

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

javascript复制代码const port = Number.parseInt(process.env.PORT) || 6060
const mongoUri =
    'mongodb://abcd:************@0.0.0.0:3717/?authSource=admin'
const root = '/Users/liwenqiang/codespace/git.ichoice.cc/makefuture/make'
const host = 'http://localhost:6060'
module.exports = {
    host,
    port,
    cdn: 'https://cdn.izelas.run',
    hostName: 'makefuture.app',
    home: root,
    static: `${root}/static`,
    views: `${root}/src/views`,
    mongoUri,
    dbName: 'makefuture',
    // 存放session的库和表配置
    sessionURL: {
        url: `${mongoUri}&poolSize=5&useUnifiedTopology=true&useNewUrlParser=true`,
        db: 'makefuture',
        collection: 'session',
        // 这里设置的是数据库session定期清除的时间，与cookie的过期时间应保持一致，
        // cookie由浏览器负责定时清除，需要注意的是索引一旦建立修改的时候需要删除旧的索引。
        // 此处的时间是秒为单位，cookie的maxAge是毫秒为单位
        maxAge: 24 * 60 * 60,
    }
}

利用global把一些通用的库加入到开发体系，就不用一遍一遍的引用了。这里把lodash引入，我觉lodash的方法足够我们用了。还有加密算法crypto，其实做的久了，基于业务的所有开发都是各种成熟组件的装配。解决方案都很成熟，所以我不认为有难做的业务，至少难不在技术，而是产品形态。

为啥把models作为global全局，因为这块是可以独立于koa初始化的，也没有必要纳入整个koa的请求上下文，但是业务处理的时候又要频繁使用，那么干脆初始化到global里面，随时取用。

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

ini复制代码const mongoose = require('mongoose')
const { mongoUri, dbName } = require('../../config')

mongoose.Promise = global.Promise

const conn = mongoose.createConnection(mongoUri, {
    dbName,
    useNewUrlParser: true,
    useUnifiedTopology: true,
})

const db = {}

const sysinfo = require('./_sysinfo')
const user = require('./_user')
const project = require('./_project')
const page = require('./_page')
const component = require('./_component')
const file = require('./_file')
const models = [sysinfo, user, project, page, component, file]

models.forEach((item) => {
    let newSchema = new mongoose.Schema(
        (typeof item.schema === 'function' && item.schema(mongoose.Schema)) ||
            item.schema,
        { collection: item.name }
    )
    db[item.name] = conn.model(item.name, newSchema)
})

module.exports = db

简单的不要不要的，就是把config里面配置的连接字符串，放到组件的方法里面，mongoose.createConnection，mongoose是用的比较好的mongodb连接组件，每中数据库都有相应的js帮助做操作。我比较喜欢mongodb是因为跟json结构完美契合，理解上比较一致，操作也基本符合js的操作规范。不用写sql真的舒服。随便找一个数据对象看下大家就明白了。

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

javascript复制代码const model = {
    name: 'user',
    schema: {
        nick_name: String, // 昵称
        phone: String, // 手机
        pass: String, // 密码
        avatar_url: String, // 头像
        sms_code: Number, // 短信登录码
        expire_at: Number, // 验证码过期时间
        create_at: Number, // 创建时间
    },
}

module.exports = model

name就是数据库中对应的表名，也是调用的时候的名字。跟文件名可以不一致。文件名我叫_user,name：user，调用的时候是model.user，非常方便。里面的字段名称和类型对应mongo的collection，也就是表。也可以跟对象和function，具体操作可以去看www.npmjs.com/package/mon… mongo的官方操作方法都支持，具体要看mongoose的版本和mongodb的版本，官方有的都可以用。创建完model就可以在方法中使用 await model.* 的方式调用了。方法是异步的，前面要加await，不喜欢可以直接调用同步方法。不建议，毕竟有语法支持，又不难理解。

回到index.js，后面的设置静态目录，设置模板，就比较简单，模板就是js的入口页面，为啥把js的入口放到服务，而不是做成静态放cdn，nodejs就是做静态渲染的，目的就是可以很好的把数据注入页面，所以js入口自己做方便很多。session这块要讲一下，这个图省事，大家都session概念不了解的麻烦自行阅读相关资料。这里的session模块干了几件事。

• 在koa ctx上下文中注入一个session对象，ctx.session = {}，可以把登录信息全部放进去。比如

1
2
3
4
5
6

ini复制代码 bo = {
    userid: u._id.toString(),
    nick_name: u.nick_name,
    avatar_url: u.avatar_url,
  }
  ctx.session = bo

• 自动把ctx.session中的数据存储到指定的数据库，这里就用的mongodb，这样做就不会因为服务宕机造成登录信息丢失。
• 根据配置，在session超时之后清除session信息。

所以我们只要做一个全局的过滤器，每次检测ctx.session中有没有登录的信息就好了。比如userid，userid为空就直接redirect到login页面就好了。

下面就剩一个就是处理请求了，这才是web服务的重头戏。我们利用bodyparse插件，将所有请求过滤一遍，把请求参数整理成json对象。

1
2
3
4
5
6

less复制代码// 解析报文的body
app.use(
    koaBodyParser({
        enableTypes: ['json', 'form', 'text'],
    })
)

然后就可以在ctx.request.query或者ctx.request.body中取用。ctx.request.query代表的是get请求的参数。ctx.request.body存post请求参数。当然也可以处理head，put，delete等请求。其实我觉得用不到。各位自己看。

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

php复制代码/**
 * 路由定义
 */
const Router = require('@koa/router')
const router = Router({
  prefix: '/api/user',
})

/**
 * 用户短信登录，朝用户手机中发送登录短息
 * @path - /api/user/smscode
 * @method - POST
 * @params
 *  phone - 用户手机号
 * @returns
 *  data - string | object 返回数据
 *  code - 0 || 500,
 *  success - true | false,
 *  message - ''
 */

router.post('/smscode', sendSmsCode)

以上就是一个路由定义的方法，prefix表示统一的前缀。post就是方法名，意思就是初始化了一个接口，访问路径就是/api/user/smscode，具体看代码，对应的执行方法就是sendSmsCode，这就是一个向用户端发送短信验证码的接口。当然短息我不用写，直接用云服务就好。我用阿里云的短信接口，发的66的。

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

javascript复制代码const sendSmsCode = async (ctx) => {
  const { phone } = ctx.request.body
  if (!phone) {
    ctx.body = getResponse(false, 'e501')
    return
  }
  if (!checkPhone(phone)) {
    ctx.body = getResponse(false, 'e552')
    return
  }
  let u = await model.user.findOne({ phone }).lean()
  // 是否已经发过
  if (u && u.sms_code && u.expire_at && u.expire_at > Date.now()) {
    ctx.body = getResponse(false, 'e553')
    return
  }
  // 获取阿里云access信息
  let aliyun = await model.sysinfo.findOne({ key: 'aliyun' }).lean()
  if (aliyun && aliyun.val.accessKeyId && aliyun.val.accessSecret) {
    const client = new Alicloud({
      accessKeyId: aliyun.val.accessKeyId,
      accessKeySecret: aliyun.val.accessSecret,
      endpoint: 'https://dysmsapi.aliyuncs.com',
      apiVersion: '2017-05-25',
    })
    // 获取随机的6个数字
    const code = +`${_.random(9)}${_.random(9)}${_.random(9)}${_.random(
      9
    )}${_.random(9)}${_.random(9)}`
    const params = {
      RegionId: 'cn-hangzhou',
      PhoneNumbers: phone,
      SignName: '码客未来',
      TemplateCode: 'SMS_203670207',
      TemplateParam: JSON.stringify({ code }),
    }
    const res = await client.request('SendSms', params, { method: 'POST' })
    if (res && res.Code === 'OK') {
      if (u) {
        await model.user.updateOne(
          { phone },
          { sms_code: code, expire_at: Date.now() + 15 * 60 * 1000 }
        )
      } else {
        await model.user.create({
          nick_name: '码客',
          phone,
          create_at: Date.now(),
          sms_code: code,
          expire_at: Date.now() + 15 * 60 * 1000,
        })
      }
      ctx.body = getResponse(true, '操作成功')
    } else {
      // fmtp('阿里云短息发送失败，原因：', res.Message)
      console.log('阿里云短息发送失败，原因：', res.Message)
      ctx.body = getResponse(false, 'e500')
    }
  } else {
    fmtp('aliyun --->', aliyun.val)
    ctx.body = getResponse(false, 'e500')
  }
  return
}

大家自行看代码吧，懒得讲太多，很多东西上面提到过。这个代码基本上是可以跑起来的，但是呢，我上传的代码是修改了关键信息的，如果希望接数据库就把自己的数据库参数输入。具体的看配置文件。当然希望你很会玩代码，自行排查一些问题。没那个闲功夫的就看看文章就好了，其实业务层，根据很多公司的业务发展不一样，需要不一样的代码，还是那句话，产品形态决定代码形态。

精髓就是，站着云厂商的肩膀上，快速把自己业务落地，AI类的我喜欢接百度，语音分析接科大讯飞，服务器、存储、云数据库、短信接阿里云，IM一般接腾讯，毕竟小程序场景还是微信生态稳。我的创业经验就是，小程序（微信）+Nodejs服务+阿里云企业服务+腾讯智能服务，麻溜的干业务，之所以没有成功，就是特么这个操蛋的创业环境。

各位看官仔细想想，在云服务商竞争激烈的今天，给了很多小而美的企业生机。以前要想做到的事情需要各种各样的人才，耗时耗力，关键还未必做的好。现在大部分都是调用现有的云服务，我们小企业就做好自己的业务，方便又快捷。而且我不用讨论什么压力，什么稳定。我把安全稳定的问题全部用分润的方式，利用云服务上的能力解决了。是！我啥底层都不懂，但不妨碍咱们成为一家企业的顶梁柱。未必比一些人差，很多专注于技术的人员，不但没有解决企业的问题，还增加了企业的负担，最后还是说企业这没有那没有。想想问题在哪？牢骚两句，看官自便。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

你好呀，我是Bella酱～

“又不是不能用，能用就行。”“又不是不能跑，能跑就行。程序和人有一个能跑就行。”

相信很多同学都听过这2句话。乍听没毛病。编程3部曲，“make it run，make it fast，make it better”。上述2句话，我将其归结为“make it run”这一个阶段，单看这一阶段，没有问题，但是若始终在这一阶段，那就有问题了。

什么是稳定性呢？我将其归类到“make it better”这个阶段。稳定性是通过一系列手段提前发现问题，力求将问题扼杀在襁褓中；稳定性要求在问题发生时，先于业务感知问题，处理问题，进而将问题影响面降至最小；稳定性要求问题发生后要有复盘，复盘哪里可以改进，以避免再次发生此类事情。简言之，稳定性就是为了保障系统正常运行，保障业务如丝般顺滑，保障数据准确无误。

稳定性并非只能够做一些零零散散的事情，稳定性也是有章可循的，有体系化的方法的。本文将从机制管控、监控告警、梳理系统风险点、保命措施、线上问题应急机制、故障演练、事后复盘、宣讲这8个方面来讲解业务团队如果在日常工作中做稳定性，可谓是涵盖了事前、事中、事后的方方面面。

1.机制管控

所有事情，能上系统就上系统，靠人来运转风险是极高的。这个道理，我相信很多同学都是明白的。稳定性保障亦是如此，通过一系列机制和系统强管控来规范日常工作中的一些行为。

下面这些都是可以参考的：

1）分支发布必须merge origin master分支的代码。

2）分支发布必须cr，且必须通过测试人员同意才可发布。开发、测试不可为同一人，除非测试认可开发自测即可。发布人员和cr人员不可为同一人。

3）分支发布需要在系统中有记录，内容至少包含发布时间、发布分支、发布内容、影响面、是否测试通过等等。

4）制定发布窗口，窗口外的时间如果需要发布，则需要走审批，审批人员可以是老板，也可以是团队内负责稳定性工作的人员，也可以视团队情况设置为其他的人员，例如团队核心开发等。制定发布窗口主要是为了防止晚上或周末发布的情况，避免发布出现问题时响应和处理不及时。

5）制定发布流程，主要是服务器环境这方面的，例如发布系统设置分支必须先在日常环境部署成功，才能部署预发；只有预发环境部署成功，才能部署线上；发布线上时，必须先在beta环境部署成功，且观察规定时间内，才能继续发布至线上；线上至少分几批来发布，每一批至少观察多久等等。

6）禁止流量一刀切，必须有逐步灰度的过程。通过发布机器的占比也好，通过业务中带灰度标的方式也好，必须要逐步灰度，禁止流量一次性全部切换。逐步灰度的过程是一个验证功能，暴露问题的过程，灰度范围是可控的，可以将灰度范围告知业务方，以便出现问题时，业务方知道哪里有了问题，而不是摸不着头脑。

7）数据修复必须通过工具来操作，工具要可获取当前操作人员，操作时间等，工具要具备check数据修复正确性的能力，工具的每次使用要在后台有记录，方便以后查看历史操作记录。

2.监控告警

监控告警的意义在于先于业务方发现问题，可以极大的提高响应速度，更快的开始定位问题。问题一旦定位到，影响面也就可以评估出来了，此时应该考虑如何快速止血，将影响面降至最低。

监控告警设置时有以下几点需要注意：

1）告警范围。切忌只告警给某个人，这样风险太大，例如当这个人在洗漱时就可能会错过告警信息。可以建一个钉钉告警群，把和这个业务相关的同学都拉进去，以及团队内负责稳定性的同学、老板等等。这样避免了“单点不可用造成服务宕机的情况”，即使一个人错过了告警信息，还有其他人，只要有人看到报警信息并处理就ok。

2）告警优先级。不同情况需要设置不同的告警优先级。举一个简单的例子，以服务成功率来说，持续3分钟成功率低于97%，可以告警至钉钉告警群；持续5分钟成功率低于97%，可以短信告警至订阅人；持续10分钟成功率低于97%，可以直接电话订阅人。

3）告警内容。这个需要视团队服务现状而定。一般来说，DB层面的监控告警、服务成功率层面的监控告警、服务处理失败数的监控告警、机器层面的监控告警等是必须的。DB监控告警例如CPU使用率、连接数、QPS、TPS、RT、磁盘使用率等。机器层面的监控告警例如CPU使用率、load、磁盘利用率等。服务成功率的告警配置时需要考虑持续时长。

4）告警有限性验证。这点是非常关键的，如果告警无效，那上述告警范围、优先级、内容这3个工作就是徒劳的。如果告警设置的阈值太高，那告警将起不到任何作用。如果告警设置的阈值太低，那每天都会接受到大量的告警，久而久之，就对这些告警疲倦了，等哪天狼真的来了，人们可能已经不相信了。经过压测，摸清了服务、DB、机器等的基线后，根据基线设置的告警是最可信的。若前期没有压测，可以先观察一下日常的水位，将告警阈值设置低一些，后面再根据告警情况慢慢调整告警的阈值。

3.梳理系统风险点

古人云，知己知彼，百战不殆。为何要知彼呢？知彼，你才知道他的弱点。做稳定性亦是如此。了解系统，你才会知道系统的风险点在哪里。当然，完全了解一个系统，是需要投入大量的时间和精力的。古人又云，君子性非异也，善假于物也。那梳理系统风险点的前期，你可以借一下你可爱的同事们的力呀～找到对应应用的owner，了解一下现状，以及潜在的风险点，以及是否有应对措施。

在梳理系统风险点时，需要关注以下几点：

1）链路是否闭环。这点是非常关键的，因为如果是系统层面的缺陷，那影响面一定是很大的。梳理链路是否闭环时，需要你跳出开发的思维，以审视一个产品的角度来考虑，这个产品是否可以在任何情况下正常的run起来。如果发现系统不闭环，一定要第一时间告知产品和老板。然后再从产品的角度来考虑如何解决这个问题，需要开发什么功能才ok，然后再将此作为高优先级的开发任务去进行排期修复。

2）慢SQL。慢SQL的威力是非常大的，一条慢SQL的执行可能会直接拖垮一个库，此时如果再有其他SQL请求执行，那其他SQL的执行耗时也会放大很多，这个时候，对于开发人员来说往往难以分辨哪些才是真正的慢SQL。此时需要静下心来根据时间点来慢慢查找真正的慢SQL，或者求助DBA同学，专业的同学做专业的事情，结果会更可靠，耗时也更短一些。如果是自己找到了慢SQL，最好和DBA同学求证一下自己查找的结果是否正确，以免错过了真正的慢SQL。

3）核心应用、非核心应用是否互相影响。不同重要等级的业务，应该在物理维度直接隔离。常说的读写分离也是这个道理，读写请求分别访问不同的机器组，以免互相影响。除此之外，还有DB维度的分离。

4.保命措施

机制管控、监控告警、梳理系统风险点都是为了防止问题的发生。可是，常在河边走哪有不湿鞋？如果线上发生了问题，要怎么快速止血呢？这个时候，就需要日常工作中准备好一些保命措施了，如果等到问题发生时，再去做准备，就有点太晚了。

那有哪些保命措施呢？

1）限流。虽然限流会导致一部分请求失败，但是他会减轻服务压力，减轻DB压力呀，在有些时候是可以用来保命的。常用的限流工具有Sentinel。日常工作中，可以先在Sentinel控制台配置好限流值，问题发生时，直接推限流开关即可。可以针对集群限流，也可以对单机配置限流，这个要看具体的业务场景和需求了。可以对所有应用来源进行限流，也可以对某些特定应用配置限流值，这个也是看具体的需求了。

2）降级。降级分为有损降级和无损降级。有损降级是业务感知的，无损降级只是技术侧的，例如查询从一个数据源降级为另外一个备用数据源，业务侧毫无感知。有损降级，比较典型的例子就是春晚抢红包活动，百度直接公告通知除夕当晚，百度云盘登录注册降级，以保障春晚抢红包活动顺利进行。对于有损降级来说，要在日常工作中就定义好，何种情况下，系统达到什么指标了，才可以进行有损降级。以免问题发生时，考虑过少，直接降级，导致更严重的问题发生。

3）切流。当一个机房的服务器发生了网络问题或硬件设施问题或其他导致机房不可用的问题时，可以切流至其他机房。当一个数据源不可用时，也可以切流至备用数据源。这些都可以减小问题的影响面，或解决问题。

4）布控。如果是和外部用户打交道的服务出现了问题，可以通过业务方或其他方式通知外部用户，告诉他们此时系统发生了问题，相关同学正在解决中，以安抚他们，而不至于让外部用户一脸懵逼，不知所措。一些官方号通过发微博的方式告知用户，也是布控的一种。

5）上下游、DBA等的联系方式。有时候可能并非自己系统的问题，而是上下游系统异常，间接导致自己系统的成功率下跌，这个时候，掌握上下游合作方同学的联系方式就很重要了，可以快速通知对方，让对方介入排查。有时候也可能是DB发生了问题，需要DBA协助解决才行，所以掌握DBA的联系方式也是非常重要的，DBA同学一个小小的命令可能就能拯救你于水火之中。

限流、降级、切流、布控等保命措施均需要清晰的定义系统哪些指标达到什么值时，才可以执行这些措施。这样当问题发生时，可以快速做出决策判断，也避免了使用不当造成更大问题。

5.线上问题应急机制

真的有线上问题发生了，怎么办？莫慌，越慌越乱。

首先，需要做到快速响应，表明已经有人在定位问题了。问题定位到之后，应该将如何快速止血放在第一位。此时就可以将上述的保命措施用上了。如果保命措施都用不到怎么办？只能采取其他措施了，通过发布解决也是一种方式。

一个人是无法很好的快速应急的。需要有通讯员和外部沟通，及时汇报当前进展；需要有处理人定位问题，评估影响面，给出建议；需要有决策人员，可能是团队内稳定性负责人，可能是老板、也可能是业务方等，决策如何快速止血。达成一致决策后，处理人按照决策执行即可，通讯员保持和外部及时的沟通。

6.故障演练

故障演练的目的是模拟故障发生时，大家的真实反应，以及是如何处理问题的。所以故障演练不要提前告知团队同学，更不要提前告知是何种线上问题，否则演练将毫无意义。当然，故障演练还应该把握好度，以免影响到线上业务。

7.事后复盘

线上问题发生后，无论大小，都应该有相应的复盘，小问题可以是小范围的复盘，大问题可以进一步扩大复盘参与人员的范围。

主要复盘哪些内容呢？

1）何时发现问题？

确定时间点

2）哪种方式发现的问题？监控告警还是业务方反馈？

以确定监控告警是否有效，是否有可优化的空间。

3）何时响应问题？

以确定响应是否及时，是否有可优化空间

4）何时定位到问题？

以评估对系统的熟悉度，或对线上问题的响应处理能力。有的同学可能是因为对系统不够了解，所以需要耗时很久才能够定位到问题；有的同学可能是因为遇到事情时，或高压下，心理素质不够，导致手忙脚乱，所以才耗时比平时更久。两种情况，可提升的能力是不同的。

5）有无快速止血措施？

例如限流、降级、切流等。可看平时保障工作是否到位。如果发现有缺失，则可以记为一个action，后面工作中把这一块给补上。最好的方式是，同步梳理下，系统中是否还有其他缺失的快速止血措施，并一起做掉。

6）需要上下游或DBA介入的情况，协同是否顺畅？

以评估协同沟通是否有可提升的空间，毕竟需要上下游或DBA介入的情况，如果可以快速联系到对应同学的话，是可以节省很多时间的，有助于更快速的解决问题。

8.宣讲

稳定性从来都不是一个人的事情，它关系到每一位同学，也是每一位同学应该铭记于心的事情。

在接需求时，需要考虑需求是否合理，链路是否闭环。

在写代码时，需要考虑代码的健壮性，语法使用是否正确，是否在写慢SQL。

在发布时，需要考虑测试是否通过，是否在发布窗口期，是否有灰度策略，发布时若发现问题应如何处理，是否可回滚等等。

在线上问题发生时，如何快速应急也是每一位同学都应该了解的。

一些宣讲还是很有必要的，以帮助每一位同学更好的了解自己应该如何做以保障系统的稳定。

好啦，我是Bella酱，一个在BAT写代码的妹子，欢迎关注我个人微信公众号（公众号：Bella的技术轮子）一起学习一起成长！今天的分享就到这里，我们下期见～

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

leoay技术圈的第8篇原创文章

字数: 4075

你好，我是 leoay, 这是我鸽了无数天之后开始写的又一篇文章，我保证，这是最后一次鸽……（不骗人！！！）

之前想了好久不知道要写啥，其实之前有一篇文章已经写了一半，但是觉得不好，就束之高阁了，本来想写一个从零开始的 Go 语言的系列文章，但是觉得没有必要，因为零基础的参考资料太多了。

最后，我决定还是写我当前用到最多，而且我认为也比较重要的一些东西吧，我觉得这样的话是一件一举多得的事情，性价比更高。

今天分享的这篇文章其实是我从前几天在公司内部分享的文章中截取的一部分，也是一个关于Go语言的很基础的介绍，正好适合这个号，所以就偷个懒，直接拿来了。

今天我想跟大家分享一些关于 Go 语言的一点知识，主要是为了说明一下“我为什么建议你学习一下 Go 语言”吧，我主要想从以下几个方面展开：

1. Go 语言的简单介绍
2. Go 语言的跨平台
3. Go 语言的网络编程
4. Go 语言的并发编程
5. 关于 Go 语言的学习资源分享

由于我学习 Go 语言，并在工作中运用也不是很长时间，零零碎碎大概一年吧，所以文章中一定有理解和表达不合适的地方，当然更会有不全面的地方，还请大家谅解，如果大家发现不对的地方，还请多多指教，当然我更希望我的文章能帮助到你，让 Go 语言的各种优势能在我们日常的工作中大放异彩，下面我们就进入正题。

Go 语言的简单介绍

Go 语言，也称 Golang, 最初是在2007年由Google的三位工程师 Robert Griesemer、Rob Pike 和 Ken Thompson 设计开发，他们的初衷是想用一种新的语言提高开发可靠、健壮和高效的软件。

刚开始，这仅仅是这个开发小组维护的一个小项目，但是后来慢慢觉得这个项目很有发展潜力，所以 2009年11月10日，Google 在开源博客上发布了这个项目，之后便有来自全球的开发者参与维护和开发。

Go 语言官网对这门语言的评价是:”Go 是一种开源编程语言，可以轻松构建简单、可靠和高效的软件”。从这里我们也能看出 Go 语言的一些特点， 简单、可靠、高效，除此之外，Go 语言还有很多其他特点，比如：

• 静态类型
• 清晰的语法（语法格式比较严格）
• 跨平台，本地交叉编译方便
• 编译效率高
• 运行时无依赖
• 支持垃圾回收
• 有丰富强大的标准库

怎么样，看到这么多优点大家是不是觉得Go语言简直就是“理想型”，其实Go语言的设计灵感主要来源于 Oberon、Pascal、C还有Alef等，也算是取各语言之长吧，集这么多优点于一身，其实Go语言也只是为了解决开发应用时在构建软件时遇到的复杂问题和构建大规模应用时的复杂性问题。

如果你使用过Go语言一段儿时间，一定会被它的简单和高效吸引，它不会让你纠结于语言本身的诸多特性，不像 C++ 一样给你铺设各种各样的坑，也不像 Java, Python，即使语法上像 Python 一样简洁，但是从硬件本身上来讲，它更像 C 语言，编译出应用就是可执行的二进制文件。

从编程范式上说，Go语言和C一样都是 “过程式语言”，Go 语言也想C 一样都是没有“对象”的“单身狗”，但是请不要以为Go就不支持面向对象编程了，相反，继承、多态以及基于对象，这些特征一点不少，可以说 Go 语言是一门骨子里有面向对象精神的过程式编程语言（来自突发奇想地胡扯）。

Go语言的跨平台

好了上面吹了那么多Go语言的优点，接下来我们也不说Go语言的缺点了，我们聊聊Go语言的跨平台。

为啥跨平台?因为编译器跨平台呗，像C和C++一样，Go的代码都是经过自己的编译器编译成机器能识别的二进制文件，然后直接运行。

这一点其实没啥好说的，我要说一点比较牛的地方，就是Go是自带交叉编译工具的。比如我在macOS上写了一段 go 代码，我想编译出能直接在 Windows上运行的程序怎么办呢？按照C语言的常规操作，就是先安装交叉编译工具链，修改Makefile 然后 make。

但是Go就不一样了，它只需要在编译命令前跟上几个参数就可以交叉编译了，下面列举编译不同平台的命令参数：

• 编译Linux程序

CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build filename.go

• 编译Arm Linux程序

CGO_ENABLED=0 GOOS=linux GOARCH=arm go build filename.go

• 编译Window程序

CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build filename.go

• 编译MacOS程序

CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build filename.go

上面的命令可以在任何平台上直接运行，就可以编译出目标平台的程序了，怎么样，是不是很方便？妈妈再也不用担心我们折腾工具链搞半天了，我们只需要专注于自己的业务代码了，Goooooo!

Go语言的网络编程

提到Go语言，我就不得不吹一下Go语言的网络编程了，我觉得像Go语言这样的网络编程极其简单高效的语言不多了，大家且用且珍惜。

为啥高效呢？我先写一段儿代码吧！

比如，我想用最原始的Go代码(不用任何第三方web框架)写一个web服务，我可以这么写

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

go复制代码package main
import (
    "io"
    "net/http"
    "log"
)
func HelloLeoay(w http.ResponseWriter, req *http.Request) {
    io.WriteString(w, "hello, leoay!\n")
}
func main() {
    http.HandleFunc("/hello", HelloLeoay)
    err := http.ListenAndServe(":6666", nil)
    if err != nil {
        log.Fatal("ListenAndServe: ", err)
    }
}

简单几行代码就能写一个，能接收 get 请求的web server, 如果使用更高效的第三方 web 框架，将更加如鱼得水，比如 gin、Iris、beego、gRPC(Google开源的RPC框架)、Kratos(B站开源的微服务框架)等等。

当然，Go语言发送get、post等网络请求也是非常方便的，这里就不再展开写了，大家感兴趣的可以去查一下。

Go语言的并发编程

下面，我想重点介绍一下Go语言的并发编程特性。虽然说 Go语言有众多优点，但是很大一部分人选择 Go语言作为自己的目标语言主要还是看重其在并发编程上的先天优势。

说到并发编程，其实主要包含三个方面：

（1）多线程编程

（2）多进程编程

（3）分布式编程

不过本篇文章就不展开那么多了，就简单谈一下Go语言最常用的多线程编程，其实我们日常的工作中最常用的并发编程也是多线程。

其实 go语言中的线程跟我们经常接触的 C/C++ 中的线程、java中的线程是不一样的，在Go语言中 其实是使用 goroutine 代表线程，但是goroutine是比线程更细粒度的，它允许我们以非常低的代价在同一个地址空间中并行地执行多个函数或者方法。相比于线程，它的创建和销毁的代价要小很多，并且它的调度是独立于线程的。在golang中创建一个goroutine非常简单，使用“go”关键字即可：

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

go复制代码package main

import ( "fmt" "time")

func learning(){ 
    fmt.Println("My first goroutine")
}

func main() { 
    //创建一个goroutine 执行 learning 函数
    go learning() 
    time.Sleep(1 * time.Second) fmt.Println("main function")
}

如上代码所示，我们只需要在函数之前加上go关键字，就创建了一个线程，而其他语言，如C/C++、java、python等与之相比则要复杂一些。

下面简单说一下 Go 语言线程间通信，这个还是比较有特点的，我们知道C/C++中的线程间通信是基于内存的，这个就容易导致很多难以定位的问题，其中最臭名昭著的就是“踩内存”。

但是，Go语言就没有这个烦恼了， 因为Go 是基于消息进行线程间通信的，Rob Pike有一句名言是这样说的：“Don’t communicate by sharing memory; share memory by communicating. (R. Pike)”。

什么意思呢？说白了就是说在 Go 语言中不要使用共享内存的方式进行通信，而要通过通信的方式共享内存，其实就是说明了 Go语言中进程间通信的方式。

那么还如何做呢？其实Go语言中有一个关键字 channel, 含义是通道，其实就是用于通信的，上面提到的进程间通信就可以通过这个channel实现，具体看代码：

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

go复制代码package main

import "fmt"

func sum(s []int, c chan int) {
 sum := 0
 for _, v := range s {
  sum += v
 }
 c <- sum 
}

func main() {
 s := []int{7, 2, 8, -9, 4, 0}

 c := make(chan int)
 go sum(s[:len(s)/2], c)
 go sum(s[len(s)/2:], c)
 x, y := <-c, <-c

 fmt.Println(x, y, x+y)
}

代码中有一个 channel 变量 c，就是用来传递两个sun进程内的消息的，就好像一根电话线一样连接而者，实现进程间通信。

怎么样，你觉得这样做是不是很巧妙，安全，而且还很简单。

关于 Go 语言的学习资源分享

最后，我就简单分享一些我学习Go语言时参考的资源吧，也希望能给想入坑的朋友一些帮助。

1. golang.google.cn/

最先推荐的当然是Go语言官网了，里面有各种入门小技巧以及非常详细的参考文档，从安装，到写第一行 Hello World, 再到写自己的模块，写一个web服务应有尽有，比较推荐的一本书就是里面的《Effective Go》，可以说是入门必备。

此外Packages 则是所有标准库的参考文档，遇到不熟悉的标准库，在这里面找答案就对了。

1. books.studygolang.com/gopl-zh/

第二本书就是《Go语言圣经》，也是一本很经典的入门书，总之学就对了

1. draveness.me/golang/

第三本是 draveness 的《Go 语言设计与实现》，这是 draveness 的个人博客上分享的书，写得很不错，无论文章排版，还是配图，以及内容深度，都很不错，推荐。

1. github.com/Terry-Mao/g…

第四个推荐一下B站的开源项目 goim， 就是用Go写的一个 IM (即时通信) 框架，你可知现如今B站那猛如洪流的弹幕可是基于这个框架开发的？至于其他优点，大家可以直接上 github 浏览。所以，不要犹豫的，赶紧学它就是了。当然，其中还涉及kafka、gRPC、Discovery等其他组件，所以完全学会还是不太容易的

1. github.com/go-kratos/k…

最后推荐一下 kratos 这个微服务框架，也是B站开源的，目前B站后台许多微服务都是基于这个框架开发，按照官方的话有如下优点：

• 简单：不过度设计，代码平实简单；
• 通用：通用业务开发所需要的基础库的功能；
• 高效：提高业务迭代的效率；
• 稳定：基础库可测试性高，覆盖率高，有线上实践安全可靠；
• 健壮：通过良好的基础库设计，减少错用；
• 高性能：性能高，但不特定为了性能做 hack 优化，引入 unsafe ；
• 扩展性：良好的接口设计，来扩展实现，或者通过新增基础库目录来扩展功能；
• 容错性：为失败设计，大量引入对 SRE 的理解，鲁棒性高；
• 工具链：包含大量工具链，比如 cache 代码生成，lint 工具等等；

好了，今天就简单分享这几点吧，因为是第一篇，所以没有写很深，更偏向于科普性质，不过我觉得可以帮助大家对Go语言有一个粗浅的认识，其实每一部分都可以详细展开的，这个等后面有时间再单独写吧。

Go 作为互联网时代的C语言，无论是在微服务，还是云原生都有着先天的优势，如果大家有 web 端或者高并发等业务需求，可以考虑一下如果用Go语言开发是不是更有优势。

好了，今天的分享就到这里，感谢大家阅读，如果有什么疑问也欢迎大家和我讨论交流。

上面就是摘录的文章，其实对于 Go 语言，我觉得它最大的有点就是语法简单，环境配置简单，代码比较简洁，不需要我们写很多没有用的代码，我觉得这也是大部分人很快喜欢上 Go 语言的原因。

但是，不管喜不喜欢，我们的目标是解决问题，然后用它产生价值，语言只是千万工具中的一种。

后面的文章，我会继续深入分享 Go 语言在实际项目中的各种应用，以及一些开源的小项目，欢迎大家关注，我们一起成长，一起加油吧！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

为什么要做签名？

想象一个场景：一位许久不见的朋友，突然在微信里面跟你说“朋友，借200应个急”，你会怎么反应？

我想大部分人马上的反应就是：是不是被盗号了？他是本人吗？

实际上这是我们日常生活中常见的通讯行为，系统间调用API和传输数据的过程无异于你和朋友间的微信沟通，所有处于开放环境的数据传输都是可以被截取，甚至被篡改的。因而数据传输存在着极大的危险，所以必须签名加密。

签名核心解决两个问题：

1. 请求是否合法：是否是我规定的那个人
2. 请求是否被篡改：是否被第三方劫持并篡改参数
3. 防止重复请求（防重放）：是否重复请求

如何进行签名

签名算法逻辑

第一步， 设所有发送或者接收到的数据为集合M，将集合M内非空参数值的参数按照参数名ASCII码从小到大排序（字典序），使用以下格式拼接成字符串stringA

1

erlang复制代码key1value1key2value2...

特别注意以下重要规则：

• 参数名ASCII码从小到大排序（字典序）；
• 如果参数的值为空不参与签名；
• 参数名区分大小写；
• 传送的sign参数不参与签名；

第二步，在stringA最后拼接上secret密钥得到stringSignTemp字符串

第三步，对stringSignTemp进行MD5加密得到signValue

防重放攻击

以上措施依然不是最严谨的，虽然仿冒者无法轻易模仿签名规则再生成一模一样的签名，可事实上，如果仿冒者监听并截取到了请求片段，然后把签名单独截取出来模仿正式请求方欺骗服务器进行重复请求，这也会造成安全问题，这攻击方式就叫重放攻击（replay 攻击）。

我们可以通过加入 timestamp + nonce 两个参数来控制请求有效性，防止重放攻击。

timestamp

请求端：timestamp由请求方生成，代表请求被发送的时间（需双方共用一套时间计数系统）随请求参数一并发出，并将 timestamp作为一个参数加入 sign 加密计算。

服务端：平台服务器接到请求后对比当前时间戳，设定不超过60s 即认为该请求正常，否则认为超时并不反馈结果（由于实际传输时间差的存在所以不可能无限缩小超时时间）。
但是这样仍然是仅仅不够的，仿冒者仍然有60秒的时间来模仿请求进行重放攻击。所以更进一步地，可以为sign 加上一个随机码（称之为盐值）这里我们定义为 nonce。

nonce

请求端：nonce 是由请求方生成的随机数（在规定的时间内保证有充足的随机数产生，即在60s 内产生的随机数重复的概率为0）也作为参数之一加入 sign 签名。

服务端：服务器接受到请求先判定 nonce 是否被请求过(一般会放到redis中)，如果发现 nonce 参数在规定时间是全新的则正常返回结果，反之，则判定是重放攻击。而由于以上2个参数也写入了签名当中，攻击方刻意增加或伪造 timestamp 和 nonce 企图逃过重放判定都会导致签名不通过而失败。

前端生成签名

一般在axios发送请求处统一拦截

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

javascript复制代码// npm install crypto-js
import MD5 from 'crypto-js/md5';

// 获取指定位数的随机数
function getRandom(num) {
  return Math.floor((Math.random() + Math.floor(Math.random() * 9 + 1)) * Math.pow(10, num - 1))
}

function genSign(params) {
  // 密钥
  const secret = 'xxxxxxxxxxxxxxxxxxxxxxx'
  // 1O位时间戳
  const timestampStr = parseInt(new Date().getTime() / 1000).toString()
  // 20位随机数
  const nonce = getRandom(20).toString()
  
  params.timestampStr = timestampStr
  params.nonce = nonce

  // 取 key
  const sortedKeys = []
  for (const key in params) {
    // 注意这里，要剔除掉 sign 参数本身
    if (key !== 'sign') {
      sortedKeys.push(key)
    }
  }
  // 参数名 ASCII 码从小到大排序（字典序）
  sortedKeys.sort()

  // 1 拼接参数
  let str = ''
  sortedKeys.forEach(key => {
    str += key + params[key]
  })
  // 2 拼接密钥
  str += secret
  // 3 MD5加密
  params.sign = MD5(str).toString().toUpperCase()
}

export default genSign

如何解决时间差问题

如果客户端时间与服务器时间不一致时(客户端时间比服务端快2分钟)， 如果验签时规定 一分钟内的请求有效，则该签名永远无法通过。

1. 第一次打开应用获取本地时间，然后请求接口获取服务器时间。

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

js复制代码  // 获取服务器时间
  function ajax(option){
      var xhr = null;
      if(window.XMLHttpRequest){
        xhr = new window.XMLHttpRequest();
      }else{ // ie
        xhr = new ActiveObject("Microsoft")
      }
      // 通过get的方式请求当前文件
      xhr.open("get","/");
      xhr.send(null);
      // 监听请求状态变化
      xhr.onreadystatechange = function(){
        var time = null,
            curDate = null;
        if(xhr.readyState===2){
          // 获取响应头里的时间戳
          time = xhr.getResponseHeader("Date");
          console.log(xhr.getAllResponseHeaders())
          curDate = new Date(time);
          console.log(curDate)
        }
      }
    }

2. 把时间差保存到本地存储
3. 请求接口的时候把本地时间和时间差相加。

后端生成签名

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

java复制代码import org.apache.commons.codec.binary.Hex;
import java.security.MessageDigest;

@Slf4j
public class SignUtil {
    // 密钥
    private static final String SECRET = "xxxxxxxxxxxxxxxxxxxxxxxxxx";
    private static final String SIGN = "sign";
    private static final String NONCE = "nonce";
    private static final String TIMESTAMP = "timestamp";
    private static final String SIGN_KEY = "apisign_";

    /**
     * 生成
     * @param params
     * @return
     */
    public static String genSign(TreeMap<String, Object> params) {
        params.remove(SIGN);
        StringBuilder str = new StringBuilder();
        for (String key : params.keySet()) {
            Object val = params.get(key);
            if (ObjectUtil.isNotNull(val)) {
                // 1 拼接参数
                str.append(key).append(val);
            }
        }
        // 2 拼接秘钥
        str.append(SECRET);
        // 3 MD5加密
        return md5(str.toString());
    }

    public static String md5(String source) {
        String md5Result = null;
        try {
            byte[] hash = org.apache.commons.codec.binary.StringUtils.getBytesUtf8(source);
            MessageDigest messageDigest = MessageDigest.getInstance("MD5");
            messageDigest.update(hash);
            hash = messageDigest.digest();
            md5Result = Hex.encodeHexString(hash);
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
        }
        return md5Result;
    }
}

验证签名

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

java复制代码  public static void validateSign(Map<String, Object> params) {
         // redis
        SingleRedisCacheClient cacheClient = ServiceBean.getSpringContext().getBean(SingleRedisCacheClient.class);

        String sign = (String) params.get(SIGN);
        if (StringUtils.isNotBlank(sign)) {
            if (StringUtils.isBlank(sign)) {
                throw new RuntimeException("签名不能为空");
            }

            String nonce = (String) params.get(NONCE);
            if (StringUtils.isBlank(nonce)) {
                throw new RuntimeException("随机字符串不能为空");
            }

            String timestampStr = (String) params.get(TIMESTAMP);
            if (StringUtils.isBlank(timestampStr)) {
                throw new RuntimeException("时间戳不能为空");
            }

            long timestamp = 0;
            try {
                timestamp = Long.parseLong(timestampStr);
            } catch (Exception e) {
                log.error("发生异常",e);
            }
            // 请求传过来的时间戳与服务器当前时间戳差值大于120，则当前请求的timestamp无效
            if (Math.abs(timestamp - System.currentTimeMillis() / 1000) > 120) {
                throw new RuntimeException("签名已过期");
            }
            
            // 请求传过来的随机数如果在redis中存在，则当前请求的nonce无效
            boolean nonceExists = cacheClient.hasKey(SIGN_KEY + timestampStr + nonce);
            if (nonceExists) {
                throw new RuntimeException("随机字符串已存在");
            }

            // 根据请求传过来的参数构造签名，如果和接口的签名不一致，则请求参数被篡改
            TreeMap<String, Object> signTreeMap = new TreeMap<>();
            signTreeMap.putAll(params);
            String currentSign = genSign(signTreeMap);
            if (!sign.equalsIgnoreCase(currentSign)) {
                throw new RuntimeException("签名不匹配");
            }

            // 存入redis
            cacheClient.setCacheWithExpire(SIGN_KEY + timestampStr+ nonce, nonce, 120L);
        }

    }

🚀🚀🚀🚀个人开源了一款基于React、TypeScript、Zustand、Ant Design开发的高颜值后台管理系统Slash Admin, 马上就有1000star了，感兴趣的可以了解下🚀🚀🚀🚀

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

「本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！」
首先请原谅楼主标题党（真香），步入正题：当入职3-5年后，相信你接口已经写得贼溜了。一天写个20个简单接口应该没啥问题。这时候一些老铁是不是认为咱以后也能干干架构。我认为你认为的很对。无论从未来发展与钱途上都是必需的。而那部分只会写业务的老铁在中年注定会被淘汰。

那么你又问了，公司不给我机会啊。也不让我转部门，然后因为没有经验，面试也不给我机会。

根据楼主多年教学（chui niu）经验，转到公司架构部门几率是非常小，因为一个小公司也就一个管架构的，你让他干啥去。那么只有投奔新公司才是正道。本文将从知识储备，丰满简历，迎接面试（hu you）等方面带你解决图中问题。

本文将提供准备的大纲，技术篇章后续会更新，如果有兴趣请关注楼主。

本文仅适合二线小企业开发人员，不可能雷同。温馨提示，千万不要知道就是掌握，了解就是精通。

准备工作

1. 确认自己未来偏向code，而不是管理。
2. 确保自己接口已经写的贼溜了。普通的业务代码已经难不到你了。
3. 不甘平凡，有充分的时间准备，充满饱满的热情持续学习。

如果准备好了，干就完事了。

知识储备

springboot高级接口功能实现

除了日常的接口外，springboot还有比普通接口复杂的功能，在日常中常用，在面试过程中讲述或许更有亮点。例如

1. springboot像内外网邮箱发送邮件。
2. websocket实现。（可以使用spring提供的，易上手，可二次开发）
3. springboot发送短信。
4. springboot整合swagger接口文档。
5. 上传文件到云。
6. 自定义注解监控日志。

此类功能可自行查看，而且功能都不是太难，是不是只写在简历上比curd更高级一点呢。

springboot架构级封装与理解

可能老铁对架构没有一个清晰的认识，对于小型公司架构，我有以下理解：

1. 对领导要求功能的实现。比如：认证与鉴权框架，在线阅读需求等。
2. 简化于小伙伴的开发，如使用@RestControllerAdvice简化异常的处理，自定义注解等。
3. 规范小伙伴的开发，如规范的命名，规范的项目结构，统一的返回对象封装等。

可能你又说了，这方面没有具体的了解，无从入手怎么办。这里提供了一个本文的中心思想！参考！读书人的事怎么能叫。。。遇到问题，第一时间百度或者去github或gitee寻找该轮子是否有人制造。

这里推荐两个开源项目：

1. jeecg:gitee.com/jeecg/jeecg…
2. ruoyi:gitee.com/y_project/R…

前者功能非常全，但是因为代码非一个人所写，格式比较混乱，建议仅学习功能。后者功能不如前者强大，但是规则非常符合日常开发，代码很有条理，小企业可以直接使用作为开发架构。

这回知道怎么入手架构了吧，如果以上两个框架你都参透了，在小企业定制款符合自己需求和开发习惯的架构是非常容易的。

了解源码

源码不是万能的，但是没有源码是万万不能的。在小公司，有这个技能可以说是可以吊打大部分初中级程序员，一张嘴，这个底层源码是怎么怎么写的。然后你会看到周围投来羡慕的眼光。

在源码部分你至少要掌握以下部分。

1. spring ioc与aop的源码。
2. spring bean的加载过程源码。
3. spring boot启动流程源码。
4. spring boot约定大于配置的实现源码。
5. mybatis执行流程源码。

当看到你简历上写阅读过源码，面试官是会嘿嘿一笑，对老板说这个小伙子还不错，仅次于我。切记，在面试过程中，不建议去背每个类名，给人一种我背我也行的感觉。总结性的描述一下原理即可。

准备一套微服务体系

在小厂，微服务可能在并发量或者业务并不是特别契合，但是可能甲方提出：我们就要上云，或者领导因为融资需要出门跟别人吹牛，所以大多数公司都需要此技能。

这里你需要掌握一套微服务体系，相比springcloud Netflix的闭源，建议使用如今火热的springcloud alibaba。

组件建议选择：

• nacos：注册和配置中心。
• openfeign：服务间调用组件。
• sentinel:熔断，降级，限流，完美支持openfeign。
• getway：网关，zuul已经是不是这个时代的选择了。

（虽然官方推荐dubbo，但是在小公司并发量并不大而且duboo相对复杂一些，所以使用openfeign也是一个不错的选择。）

最基础的要求就是自己能够搭建一套，了解其中基础概念，成功跑通。并背一些基础的面试题。

最好了解一下服务的注册于暴露的原理，这个面试遇到的概率很大。

下面这个划重点！！

在面试过程中，不要说自己没有实践过，而是自己做的demo。你可以说参与了公司部门微服务的搭建，这样即使不会了你也可以说这部分是他人搭建的，懂得吧。具体怎么组织语言还是看你自己。（以下所有技术栈都适用此条）

权限框架

在shiro和spring security中选择一项，建议使用spring security。

至少了解如何使用，核心配置类，和需要的表结构。

如果有经历可适当的了解oauth2。人才稀缺。

从0开发的话，必备技能。

sql优化

sql优化在面试中肯定跑不了，根据楼主多次面试，回答上explain具体分析流程的很少很少，掌握explain语法。会帮助你脱颖而出。

装x神器，干就完了

中间件

由于内卷的原因，现在不会几样中间件都不好意思出门。
redis与rabbitmq是必备技能，最少你需要掌握以下内容：

1. 与spring boot的整合。redis的增删改查，mq的发送与监控消息至少都应该demo一下。别问你用什么客户端都答不上。
2. mq消息准确发送的配置需要掌握，如：磁盘固化，ack,nack等。
3. redis的基本使用类型与使用场景，以及数据固化相关。
4. 其他基础面试题。

以下中间件可以初步了解，可以提升面试几率。

如elasticsearch，prometheus，apollo

对于中间件使用场景和并发量不高为什么使用，一定要给出合理的答案，这个是楼主经常问的。

规范

建议建立一套自己的开发规范，包括：

1. 代码编写规范
2. 接口编写规范
3. 建表规范
4. 接口文档编写规范
5. 注释规范
   可以参考阿里规范自行编写。目的如下：
6. 面试彰显自己的亮点和与众不同。
7. 为日后管理团队打下基础。
8. 建议日常应用，提高自己代码编写的水平。

虽然我们是小企业，也要专业。

设计模式

这里建议将工厂模式，单例模式，策略模式，代理模式，适配器模式，状态模式吃透，面试重灾区。并在工作中合理应用几次，作为面试答案。

linux操作

因为小企业人员的原因，可能运维开发都是一个人，所以适当的了解linux可以增加竞争性。

这里建议购买或借或使用公司服务器，将上文的项目与中间件在服务器上搭建一遍，并尽可能记住命令即可。

docker与非docker的方式选择一种即可。

idea插件（可选）

适当的使用idea插件，是日常开发中，凸显专业的方面。这里推荐几个插件。

1. easycode-制定后端的代码生成器。
2. eclipse code format -自定义代码规范。
3. p3c-阿里代码规范检测

知识输出（可选）

可以准备一个长时间维护的博客，可以让面试官更加了解你。

面试可以说，掘金优质博主，有原创文章xx篇，xx阅读量。咱上来就给面试官一波暴击。

服务性软件（可选）

有一些软件在日常开发中非常好用，也可能是某项工作中必须的，适当的描述可以坐实你的架构经验。如：

1. 选择一款公司文档工具。楼主使用的是dokuwiki。
2. 可以了解私服和镜像工具。楼主使用的是Nexus和harbor。
3. 可以选择一个测试平台。请自行百度开源开源。
4. 任务分配平台。请自行百度开源开源。
5. 一个符合自我习惯的接口平台，yapi是比较流程的，可适当选择。
6. processon-流程图在线制作网站。

其他开源项目（可选）

在楼主日常工作中，也遇到了很多没有接触的技术，楼主往往先看一遍官网文档，在通过开源项目学习，会使你事半功倍。以下项目可以适当了解。也可以在面试时挑选适当的引出，如果当前企业正有此方面需求，可能会增大。

kkeking/kkFileView

地址：gitee.com/kekingcn/fi…
一款在线阅读的开源项目，简单好用，可单独部署，支持的文件种类多样，且显示的文件格式较为美观，当项目周期短，人员紧张时且想落地改需求时，可以直接接入。

mingyang66/spring-parent

地址：github.com/mingyang66/…

一套spring security+oauth2为安全框架的架构，如果你想落地应用级的spring security+oauth2，此架构可以说是学习的好资料，同时提供 /redis、rabbitmq中间件的封装，值得学习和掌握。

Swagger文档转Word

文档地址：github.com/JMCuixy/swa…

废话不多说 能够将swagger转为word

zjm16/zjmzxfzhl

地址：gitee.com/zjm16/zjmzx…

一套spring boot+vue+flowable的工作流引擎，如果你需要以上技术栈，那么不会让你失望的。

xxl-job

地址：gitee.com/xuxueli0323…

XXL-JOB是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。

善于使用和解决问题，是小公司最缺少的人才。

总结

以上就是楼主的建议了，有关以上的技术性文章，后续楼主会陆续更新。有不明白的问题，可以留言，楼主必回。

如果本文章有一点用，还望看官姥爷用你们发财的小手点个赞和关注。圆我百赞的梦想。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

「本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！」

相关文章

MyBatis系列汇总：MyBatis系列

前言

• MyBatis 的配置文件包含了会深深影响 MyBatis 行为的设置和属性信息。
• 下面我们会跟着官方文档配置实际代码来走一遍必须要掌握的配置功能和一些必须了解的配置功能！
• 官方文档
• 前置条件请看前面的几篇文章，本章所有内容的演示都是基于前面文章的基础之上。
• 重点：

+ ![image-20210706100245597.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/7cfcd513638327394ccec4b026e947f7369d18b82638ac82a13ed087db85c3d2)

一、环境配置（ environments）

• MyBatis 可以配置成适应多种环境
• 不过要记住：尽管可以配置多个环境，但每个 SqlSessionFactory 实例只能选择一种环境。
• Mybatis 默认的事务管理器是JDBC，连接池：POOLED
• 多环境配置（实际开发中都是以这种方式来的）

+ ContextAplication.xml：


    - 
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
java复制代码<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
        PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>

    <!--引入外部配置文件,这里文件名称要一样，properties文件一般我们放在resources目录下 -->
    <properties resource="application.properties">
    </properties>

    <environments default="prod">
        <!--设置默认的环境为开发环境  -->
        <environment id="development">
            <transactionManager type="JDBC"/>
            <dataSource type="POOLED">
                <property name="driver" value="${dev.driver}"/>
                <property name="url" value="${dev.url}"/>
                <property name="username" value="${dev.username}"/>
                <property name="password" value="${dev.password}"/>
            </dataSource>
        </environment>
        <!--测试环境用  -->
        <environment id="pre">
            <transactionManager type="JDBC"/>
            <dataSource type="POOLED">
                <property name="driver" value="${pre.driver}"/>
                <property name="url" value="${pre.url}"/>
                <property name="username" value="${pre.username}"/>
                <property name="password" value="${pre.password}"/>
            </dataSource>
        </environment>
        <!--生产环境用  -->
        <environment id="prod">
            <transactionManager type="JDBC"/>
            <dataSource type="POOLED">
                <property name="driver" value="${prod.driver}"/>
                <property name="url" value="${prod.url}"/>
                <property name="username" value="${prod.username}"/>
                <property name="password" value="${prod.password}"/>
            </dataSource>
        </environment>
    </environments>
    <mappers>
        <mapper resource="com/dy/mapper/UserMapper.xml"/>
    </mappers>
</configuration>


+ application.properties：


    - 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码dev.driver = com.mysql.jdbc.Driver
dev.url = jdbc:mysql://IP地址:3306/master?useSSL=false&amp;jeuc_2_1?useUnicode=true&amp;characterEncoding=utf-8&amp;autoReconnect=true&amp;failOverReadOnly=false&amp;testOnBorrow=true
dev.username = root
dev.password = 123456

pre.driver = com.mysql.jdbc.Driver
pre.url = jdbc:mysql://IP地址:3306/master?useSSL=false&amp;jeuc_2_1?useUnicode=true&amp;characterEncoding=utf-8&amp;autoReconnect=true&amp;failOverReadOnly=false&amp;testOnBorrow=true
pre.username = root
pre.password = 123456

prod.driver = com.mysql.jdbc.Driver
prod.url = jdbc:mysql://IP地址:3306/master?useSSL=false&amp;jeuc_2_1?useUnicode=true&amp;characterEncoding=utf-8&amp;autoReconnect=true&amp;failOverReadOnly=false&amp;testOnBorrow=true
prod.username = root
prod.password = 123456


    - 这个也是根据实际开发的需求来的，一般我们都是三套环境


        * dev 开发环境
        * pre 上线前转测环境
        * pro 线上环境
+ 整体目录结构如图：


    - ![image-20210706141920964.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/e6a8dc546a24c9a147d9a74f746070ad921426b67895253945c6e7a62231e4a1)
+ 这种方式可以让我们灵活的变换不同的配置文件！十分方便和简洁！
+ 这里的 ${url} `动态属性替换`下面详解

二、属性（properties）

①、外部动态替换

• 先看下本来的配置文件

+ ![image-20210706110700351.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/158535f043bee75c951e6bafcffdf1050ce1864a85f288c774886c0db385b081)

• 实际开发中我们可能是多个环境，如：dev（开发环境）、pre（准测环境）、prod（线上环境）

+ 这可能会导致我们需要配置多个ContextAplication.xml核心配置文件，这种做法不可取。
+ Mybatis提供了动态配置替换的功能。

• 使用如下：

+ 新建配置文件：application.properties
+ 
1
2
3
4
java复制代码driver = com.mysql.jdbc.Driver
url = jdbc:mysql://IP地址:3306/master?useSSL=false&jeuc_2_1?useUnicode=true&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false&testOnBorrow=true
username = root
password = 123456


+ ![image-20210706111926683.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/524dc3bd2822ebd143caaf6ebbb6bf54205b598716213fa12dd097b81e6067c9)
+ ContextAplication.xml文件修改
+ 
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
java复制代码<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
        PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>

    <!--引入外部配置文件-->
    <properties resource="application.properties">
    </properties>

    <environments default="development">
        <environment id="development">
            <transactionManager type="JDBC"/>
            <dataSource type="POOLED">
                <property name="driver" value="${driver}"/>
                <property name="url" value="${url}"/>
                <property name="username" value="${username}"/>
                <property name="password" value="${password}"/>
            </dataSource>
        </environment>
    </environments>
    <mappers>
        <mapper resource="com/dy/mapper/UserMapper.xml"/>
    </mappers>
</configuration>


+ 对应关系：
+ ![image-20210706112336461.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/3a71556cbafb73ca7a83b019531123487cbaed6b55a36f6b84a8780754a755e9)
+ 好处：


    - 多环境配置可以ContextAplication.xml无需修改
    - 只需要配置application.properties文件，在实际开发中我们也是这样做的！
    - 方便测试和开发！
+ 执行结果成功：


    - ![image-20210706112527688.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/97e74aca67eb13ab7f13a5cb1458d20318429b953fe5a2d8b6e769c0516fa0ba)#### ②、内部动态替换


+ 内部动态替换效果是一样的，但是没有外部配置文件的方式灵活！建议使用外部配置文件来进行灵活动态替换！
    - ![image-20210706112614626.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/4e787ac9c36373e6ca4028adf1e3d0956aa75dfc47f65146531ca45c314680da)
+ 如果两种同时有呢？
    - ![image-20210706113812629.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/66a182acd77712ba90e64a6aac9502b8a78171b6b2d9b5eab7ed20a24ec8f61f)
    - 内部密码瞎写是错误的，执行查看结果正常
    - 结论：优先走外面的properties文件
    - ![image-20210706113824879.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/b9bc328806e1eab256f830fd05d34628131b9f689b32f0a850759b9fe1136a95)#### ③、方法中传入替换


+ SqlSessionFactoryBuilder.build() 方法中传入属性值
+ 我们可以先看下build这个方法的代码：


    - ![image-20210706114656985.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/02b2858f39d6a3570473005f78c932ee844188dd4a1686036cc2991a78c28fef)
    - ![image-20210706114737808.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/00447380f9f87dc348eb90e13adf1f310a7487364b56e69435fc097a9c80d920)
    - 可以看出来，可以传入inputStrem、envirnoment、properties等参数值。


        * inputStrem：字节流，这个没啥好说的，创建SqlSessionFactory必传的东西
        * envirnoment：指定的环境，跟\*\* \*\*保持中的id一致
        * properties：properties配置文件的设置
    - 具体案例如下：


        * 
1
2
3
4
5
6
7
8
java复制代码Properties properties=new Properties();
           //用的是磁盘符的绝对路径
           InputStream input = new BufferedInputStream(new FileInputStream("D:\\workSpace\\dyj-MyBatis-Project\\MyBatis-01-properties\\src\\main\\resources\\application.properties"));
           //加载到properties中
           properties.load(input);
           String resource = "ContextAplication.xml";
           InputStream inputStream = Resources.getResourceAsStream(resource);
           sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream,"prod",properties);


        * 打个断点瞅瞅：可以看见，properties文件成功被读取，所有数值都一样！
        * ![image-20210706142932738.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/6d89237d28f5dc264861011a4a63ae4846a3c7227b3233156675a04c6b2a8f2a)

• 如果三种动态替换同时存在呢？顺序如下：

+ 首先读取在 properties 元素体内指定的属性。
+ 然后根据 properties 元素中的 resource 属性读取类路径下属性文件，或根据 url 属性指定的路径读取属性文件，并覆盖之前读取过的同名属性。
+ 最后读取作为方法参数传递的属性，并覆盖之前读取过的同名属性。

④、占位符

• 从 MyBatis 3.4.2 开始，可以为占位符指定一个默认值。
• 前提条件：这个特性默认是关闭的。要启用这个特性，需要添加一个特定的属性来开启这个特性：

+ 
1
2
3
4
java复制代码<properties resource="application.properties">
        <!-- 启用默认值特性 -->
        <property name="org.apache.ibatis.parsing.PropertyParser.enable-default-value" value="true"/>
    </properties>

• 语法格式如下：

+ 
1
java复制代码 <!-- 如果属性 'username' 没有被配置，'username' 属性的值将为 '123456' -->


+ 
1
2
3
4
5
6
7
8
9
java复制代码        <environment id="prod">
            <transactionManager type="JDBC"/>
            <dataSource type="POOLED">
                <property name="driver" value="${prod.driver}"/>
                <property name="url" value="${prod.url}"/>
                <property name="username" value="${prod.username}"/>
                <property name="password" value="${prod.password:123456}"/>
            </dataSource>
        </environment>


+ 测试看看结果：


    - 去除所有关于密码的设置
    - ![image-20210706144045168.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/b9a2daba6eef626ba8caada03995eb1e3c925f69be7a99658ab8183893edca8e)
    - 执行结果如下：成功！
        * ![image-20210706144011705.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/0b4afe6c0188125fb545c8bd6ed249ac74869d83dc6d6abb703e04ea6e8b689a)

三、设置（settings）

• 此内容不作重点讲解，了解基本概念即可！
• 这是 MyBatis 中极为重要的调整设置，它们会改变 MyBatis 的运行时行为。 下表描述了设置中各项设置的含义、默认值等。

四、别名（typeAliases）

①、typeAlias

• 语法格式如下：

+ 
1
2
3
java复制代码    <typeAliases>
        <typeAlias type="com.dy.pojo.User" alias="hello"></typeAlias>
    </typeAliases>


+ type：实体类的具体位置
+ alias：设置别名

• mapper.xml

+ 
1
2
3
java复制代码<select id="getUserInfo" resultType="hello">
   select * from user
</select>


+ resultType：写上面设置的别名

• 执行结果：
• 
• 使用场景：

+ 实体类比较少的时候使用`typeAlias`
+ 当这样配置时，`hello` 可以用在任何使用 `hello` 的地方。

②、package

• 语法格式如下：

+ 
1
2
3
java复制代码    <typeAliases>
        <package name="com.dy.pojo"/>
    </typeAliases>


+ package：只需要指定到实体类所在目录即可，该包下的所有实体类名称的名字即是别名

• mapper.xml

+ 
1
2
3
java复制代码    <select id="getUserInfo" resultType="User">
        select * from user
    </select>


+ resultType：直接写实体类的名称即可

• 执行结果
• 
• 使用场景：

+ 实体类多使用 `package`
+ 第一种可以自定义，第二则不行，本身的名称即是它的`别名`，不区分大小写！

③、注解

• 使用注解为实体类标明别名
• 这个需要搭配着package来一起使用，当没有注解起别名时，本身名称就是别名
• 当有@Alias(“dayu”)时，dayu就是它的别名
• 语法格式如下：

+ 
1
java复制代码@Alias("dayu")

• mapper.xml

+ 
1
2
3
java复制代码    <select id="getUserInfo" resultType="dayu">
        select * from user
    </select>


+ resultType：别名

• 执行结果如下：
• 
• 使用场景：

+ 我们一般搭配着 package 来一起使用

• 下面是一些为常见的 Java 类型内建的类型别名。它们都是不区分大小写的，注意，为了应对原始类型的命名重复，采取了特殊的命名风格。

• 别名	映射的类型
  _byte	byte
  _long	long
  _short	short
  _int	int
  _integer	int
  _double	double
  _float	float
  _boolean	boolean
  string	String
  byte	Byte
  long	Long
  short	Short
  int	Integer
  integer	Integer
  double	Double
  float	Float
  boolean	Boolean
  date	Date
  decimal	BigDecimal
  bigdecimal	BigDecimal
  object	Object
  map	Map
  hashmap	HashMap
  list	List
  arraylist	ArrayList
  collection	Collection
  iterator	Iterator

五、映射器（mappers）

• 既然 MyBatis 的行为已经由上述元素配置完了，我们现在就要来定义 SQL 映射语句了。 但首先，我们需要告诉 MyBatis 到哪里去找到这些语句。 在自动查找资源方面，Java 并没有提供一个很好的解决方案，所以最好的办法是直接告诉 MyBatis 到哪里去找映射文件。

①、resource（使用相对于类路径的资源引用）

+ 语法格式如下：
+ 
1
2
3
xml复制代码<mappers>
    <mapper resource="com/dy/mapper/UserMapper.xml"/>
</mappers>


+ 推荐使用，比较稳定，省心！
+ 结果正常：
+ ![image-20210706152823904.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/52e1d71f005647e9b0dab089a3bb8de2bd57ea67c2c8f82f27333c3c5c749219)#### ②、url（使用完全限定资源定位符）


+ 语法格式如下：
+ 
1
2
3
xml复制代码    <mappers>
        <mapper url="file:D:\workSpace\dyj-MyBatis-Project\MyBatis-01-properties\src\main\java\com\dy\mapper\UserMapper.xml"/>
    </mappers>


+ 通俗来讲就是文件位置绝对路径
+ 不推荐使用！换个环境可能就得改动！比较麻烦！
+ 结果正常：
+ ![image-20210706152823904.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/52e1d71f005647e9b0dab089a3bb8de2bd57ea67c2c8f82f27333c3c5c749219)#### ③、class（使用映射器接口实现类的完全限定类名）


+ 语法格式如下：
+ 
1
2
3
xml复制代码<mappers>
    <mapper class="com.dy.mapper.UserMapper"/>
</mappers>


+ 不推荐使用，还是有区别的，本来指定的是xml，这个指定的mapper文件
+ 接口和它的Mapper必须同名
+ 接口和他的Mapper必须在同一包下
+ 否则报错
+ ![image-20210706160108165.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/6b0a099f833de1f64e65a740cc14dc3694f9c16712dda71ac2a599d08d225e60)
+ 改正后结果正常：
+ ![image-20210706152823904.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/52e1d71f005647e9b0dab089a3bb8de2bd57ea67c2c8f82f27333c3c5c749219)#### ④、name（将包内的映射器接口实现全部注册为映射器）


+ 语法格式如下：
+ 
1
2
3
xml复制代码    <mappers>
        <package name="com.dy.mapper"/>
    </mappers>


+ 不推荐使用
+ 接口和它的Mapper必须同名
+ 接口和他的Mapper必须在同一包下
+ 这个是指定到包目录下
+ 结果正常：
+ ![image-20210706152823904.png](https://gitee.com/songjianzaina/juejin_p14/raw/master/img/52e1d71f005647e9b0dab089a3bb8de2bd57ea67c2c8f82f27333c3c5c749219)

六、事务管理器（transactionManager）

• 其实这个属于environments中的内容，但是我觉得有必要拿出来单独讲一讲！
• 这个内容我觉得了解即可，我们大部份情况下用的都是第一种JDBC
• 但是！我们要知道有MANAGED 这么个玩意存在！（面试的时候可能会被问到哦~）
• 在 MyBatis 中有两种类型的事务管理器（也就是 type=”[JDBC|MANAGED]”）：

+ JDBC – 这个配置直接使用了 JDBC 的提交和回滚设施，它依赖从数据源获得的连接来管理事务作用域。
+ MANAGED – 这个配置几乎没做什么。它从不提交或回滚一个连接，而是让容器来管理事务的整个生命周期（比如 JEE 应用服务器的上下文）。 默认情况下它会关闭连接。然而一些容器并不希望连接被关闭，因此需要将 closeConnection 属性设置为 false 来阻止默认的关闭行为。
+ 
1
2
3
xml复制代码<transactionManager type="MANAGED">
  <property name="closeConnection" value="false"/>
</transactionManager>

• 如果你正在使用 Spring + MyBatis，则没有必要配置事务管理器，因为 Spring 模块会使用自带的管理器来覆盖前面的配置。

七、配置的顺序（补充知识点）

• 给大家分享一个好玩的东西，为啥不写在开头呢？因为只有自己试过才好玩呀！哈哈哈哈啊！
• 将mapper的映射位置换到上面去：
• 
• 报错，看提示：
• 
• 总结顺序如下：

+ properties
+ settings
+ typeAliases
+ typeHandlers
+ objectFactory
+ objectWrapperFactory
+ reflectorFactory
+ plugins
+ environments
+ databaseIdProvider
+ mappers

• 必须按照MyBatis规定的顺序来哦~否则会报错的哟！

路漫漫其修远兮，吾必将上下求索~

如果你认为i博主写的不错！写作不易，请点赞、关注、评论给博主一个鼓励吧~hahah

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

前言

多线程系列我们前面已经更新过多个章节，强烈建议小伙伴按照顺序学习：

《蹲坑也能进大厂》多线程系列文章目录****

CountDownLatch是JUC中常见的一种工具类，从类名中我们也能略窥一二这个类的含义，CountDown字面意思为倒数，latch有门闩（shuan）的意思，主要是用来控制并发流程，本文将详细介绍它的场景和用法。

一、CountDownLatch两大适用场景

• 控制子流程结束：一个线程等待多个线程执行完毕

经常网购的小伙伴对拼夕夕肯定不陌生吧，而拼夕夕有个特点，只有在商品拼团成功后，商家才会发货，如果一直没能成功，那你的订单会一直处于等待的状态；此外，游乐园里的过山车，一次可以坐10个人，商家为了节约成本，一般都是等待人满才会发车，这些都是一等多的生活场景。

• 控制子流程开始：多个线程等待一个线程执行完毕

这种场景也是非常常见，比如我们以前参加运动会时，100米短跑时都会先抬起优秀的翘臀，等待裁判员发出指令后才能起跑，只要裁判不发指令，所有运动员都要进行等待。此外还有一种场景离我们也是十分近—压测，压测也是为了模拟多个请求同时访问某个接口，预测出其支持最大压力程度。不知道有没有小伙伴遇到服务上线后，因为流量激增服务被干掉的情况，哎，不说了，都是泪。

二、CountDownLatch常见方法

CountDownLatch主要是通过AQS的共享锁机制实现的，内部逻辑比较简单，它的核心属性只有一个sync，这个后续会AQS会详细介绍

1

java复制代码private final Sync sync;

现在我们只要掌握以下4个属性在两种场景下的作用即可：

• CountDownLatch(int count)： 类中唯一的构造器，count为倒数的次数；
• await()： 调用该方法的线程会被挂起，直到count为0时才会被唤醒；
• countDown()： 将count执行减一操作，当count为0时，等待中的线程会被唤醒；
• getCount()： 获取latch数量。

三、CountDownLatch代码演示

• 一等多场景

我们开发软件的目的是上线供用户使用，那再上线前肯定需要对系统进行全面测试，而很多模块间并没有依赖关系，因此可以同时进行测试，等待所以模块测试完毕，才能进行正式上线。

本例中假设有五个模块需要测试，并且模块测试可以同步进行，当每个模块测试完毕后会调用countDown方法，执行减一操作，等到count为0时，调用了await()方法的主线程从挂起状态被激活，继续执行。

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

java复制代码public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(5);
    ExecutorService service = Executors.newFixedThreadPool(5);
    //创建5个线程，分别执行不通模块测试
    for (int i = 1; i <= 5; i++) {
        final int num = i;
        Runnable runnable = () -> {
            try {
                //模拟测试耗时
                Thread.sleep((long) (Math.random()*5000));
                System.out.println("模块："+ num +"测试完毕");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                //测试完毕后执行减一操作
                latch.countDown();
            }
        };
        service.submit(runnable);
    }
    System.out.println("等待模块检测...");
    //主线程等待，直到count为0
    latch.await();
    System.out.println("所有模块测试完毕,开始上线");
}

运行结果，当所有模块检测完毕后才开始上线：

1
2
3
4
5
6
7

java复制代码等待模块检测...
模块：2测试完毕
模块：5测试完毕
模块：4测试完毕
模块：3测试完毕
模块：1测试完毕
所有模块测试完毕,开始上线

如果小伙伴还是觉得很绕，这里画个图就基本明白了。

• 初始化count为5；
• 主线程执行await()进入等待状态；
• 各个模块调用countDown()方法对count减一，并且线程不会进入休眠，而是继续执行自身逻辑；
• 当count变成 0 后，主线程被唤醒。

• 多等一操作

这里使用运动会赛跑的栗子：在运动会上，所有运动员会先进行准备动作，只有等到裁判员发出信号，所有人才能开始比赛。

与之前的例子主要区别在于：CountDownLatch初始变量count为1，然后所有线程调用await方法进行等待，直到主线程调用countDown方法将count值减为1，此时所有线程同时被唤醒，开始执行自身逻辑。

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

java复制代码public static void main(String[] args) throws InterruptedException {
    //重点：这里初始化为1
    CountDownLatch begin = new CountDownLatch(1);
    
    ExecutorService service = Executors.newFixedThreadPool(5);
    //创建5个线程
    for (int i = 1; i <= 5; i++) {
        final int num = i;
        Runnable runnable = () -> {
            try {
                System.out.println("运动员："+ num +"等待起跑指令");
                //所有运动员等待指令
                begin.await();
                System.out.println("运动员："+ num +"开始跑步");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        service.submit(runnable);
    }
    Thread.sleep(1000);
    //裁判员发出起步指令
    begin.countDown();
    System.out.println("发令完毕");
}

运行结果，当发令法币后所有运动员才开始起跑：

1
2
3
4
5
6
7
8
9
10
11

java复制代码运动员：1等待起跑指令
运动员：3等待起跑指令
运动员：4等待起跑指令
运动员：2等待起跑指令
运动员：5等待起跑指令
发令完毕
运动员：3开始跑步
运动员：4开始跑步
运动员：1开始跑步
运动员：5开始跑步
运动员：2开始跑步

总结

以上就是CountDownLatch的全部内容，相对来说比较容易掌握，理解难度不高，结合具体应用场景选择两种方式来使用，今天的文章就到这里咯，下一期唠唠其他好用的工具类吧，好好敲一敲，轻松搞定并发编程。

点关注，防走丢

以上就是本期全部内容，如有纰漏之处，请留言指教，非常感谢。我是花GieGie ，有问题大家随时留言讨论 ，我们下期见🦮。

文章持续更新，可以微信搜一搜 花哥编程 第一时间阅读，有兴趣的小伙伴欢迎关注，一起学习，共同进步。

原创不易，你怎忍心白嫖，如果你觉得这篇文章对你有点用的话，感谢老铁为本文点个赞、评论或转发一下，因为这将是我输出更多优质文章的动力，感谢！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有