发表于

欢迎来到操作系统系列，采用图解 + 大白话的形式来讲解，让小白也能看懂，帮助大家快速科普入门。

本篇文章开始探秘用户态与内核态，虽然一般面试不会问这个，但搞清楚这块，对我们理解整个计算机系统是及其有意义的，这会让你在今后的学习中豁然开朗，你肯定会发出：“啊，原来如此的感叹！”

内容大纲

小故事

张三是某科技公司的初级Java开发工程师（低权限），目前在15楼办公码代码，公司提供的资源仅有一套电脑（用户态），张三想着这一线的房价，倍感压力山大，于是给自己定下一个目标，一定要做技术总监，在一线扎根，
奋斗B张三，奋斗5年终于当上了技术总监（高权限），之后张三搬到30楼，可以随时向资源部（系统调用）申请公司各种资源与获取公司的机密信息（内核态），所谓是走上人生巅峰。

通过这个故事，我们发现，低权限的资源范围较小，高权限的资源范围更大，所谓的「用户态与内核态只是不同权限的资源范围」。

C P U 指令集权限

在说用户态与内核态之前，有必要说一下 C P U 指令集，指令集是 C P U 实现软件指挥硬件执行的媒介，具体来说每一条汇编语句都对应了一条 C P U 指令，而非常非常多的 C P U 指令 在一起，可以组成一个、甚至多个集合，指令的集合叫 C P U 指令集。

同时 C P U 指令集 有权限分级，大家试想，C P U 指令集 可以直接操作硬件的，要是因为指令操作的不规范`，造成的错误会影响整个计算机系统的。好比你写程序，因为对硬件操作不熟悉，导致操作系统内核、及其他所有正在运行的程序，都可能会因为操作失误而受到不可挽回的错误，最后只能重启计算机才行。

而对于硬件的操作是非常复杂的，参数众多，出问题的几率相当大，必须谨慎的进行操作，对开发人员来说是个艰巨的任务，还会增加负担，同时开发人员在这方面也不被信任，所以操作系统内核直接屏蔽开发人员对硬件操作的可能，都不让你碰到这些 C P U 指令集。

针对上面的需求，硬件设备商直接提供硬件级别的支持，做法就是对 C P U 指令集设置了权限，不同级别权限能使用的 C P U 指令集 是有限的，以 Intel C P U 为例，Inter把 C P U 指令集 操作的权限由高到低划为4级：

  • ring 0
  • ring 1
  • ring 2
  • ring 3

其中 ring 0 权限最高，可以使用所有 C P U 指令集，ring 3 权限最低，仅能使用常规 C P U 指令集，不能使用操作硬件资源的 C P U 指令集，比如 I O 读写、网卡访问、申请内存都不行，Linux系统仅采用ring 0 和 ring 3 这2个权限。

高情商

  • ring 0被叫做内核态，完全在操作系统内核中运行
  • ring 3被叫做用户态，在应用程序中运行

低情商

  • 执行内核空间的代码，具有ring 0保护级别，有对硬件的所有操作权限，可以执行所有C P U 指令集，访问任意地址的内存，在内核模式下的任何异常都是灾难性的，将会导致整台机器停机
  • 在用户模式下，具有ring 3保护级别，代码没有对硬件的直接控制权限，也不能直接访问地址的内存，程序是通过调用系统接口(System Call APIs)来达到访问硬件和内存，在这种保护模式下，即时程序发生崩溃也是可以恢复的，在电脑上大部分程序都是在，用户模式下运行的

用户态与内核态

通关了C P U 指令集权限，现在再说用户态与内核态就十分简单了，用户态与内核态的概念就是C P U 指令集权限的区别，进程中要读写 I O，必然会用到 ring 0 级别的 C P U 指令集，而此时 C P U 的指令集操作权限只有 ring 3，为了可以操作ring 0 级别的 C P U 指令集， C P U 切换指令集操作权限级别为 ring 0，C P U再执行相应的ring 0 级别的 C P U 指令集（内核代码），执行的内核代码会使用当前进程的内核栈。

PS：每个进程都有两个栈，分别是用户栈与内核栈，对应用户态与内核态的使用

用户态与内核态的空间

在内存资源上的使用，操作系统对用户态与内核态也做了限制，每个进程创建都会分配「虚拟空间地址」（不懂可以参考我的另一篇文章“15分钟！一文帮小白搞懂操作系统之内存”），以Linux32位操作系统为例，它的寻址空间范围是 4G（2的32次方），而操作系统会把虚拟控制地址划分为两部分，一部分为内核空间，另一部分为用户空间，高位的 1G（从虚拟地址 0xC0000000 到 0xFFFFFFFF）由内核使用，而低位的 3G（从虚拟地址 0x00000000 到 0xBFFFFFFF）由各个进程使用。

  • 用户态：只能操作 0-3G 范围的低位虚拟空间地址
  • 内核态：0-4G 范围的虚拟空间地址都可以操作，尤其是对 3-4G 范围的高位虚拟空间地址必须由内核态去操作
  • 补充：3G-4G 部分大家是共享的（指所有进程的内核态逻辑地址是共享同一块内存地址），是内核态的地址空间，这里存放在整个内核的代码和所有的内核模块，以及内核所维护的数据

每个进程的 4G 虚拟空间地址，高位 1G 都是一样的，即内核空间。只有剩余的 3G 才归进程自己使用，换句话说就是， 高位 1G 的内核空间是被所有进程共享的！

最后做个小结，我们通过指令集权限区分用户态和内核态，还限制了内存资源的使用，操作系统为用户态与内核态划分了两块内存空间，给它们对应的指令集使用

用户态与内核态的切换

相信大家都听过这样的话「用户态和内核态切换的开销大」，但是它的开销大在那里呢？简单点来说有下面几点

  • 保留用户态现场（上下文、寄存器、用户栈等）
  • 复制用户态参数，用户栈切到内核栈，进入内核态
  • 额外的检查（因为内核代码对用户不信任）
  • 执行内核态代码
  • 复制内核态代码执行结果，回到用户态
  • 恢复用户态现场（上下文、寄存器、用户栈等）

实际上操作系统会比上述的更复杂，这里只是个大概，我们可以发现一次切换经历了「用户态 -> 内核态 -> 用户态」。

用户态要主动切换到内核态，那必须要有入口才行，实际上内核态是提供了统一的入口，下面是Linux整体架构图

从上图我们可以看出来通过系统调用将Linux整个体系分为用户态和内核态，为了使应用程序访问到内核的资源，如CPU、内存、I/O，内核必须提供一组通用的访问接口，这些接口就叫系统调用。

库函数就是屏蔽这些复杂的底层实现细节，减轻程序员的负担，从而更加关注上层的逻辑实现，它对系统调用进行封装，提供简单的基本接口给程序员。

Shell顾名思义，就是外壳的意思，就好像把内核包裹起来的外壳，它是一种特殊的应用程序，俗称命令行。Shell也是可编程的，它有标准的Shell语法，符合其语法的文本叫Shell脚本，很多人都会用Shell脚本实现一些常用的功能，可以提高工作效率。

最后来说说，什么情况会导致用户态到内核态切换

  • 系统调用：用户态进程主动切换到内核态的方式，用户态进程通过系统调用向操作系统申请资源完成工作，例如 fork（）就是一个创建新进程的系统调用，系统调用的机制核心使用了操作系统为用户特别开放的一个中断来实现，如Linux 的 int 80h 中断，也可以称为软中断
  • 异常：当 C P U 在执行用户态的进程时，发生了一些没有预知的异常，这时当前运行进程会切换到处理此异常的内核相关进程中，也就是切换到了内核态，如缺页异常
  • 中断：当 C P U 在执行用户态的进程时，外围设备完成用户请求的操作后，会向 C P U 发出相应的中断信号，这时 C P U 会暂停执行下一条即将要执行的指令，转到与中断信号对应的处理程序去执行，也就是切换到了内核态。如硬盘读写操作完成，系统会切换到硬盘读写的中断处理程序中执行后边的操作等。

关联好文章推荐

关于我

Hi这里是阿星，一个热爱技术的93年Java程序猿，在公众号 「程序猿阿星」 里将会定期分享操作系统、计算机网络、Java、分布式、数据库等精品原创文章，2021，与您在 Be Better 的路上共同成长！。

非常感谢各位人才能 看到这里，创作不易，文章有帮助可以「点个赞」或「分享与评论」，都是支持（莫要白嫖）！

愿你我都能奔赴在各自想去的路上，我们下篇文章见！

本文转载自: 掘金

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

发表于

简介

前面介绍了广告定向的实现，它是一个查询密集型 (query-intensive) 程序，所以每个发给它的请求都会引起大量计算。本文将实现一个简单的社交网站，则会尽可能地减少用户在查看页面时系统所需要做的工作。 P184

用户和状态 P185

用户对象存储了用户的基本身份标识信息、用户的关注者人数、用户已发布的状态消息数量等信息，它是构建其他可用且有趣的数据的起点。 P185

状态消息记录了不同的用户都说了什么，以及不同用户之间进行了什么交流，这些由用户创建的状态消息是社交网站真正的内容。 P185

用户信息 P185

用户信息使用 HASH 来存储，用户信息包括：用户名、用户拥有的关注者人数、用户正在关注的人的数量、用户已发布的状态消息的数量、用户的注册日期以及其他一些元信息 (meta-information) 。示例： "user:139960061": {"login": "dr_josiah", "id": "139960061", "name": "Josiah Carlson", "followers": "176", "following": "79", "posts": "386", "signup": "1272948506"} P185

创建用户 P185

创建用户时需要根据用户指定的用户名以及当时的时间戳，创建一个正在关注数量、关注者数量、已发布状态消息数量都被设置为 0 的对象。 P185

创建时除了要对用户信息进行初始化，还需要对用户名进行加锁，用以防止多个请求 (request) 在同一时间内使用相同的用户名来创建新用户（可以使用 08. 实现自动补全、分布式锁和计数信号量 中实现的分布式锁进行加锁操作）。

状态消息 P186

状态消息使用 HASH 来存储，状态消息包括：消息本身、消息发布的时间、消息发布者的 id 、用户名（冗余用户名是为了展示时避免用户信息的 HASH ，因为用户名是不会改变的）以及其他一些关于状态消息的附加信息。示例： "status:223499221154799616": {"message": "My pleasure. I was amazed that...", "posted": "1342908431", "id": "223499221154799616", "uid": "139960061", "login": "dr_josiah"} P186

创建消息时需要先获取用户名，再将相关的信息组合起来存储到 HASH 中。 P187

主页时间线 P187

主页时间线是一个列表，它由用户以及用户正在关注的人所发布的状态消息组成。个人时间线与主页时间线类似，它只包含用户自己所发布的状态消息。因为主页时间线是用户访问网站时的主要入口，所以这些数据必须尽可能地易于获取。 P187

主页时间线使用有序集合存储，成员为状态消息的 id ，成员的分值为状态消息发布的时间戳。示例： "home:139960061": {..., "227138379358277633": "1342988984", ...} P188

获取主页时间线分为两步： P188

  1. 使用 ZREVRANGE 分页获取最新状态消息的 id 列表
  2. 使用流水线和 HGETALL 获取到状态消息 id 列表中所有状态消息

关注者列表和正在关注列表 P189

用户正在关注列表以及关注者列表同样用有序集合存储，成员为用户的 id ，成员的分值为用户开始关注某人或被某人关注的时间戳。示例："followers:139960061": {..., "558960079": "1342915440", "14502701": "1342917840", ...} "following:139960061": {..., "18697326": "1339286400", "558960079": "1342742400"} P188

当用户开始关注或者停止关注另一个用户时，就需要对这两个用户的正在关注有序集合和关注者有序集合进行更新 (ZADD, ZREM)，并修改他们用户信息中的关注数量和被关注数量 (HINCRBY)，同时还需要操作者的主页时间线，添加或删除另一个用户的状态消息 (ZUNIONSTORE)。 P189

删除操作需要两个命令，因为有序集合没有求差集的命令，有两种方式可以实现：

  1. 使用 ZREVRANGE 获取另一用户个人主页对应的有序集合的成员列表，然后再使用 ZREM 从操作用户主页流水线对应的有序集合中删除这些成员
  2. 先使用 ZUNIONSTORE 合并集合，操作用户主页流水线对应的有序集合的权重为 1 ，另一用户个人流水线对应的有序集合的权重为 0 ，聚合方式使用 MIN ，然后再使用 ZREMRANGEBYSCORE key 0 0 移除所有分值为 0 的成员（可以使用流水线合并）

所思

练习题让在给定的基础上支持批量关注和取消关注的操作。其实业务中大部分情况下单个和批量操作都是类似的，并且后续很可能会支持批量操作，为了减少后期后端开发量、避免改动接口签名或类似功能接口爆炸性增长，我常常都会提前实现批量操作的接口。我们平时开发时不能仅对当前需求进行处理，还要去理解整体功能并站在用户的角度进行思考，尽早关注可能的变动，这样才能开发出具有扩展性的接口。

状态消息的发布与删除 P191

用户可以执行的一个最基本的操作就是发布状态消息，前面已将实现了创建状态消息的逻辑，接下来将实现把新的状态消息添加到每个关注者的主页时间线中。 P191

为了让发布操作尽可能快地返回，我们在创建后会仅针对前 1000 个关注者的主页时间线添加新的状态消息，如果当前发布者的关注者超过了 1000 个，则使用 09. 实现任务队列、消息拉取和文件分发 中实现的任务队列异步处理后续的关注者。 P192

删除消息时同理操作即可，先删除状态消息本身，然后处理发布者自己的统计信息，再从个人时间线中删除该消息 id ，最后再从前 1000 个关注者的主页时间线中删除，若关注者人数超过 1000 ，再使用任务队列异步处理后续的关注者。 P193

流 API P194

构建一些函数来广播 (broadcast) 简单的事件 (event) ，然后由负责进行数据分析的事件监听器 (event listener) 来接收并处理这些事件，流 API 请求能在一段比较长的时间内持续地返回数据。 P194

流 API 的作用是随着时间的推移，产生一个由时间组成的序列，以此来让整个网络上的客户端和其他服务及时地了解到网站目前正在发生的事情。 P195

构建流 API 的过程中需要进行各种各样的决策，这些决策主要和以下 3 个问题有关： P195

  • 流 API 需要对外公开哪些事件？
    • 需要公开目前实现的四种用户操作：发布消息、删除消息、关注用户和取消关注用户。本节将以创建发布消息事件和删除消息事件为例进行实现
  • 是否需要进行访问限制？如果需要的话，采取何种方式实现？
    • 目前仅在涉及用户隐私或者系统资源的时候，考虑访问限制
  • 流 API 应该提供哪些过滤选项？
    • 既可以通过关注过滤器（基于用户进行过滤）、检测过滤器（基于关键字进行过滤）和位置过滤器来获取过滤后的消息，又可以通过类似推特的 firehose （可以获取所有公开消息） 和 sample （只能获取少量公开消息） 这样的流来获取一些随机的消息。
标识客户端 P198

为了区分不同的客户端，需要对其进行标识，大部分情况下来说只用考虑已登陆用户即可，那我么使用用户 id 进行标识，为了同时实现验证用户 id 的功能，可以使用 JWT 。

过滤消息 P200

我们将使用 Redis 的 PUBLISH 命令和 SUBSCRIBE 命令（05. Redis 其他命令简介 介绍了这两个命令的用法和缺陷）来实现订阅发布功能：当用户发布一条消息时，程序会通过 PUBLISH 发送给某个频道，而各个过滤器则通过 SUBSCRIBE 来订阅并接收那个频道的消息，并在发现与过滤器相匹配的消息时，将消息回传 (yield back) （如果是发布消息，则回传实体；如果是删除消息，则回传 id 和当前状态）给 Web 服务器，然后由服务器将这些消息发送给客户端。 P200

本文首发于公众号：满赋诸机（点击查看原文） 开源在 GitHub ：reading-notes/redis-in-action

本文转载自: 掘金

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

发表于

前言

有位朋友去阿里面试，他说面试官给了几条查询SQL，问:需要执行几次树搜索操作？我朋友当时是有点懵的，后来冷静思考，才发现就是考索引的几个基础知识点~ 本文我们分九个索引知识点，一起来探讨一下。如果有不正确的话，欢迎指出哈，一起学习

  • 公众号：捡田螺的小男孩
  • github地址，感谢每颗star

github.com/whx123/Java…

  • 面试官考点之索引是什么？
  • 面试官考点之索引类型
  • 面试官考点之为什么选择B+树作为索引结构
  • 面试官考点之一次索引搜索过程
  • 面试官考点之覆盖索引
  • 面试官考点之索引失效场景
  • 面试官考点之最左前缀
  • 面试官考点之索引下推
  • 面试官考点之大表添加索引

一、面试官考点之索引是什么？

  • 索引是一种能提高数据库查询效率的数据结构。它可以比作一本字典的目录，可以帮你快速找到对应的记录。
  • 索引一般存储在磁盘的文件中，它是占用物理空间的。
  • 正所谓水能载舟，也能覆舟。适当的索引能提高查询效率，过多的索引会影响数据库表的插入和更新功能。

二、索引有哪些类型类型

数据结构维度

  • B+树索引：所有数据存储在叶子节点，复杂度为O(logn)，适合范围查询。
  • 哈希索引: 适合等值查询，检索效率高，一次到位。
  • 全文索引：MyISAM和InnoDB中都支持使用全文索引，一般在文本类型char,text,varchar类型上创建。
  • R-Tree索引: 用来对GIS数据类型创建SPATIAL索引

物理存储维度

  • 聚集索引：聚集索引就是以主键创建的索引，在叶子节点存储的是表中的数据。
  • 非聚集索引：非聚集索引就是以非主键创建的索引，在叶子节点存储的是主键和索引列。

逻辑维度

  • 主键索引：一种特殊的唯一索引，不允许有空值。
  • 普通索引：MySQL中基本索引类型，允许空值和重复值。
  • 联合索引：多个字段创建的索引，使用时遵循最左前缀原则。
  • 唯一索引：索引列中的值必须是唯一的，但是允许为空值。
  • 空间索引：MySQL5.7之后支持空间索引，在空间索引这方面遵循OpenGIS几何数据模型规则。

三、面试官考点之为什么选择B+树作为索引结构

可以从几个维度去看这个问题，查询是否够快，效率是否稳定，存储数据多少，以及查找磁盘次数等等。为什么不是哈希结构？为什么不是二叉树，为什么不是平衡二叉树，为什么不是B树，而偏偏是B+树呢？

我们写业务SQL查询时，大多数情况下，都是范围查询的，如一下SQL

1
sql复制代码select * from employee where age between 18 and 28;

为什么不使用哈希结构？

我们知道哈希结构，类似k-v结构，也就是，key和value是一对一关系。它用于等值查询还可以，但是范围查询它是无能为力的哦。

为什么不使用二叉树呢？

先回忆下二叉树相关知识啦~ 所谓二叉树，特点如下：

  • 每个结点最多两个子树，分别称为左子树和右子树。
  • 左子节点的值小于当前节点的值，当前节点值小于右子节点值
  • 顶端的节点称为跟节点，没有子节点的节点值称为叶子节点。

我们脑海中，很容易就浮现出这种二叉树结构图：

但是呢，有些特殊二叉树，它可能这样的哦：

如果二叉树特殊化为一个链表，相当于全表扫描。那么还要索引干嘛呀？因此，一般二叉树不适合作为索引结构。

为什么不使用平衡二叉树呢？

平衡二叉树特点：它也是一颗二叉查找树，任何节点的两个子树高度最大差为1。所以就不会出现特殊化一个链表的情况啦。

但是呢：

  • 平衡二叉树插入或者更新是，需要左旋右旋维持平衡，维护代价大
  • 如果数量多的话，树的高度会很高。因为数据是存在磁盘的，以它作为索引结构，每次从磁盘读取一个节点，操作IO的次数就多啦。

为什么不使用B树呢？

数据量大的话，平衡二叉树的高度会很高，会增加IO嘛。那为什么不选择同样数据量，高度更矮的B树呢？

B树相对于平衡二叉树，就可以存储更多的数据，高度更低。但是最后为甚选择B+树呢？因为B+树是B树的升级版：

  • B+树非叶子节点上是不存储数据的，仅存储键值，而B树节点中不仅存储键值，也会存储数据。innodb中页的默认大小是16KB，如果不存储数据，那么就会存储更多的键值，相应的树的阶数（节点的子节点树）就会更大，树就会更矮更胖，如此一来我们查找数据进行磁盘的IO次数有会再次减少，数据查询的效率也会更快。
  • B+树索引的所有数据均存储在叶子节点，而且数据是按照顺序排列的，链表连着的。那么B+树使得范围查找，排序查找，分组查找以及去重查找变得异常简单。

四、面试官考点之一次B+树索引搜索过程

面试官： 假设有以下表结构，并且有这几条数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
sql复制代码CREATE TABLE `employee` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  `date` datetime DEFAULT NULL,
  `sex` int(1) DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_age` (`age`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

insert into employee values(100,'小伦',43,'2021-01-20','0');
insert into employee values(200,'俊杰',48,'2021-01-21','0');
insert into employee values(300,'紫琪',36,'2020-01-21','1');
insert into employee values(400,'立红',32,'2020-01-21','0');
insert into employee values(500,'易迅',37,'2020-01-21','1');
insert into employee values(600,'小军',49,'2021-01-21','0');
insert into employee values(700,'小燕',28,'2021-01-21','1');

面试官： 如果执行以下的查询SQL，需要执行几次的树搜索操作？可以画下对应的索引结构图~

1
csharp复制代码select * from Temployee where age=32;

解析： 其实这个，面试官就是考察候选人是否熟悉B+树索引结构图。可以像酱紫回答~

  • 先画出idx_age索引的索引结构图，大概如下：

  • 再画出id主键索引，我们先画出聚族索引结构图，如下：

因此，这条 SQL 查询语句执行大概流程就是酱紫：

    1. 搜索idx_age索引树，将磁盘块1加载到内存，由于32<37,搜索左路分支，到磁盘寻址磁盘块2。
    1. 将磁盘块2加载到内存中，在内存继续遍历，找到age=32的记录，取得id = 400.
    1. 拿到id=400后，回到id主键索引树。
    1. 搜索id主键索引树，将磁盘块1加载内存，在内存遍历，找到了400，但是B+树索引非叶子节点是不保存数据的。索引会继续搜索400的右分支，到磁盘寻址磁盘块3.
    1. 将磁盘块3加载内存，在内存遍历，找到id=400的记录，拿到R4这一行的数据，好的，大功告成。

因此，这个SQL查询，执行了几次树的搜索操作，是不是一步了然了呀。特别的，在idx_age二级索引树找到主键id后，回到id主键索引搜索的过程,就称为回表。

什么是回表？拿到主键再回到主键索引查询的过程，就叫做回表

五、面试官考点之覆盖索引

面试官： 如果不用select *, 而是使用select id,age，以上的题目执行了几次树搜索操作呢？

解析： 这个问题，主要考察候选人的覆盖索引知识点。回到idx_age索引树，你可以发现查询选项id和age都在叶子节点上了。因此，可以直接提供查询结果啦，根本就不需要再回表了~

覆盖索引：在查询的数据列里面，不需要回表去查，直接从索引列就能取到想要的结果。换句话说，你SQL用到的索引列数据，覆盖了查询结果的列，就算上覆盖索引了。

所以，相对于上个问题，就是省去了回表的树搜索操作。

六、面试官考点之索引失效

面试官： 如果我现在给name字段加上普通索引，然后用个like模糊搜索，那会执行多少次查询呢？SQL如下：

1
sql复制代码select * from employee where name like '%杰伦%';

解析： 这里考察的知识点就是，like是否会导致不走索引，看先该SQL的explain执行计划吧。其实like 模糊搜索，会导致不走索引的，如下：

因此，这条SQL最后就全表扫描啦~日常开发中，这几种骚操作都可能会导致索引失效，如下：

  • 查询条件包含or，可能导致索引失效
  • 如何字段类型是字符串，where时一定用引号括起来，否则索引失效
  • like通配符可能导致索引失效。
  • 联合索引，查询时的条件列不是联合索引中的第一个列，索引失效。
  • 在索引列上使用mysql的内置函数，索引失效。
  • 对索引列运算（如，+、-、*、/），索引失效。
  • 索引字段上使用（！= 或者 < >，not in）时，可能会导致索引失效。
  • 索引字段上使用is null， is not null，可能导致索引失效。
  • 左连接查询或者右连接查询查询关联的字段编码格式不一样，可能导致索引失效。
  • mysql估计使用全表扫描要比使用索引快,则不使用索引。

七、面试官考点联合索引之最左前缀原则

面试官： 如果我现在给name,age字段加上联合索引索引，以下SQL执行多少次树搜索呢？先画下索引树？

1
sql复制代码select * from employee where name like '小%' order by age desc;

解析： 这里考察联合索引的最左前缀原则以及like是否中索引的知识点。组合索引树示意图大概如下：

联合索引项是先按姓名name从小到大排序，如果名字name相同，则按年龄age从小到大排序。面试官要求查所有名字第一个字是“小”的人，SQL的like ‘小%’是可以用上idx_name_age联合索引的。

该查询会沿着idx_name_age索引树，找到第一个字是小的索引值，因此依次找到小军、小伦、小燕、，分别拿到Id=600、100、700，然后回三次表，去找对应的记录。 这里面的最左前缀小，就是字符串索引的最左M个字符。实际上，

  • 这个最左前缀可以是联合索引的最左N个字段。比如组合索引（a,b,c）可以相当于建了（a），（a,b）,(a,b,c)三个索引，大大提高了索引复用能力。
  • 最左前缀也可以是字符串索引的最左M个字符。

八、面试官考点之索引下推

面试官： 我们还是居于组合索引 idx_name_age，以下这个SQL执行几次树搜索呢？

1
sql复制代码select * from employee where name like '小%' and age=28 and sex='0';

解析： 这里考察索引下推的知识点，如果是Mysql5.6之前，在idx_name_age索引树，找出所有名字第一个字是“小”的人，拿到它们的主键id，然后回表找出数据行，再去对比年龄和性别等其他字段。如图：

有些朋友可能觉得奇怪，（name,age)不是联合索引嘛？为什么选出包含“小”字后，不再顺便看下年龄age再回表呢，不是更高效嘛？所以呀，MySQL 5.6 就引入了索引下推优化，可以在索引遍历过程中，对索引中包含的字段先做判断，直接过滤掉不满足条件的记录，减少回表次数。

因此，MySQL5.6版本之后，选出包含“小”字后，顺表过滤age=28，,所以就只需一次回表。

九、 面试官考点之大表添加索引

面试官： 如果一张表数据量级是千万级别以上的，那么，给这张表添加索引，你需要怎么做呢？

解析： 我们需要知道一点，给表添加索引的时候，是会对表加锁的。如果不谨慎操作，有可能出现生产事故的。可以参考以下方法：

  • 1.先创建一张跟原表A数据结构相同的新表B。
  • 2.在新表B添加需要加上的新索引。
  • 3.把原表A数据导到新表B
  • 4.rename新表B为原表的表名A，原表A换别的表名；

总结与练习

本文主要讲解了索引的9大关键知识点，希望对大家有帮助。接下来呢，给大家出一道，有关于我最近业务开发遇到的加索引SQL，看下大家是怎么回答的，有兴趣可以联系我哈~题目如下：

1
2
sql复制代码
select * from A where type ='1' and status ='s' order by create_time desc;

假设type有9种类型，区分度性还算可以，status的区分度不高（有3种类型），那么你是如何加索引呢？

  • 是给type加单索引
  • 还是（type，status，create_time）联合索引
  • 还是（type，create_time）联合索引呢？

参看与感谢

本文转载自: 掘金

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

发表于

本文分享Spring中如何实现Redis响应式交互模式。

本文将模拟一个用户服务，并使用Redis作为数据存储服务器。

本文涉及两个java bean，用户与权益

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
kotlin复制代码public class User {
    private long id;
    private String name;
    // 标签
    private String label;
    // 收货地址经度
    private Double deliveryAddressLon;
    // 收货地址维度
    private Double deliveryAddressLat;
    // 最新签到日
    private String lastSigninDay;
    // 积分
    private Integer score;
    // 权益
    private List<Rights> rights;
    ...
}

public class Rights {
    private Long id;
    private Long userId;
    private String name;
    ...
}

启动

引入依赖

1
2
3
4
xml复制代码    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
    </dependency>

添加Redis配置

1
2
3
4
ini复制代码spring.redis.host=192.168.56.102
spring.redis.port=6379
spring.redis.password=
spring.redis.timeout=5000

SpringBoot启动

1
2
3
4
5
6
7
8
java复制代码@SpringBootApplication
public class UserServiceReactive {
    public static void main(String[] args) {
        new SpringApplicationBuilder(
                UserServiceReactive.class)
                .web(WebApplicationType.REACTIVE).run(args);
    }
}

应用启动后，Spring会自动生成ReactiveRedisTemplate（它的底层框架是Lettuce）。

ReactiveRedisTemplate与RedisTemplate使用类似，但它提供的是异步的，响应式Redis交互方式。

这里再强调一下，响应式编程是异步的，ReactiveRedisTemplate发送Redis请求后不会阻塞线程，当前线程可以去执行其他任务。

等到Redis响应数据返回后，ReactiveRedisTemplate再调度线程处理响应数据。

响应式编程可以通过优雅的方式实现异步调用以及处理异步结果，正是它的最大的意义。

序列化

ReactiveRedisTemplate默认使用的序列化是Jdk序列化，我们可以配置为json序列化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码@Bean
public RedisSerializationContext redisSerializationContext() {
    RedisSerializationContext.RedisSerializationContextBuilder builder = RedisSerializationContext.newSerializationContext();
    builder.key(StringRedisSerializer.UTF_8);
    builder.value(RedisSerializer.json());
    builder.hashKey(StringRedisSerializer.UTF_8);
    builder.hashValue(StringRedisSerializer.UTF_8);

    return builder.build();
}

@Bean
public ReactiveRedisTemplate reactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory) {
    RedisSerializationContext serializationContext = redisSerializationContext();
    ReactiveRedisTemplate reactiveRedisTemplate = new ReactiveRedisTemplate(connectionFactory,serializationContext);
    return reactiveRedisTemplate;
}

builder.hashValue方法指定Redis列表值的序列化方式，由于本文Redis列表值只存放字符串，所以还是设置为StringRedisSerializer.UTF_8。

基本数据类型

ReactiveRedisTemplate支持Redis字符串，散列，列表，集合，有序集合等基本的数据类型。

本文使用散列保存用户信息，列表保存用户权益，其他基本数据类型的使用本文不展开。

1
2
3
4
5
6
7
8
9
10
11
typescript复制代码public Mono<Boolean>  save(User user) {
    ReactiveHashOperations<String, String, String> opsForHash = redisTemplate.opsForHash();
    Mono<Boolean>  userRs = opsForHash.putAll("user:" + user.getId(), beanToMap(user));
    if(user.getRights() != null) {
        ReactiveListOperations<String, Rights> opsForRights = redisTemplate.opsForList();
        opsForRights.leftPushAll("user:rights:" + user.getId(), user.getRights()).subscribe(l -> {
            logger.info("add rights:{}", l);
        });
    }
    return userRs;
}

beanToMap方法负责将User类转化为map。

HyperLogLog

Redis HyperLogLog结构可以统计一个集合内不同元素的数量。

使用HyperLogLog统计每天登录的用户量

1
2
3
4
vbscript复制代码public Mono<Long>  login(User user) {
    ReactiveHyperLogLogOperations<String, Long> opsForHyperLogLog = redisTemplate.opsForHyperLogLog();
    return opsForHyperLogLog.add("user:login:number:" + LocalDateTime.now().toString().substring(0, 10), user.getId());
}

BitMap

Redis BitMap（位图）通过一个Bit位表示某个元素对应的值或者状态。由于Bit是计算机存储中最小的单位，使用它进行储存将非常节省空间。

使用BitMap记录用户本周是否有签到

1
2
3
4
5
6
vbnet复制代码public void addSignInFlag(long userId) {
    String key = "user:signIn:" + LocalDateTime.now().getDayOfYear()/7 + (userId >> 16);
    redisTemplate.opsForValue().setBit(
            key, userId & 0xffff , true)
    .subscribe(b -> logger.info("set:{},result:{}", key, b));
}

userId高48位用于将用户划分到不同的key，低16位作为位图偏移参数offset。

offset参数必须大于或等于0，小于2^32(bit 映射被限制在 512 MB 之内)。

Geo

Redis Geo可以存储地理位置信息，并对地理位置进行计算。

如查找给定范围内的仓库信息

1
2
3
4
5
6
7
scss复制代码public Flux getWarehouseInDist(User u, double dist) {
    ReactiveGeoOperations<String, String> geo = redisTemplate.opsForGeo();
    Circle circle = new Circle(new Point(u.getDeliveryAddressLon(), u.getDeliveryAddressLat()), dist);
    RedisGeoCommands.GeoRadiusCommandArgs args =
            RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().includeDistance().sortAscending();
    return geo.radius("warehouse:address", circle, args);
}

warehouse:address这个集合中需要先保存好仓库地理位置信息。

ReactiveGeoOperations#radius方法可以查找集合中地理位置在给定范围内的元素，它中还支持添加元素到集合，计算集合中两个元素地理位置距离等操作。

Lua

ReactiveRedisTemplate也可以执行Lua脚本。

下面通过Lua脚本完成用户签到逻辑：如果用户今天未签到，允许签到，积分加1，如果用户今天已签到，则拒接操作。

1
2
3
4
5
6
7
8
arduino复制代码public Flux<String> addScore(long userId) {
    DefaultRedisScript<String> script = new DefaultRedisScript<>();
    script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/signin.lua")));
    List<String> keys = new ArrayList<>();
    keys.add(String.valueOf(userId));
    keys.add(LocalDateTime.now().toString().substring(0, 10));
    return redisTemplate.execute(script, keys);
}

signin.lua内容如下

1
2
3
4
5
6
7
8
9
sql复制代码local score=redis.call('hget','user:'..KEYS[1],'score')
local day=redis.call('hget','user:'..KEYS[1],'lastSigninDay')
if(day==KEYS[2])
    then
    return '0'
else
    redis.call('hset','user:'..KEYS[1],'score', score+1,'lastSigninDay',KEYS[2])
    return '1'
end

Stream

Redis Stream 是 Redis 5.0 版本新增加的数据类型。该类型可以实现消息队列，并提供消息的持久化和主备复制功能，并且可以记住每一个客户端的访问位置，还能保证消息不丢失。

Redis借鉴了kafka的设计，一个Stream内可以存在多个消费组，一个消费组内可以存在多个消费者。

如果一个消费组内某个消费者消费了Stream中某条消息，则这消息不会被该消费组其他消费者消费到，当然，它还可以被其他消费组中某个消费者消费到。

下面定义一个Stream消费者，负责处理接收到的权益数据

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
java复制代码@Component
public class RightsStreamConsumer implements ApplicationRunner, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(RightsStreamConsumer.class);

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    private StreamMessageListenerContainer<String, ObjectRecord<String, Rights>> container;
    // Stream队列
    private static final String STREAM_KEY = "stream:user:rights";
    // 消费组
    private static final String STREAM_GROUP = "user-service";
    // 消费者
    private static final String STREAM_CONSUMER = "consumer-1";

    @Autowired
    @Qualifier("reactiveRedisTemplate")
    private ReactiveRedisTemplate redisTemplate;

    public void run(ApplicationArguments args) throws Exception {

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

        // prepareStreamAndGroup查找Stream信息，如果不存在，则创建Stream
        prepareStreamAndGroup(redisTemplate.opsForStream(), STREAM_KEY , STREAM_GROUP)
                .subscribe(stream -> {
            // 为Stream创建一个消费者，并绑定处理类
            container.receive(Consumer.from(STREAM_GROUP, STREAM_CONSUMER),
                    StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()),
                    new StreamMessageListener());
            container.start();
        });
    }

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

    // 查找Stream信息，如果不存在，则创建Stream
    private Mono<StreamInfo.XInfoStream> prepareStreamAndGroup(ReactiveStreamOperations<String, ?, ?> ops, String stream, String group) {
        // info方法查询Stream信息，如果该Stream不存在，底层会报错，这时会调用onErrorResume方法。
        return ops.info(stream).onErrorResume(err -> {
            logger.warn("query stream err:{}", err.getMessage());
            // createGroup方法创建Stream
            return ops.createGroup(stream, group).flatMap(s -> ops.info(stream));
        });
    }

    // 消息处理对象
    class  StreamMessageListener implements StreamListener<String, ObjectRecord<String, Rights>> {
        public void onMessage(ObjectRecord<String, Rights> message) {
            // 处理消息
            RecordId id = message.getId();
            Rights rights = message.getValue();
            logger.info("receive id:{},rights:{}", id, rights);
            redisTemplate.opsForList().leftPush("user:rights:" + rights.getUserId(), rights).subscribe(l -> {
                logger.info("add rights:{}", l);
            });
        }
    }
}

下面看一下如何发送信息

1
2
3
4
5
6
java复制代码public Mono<RecordId> addRights(Rights r) {
    String streamKey = "stream:user:rights";//stream key
    ObjectRecord<String, Rights> record = ObjectRecord.create(streamKey, r);
    Mono<RecordId> mono = redisTemplate.opsForStream().add(record);
    return mono;
}

创建一个消息记录对象ObjectRecord，并通过ReactiveStreamOperations发送信息记录。

Sentinel、Cluster

ReactiveRedisTemplate也支持Redis Sentinel、Cluster集群模式，只需要调整配置即可。

Sentinel配置如下

1
2
3
ini复制代码spring.redis.sentinel.master=mymaster
spring.redis.sentinel.nodes=172.17.0.4:26379,172.17.0.5:26379,172.17.0.6:26379
spring.redis.sentinel.password=

spring.redis.sentinel.nodes配置的是Sentinel节点IP地址和端口，不是Redis实例节点IP地址和端口。

Cluster配置如下

1
2
3
ini复制代码spring.redis.cluster.nodes=172.17.0.2:6379,172.17.0.3:6379,172.17.0.4:6379,172.17.0.5:6379,172.17.0.6:6379,172.17.0.7:6379
spring.redis.lettuce.cluster.refresh.period=10000
spring.redis.lettuce.cluster.refresh.adaptive=true

如Redis Cluster中node2是node1的从节点，Lettuce中会缓存该信息，当node1宕机后，Redis Cluster会将node2升级为主节点。但Lettuce不会自动将请求切换到node2，因为它的缓冲没有刷新。

开启spring.redis.lettuce.cluster.refresh.adaptive配置，Lettuce可以定时刷新Redis Cluster集群缓存信息，动态改变客户端的节点情况，完成故障转移。

暂时未发现ReactiveRedisTemplate实现pipeline，事务的方案。

官方文档：docs.spring.io/spring-data…

文章完整代码：gitee.com/binecy/bin-…

如果您觉得本文不错，欢迎关注我的微信公众号，系列文章持续更新中。您的关注是我坚持的动力！

本文转载自: 掘金

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

发表于

  • 场景还原

事情是这个样子的，日常看git提交记录摸鱼的时候，看到同事提交了一段代码，简化代码如下：

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
java复制代码@Configuration
public class BaseConfig implements AsyncConfigurer {

    @Autowired
    private AsyncService asyncService;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(50);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("defaultExecutor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return ((Throwable var1, Method var2, Object... var3)->{
            this.asyncService.Print();
            log.info("怎么又是你！");
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码@Slf4j
@Component
public class AsyncServiceImpl implements AsyncService {

    @Async
    @Override
    public void asyncPrint() {
        Thread thread = Thread.currentThread();
        log.info("current thread id:[{}], name:[{}]", thread.getId(), thread.getName());
    }
    @Override
    public void Print() {
        log.info("打印。。。");
    }
}

代码就是实现了AsyncConfigurer这个接口，然后重写了getAsyncExecutor，getAsyncUncaughtExceptionHandler这两个方法，前一个是自定义异步任务的线程池，后者是可以在异步任务发生异常的时候捕捉到做一些特殊处理。AsyncService代码很简单就是打印了一下当前线程的标识，然后我尝试写个测试类测试一下handler这个方法是否可以在异步的asyncPrint执行之后抓到异常，测试类如下，只是简单的调用一下异步方法，然后在异步方法里面加一个int i = 1/0;让程序抛错：

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码@SpringBootTest(classes = EvtwboApplication.class)
@RunWith(SpringRunner.class)
public class SpringDemoTest {

    @Autowired
    private AsyncService asyncService;

    @Test
    public void testAsync(){
        this.asyncService.asyncPrint();
    }
}

测试方法执行后输出如下日志：

1
2
3
4
5
6
java复制代码2021-01-30 21:06:05.452  INFO 8320 --- [           main] com.demo.service.impl.AsyncServiceImpl   : current thread id:[1], name:[main]

java.lang.ArithmeticException: / by zero

	at com.demo.service.impl.AsyncServiceImpl.asyncPrint(AsyncServiceImpl.java:17)
	at com.demo.config.SpringDemoTest.testAsync(SpringDemoTest.java:20)

结果很奇怪，异常并没有被之前写的异常处理方法捕捉到，而且可以注意到异步service打印的日志打印出了主线程main，这说明是主线程执行了这段代码，@Async注解并没有生效。

先说下结论再分析吧，其实就是bean注入顺序的问题，在配置类里面注入了业务类，导致了业务类在BeanPostProcessor实例化之前就实例化了，这个时候通过aop生成代理对象来起作用的注解就会失效。

  • 问题分析

要想知道为什么@Async注解为什么没有生效，那我们得先知道它是怎么对我们的方法起作用的。
在使用@Async这个注解之前，我们会先在一个配置类上面添加一个@EnableAsync注解，代码类似：

1
2
3
4
5
6
7
8
java复制代码@EnableAsync
@SpringBootApplication
public class EvtwboApplication {

    public static void main(String[] args) {
        new SpringApplication(EvtwboApplication.class).run(args);
    }
}

现在的很多框架里面都会使用这种类似的注解，看起来像是个开关一样，实际上也是如此，通过这样一个注解会import一些bean，使像@Async这样的自定义注解生效，我们点进去看一下

1
2
3
4
5
6
7
java复制代码@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
AdviceMode mode() default AdviceMode.PROXY;
}

这个注解里面还有一些属性，暂时用不到，先不看，列出的这个属性我们可以看到AdviceMode默认是AdviceMode.PROXY，上面的元注解@Import(AsyncConfigurationSelector.class)就是它导入的类，我们点进这个类看一下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
java复制代码public class AsyncConfigurationSelector extends AdviceModeImportSelector<EnableAsync> {

	@Override
	@Nullable
	public String[] selectImports(AdviceMode adviceMode) {
		switch (adviceMode) {
			case PROXY:
				return new String[] {ProxyAsyncConfiguration.class.getName()};
			case ASPECTJ:
				return new String[] {ASYNC_EXECUTION_ASPECT_CONFIGURATION_CLASS_NAME};
			default:
				return null;
		}
	}

}

这边因为咱们没有给AdviceMode赋值，所以会走上面一个分支，返回的是一个ProxyAsyncConfiguration，也就是它实际上导入的类，点进这个类看一下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
java复制代码@Configuration
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyAsyncConfiguration extends AbstractAsyncConfiguration {

	@Bean(name = TaskManagementConfigUtils.ASYNC_ANNOTATION_PROCESSOR_BEAN_NAME)
	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
	public AsyncAnnotationBeanPostProcessor asyncAdvisor() {
		Assert.notNull(this.enableAsync, "@EnableAsync annotation metadata was not injected");
		AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
		bpp.configure(this.executor, this.exceptionHandler);
		Class<? extends Annotation> customAsyncAnnotation = this.enableAsync.getClass("annotation");
		if (customAsyncAnnotation != AnnotationUtils.getDefaultValue(EnableAsync.class, "annotation")) {
			bpp.setAsyncAnnotationType(customAsyncAnnotation);
		}
		bpp.setProxyTargetClass(this.enableAsync.getBoolean("proxyTargetClass"));
		bpp.setOrder(this.enableAsync.<Integer>getNumber("order"));
		return bpp;
	}
}

可以看到它是一个配置类，导入了AsyncAnnotationBeanPostProcessor，并且beanname是internalAsyncAnnotationProcessor，看名字很明显这是一个BeanPostProcessor，看那一下这个接口

1
2
3
4
5
6
7
8
9
10
11
java复制代码public interface BeanPostProcessor {
    @Nullable
    default Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Nullable
    default Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }
}

只有两个默认方法，方法名还是很好了解的postProcessBeforeInitialization这个方法实在bean初始化前调用的，可以对创建出来的bean做一些处理，而postProcessAfterInitialization自然就是在bean初始化之后调用的，这两个方法都是在对象实例创建之后initializeBean中调用初始方法前后的埋点，由于这些知识属于spring的refresh过程，这里不展开，稍微提一下下面会用到的知识点，首先是AbstractApplicationContext的refresh方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码@Override
	public void refresh() throws BeansException, IllegalStateException {
			prepareRefresh();
			ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory();
			prepareBeanFactory(beanFactory);
				postProcessBeanFactory(beanFactory);
				invokeBeanFactoryPostProcessors(beanFactory);
				// Register bean processors that intercept bean creation.
				registerBeanPostProcessors(beanFactory);
				initMessageSource();
				initApplicationEventMulticaster();
				onRefresh();
				registerListeners();
				// Instantiate all remaining (non-lazy-init) singletons.
				finishBeanFactoryInitialization(beanFactory);
				finishRefresh();
			}

代码里面移除了一些异常处理及无关的注释，想要了解的可以直接到源码里面看，这个refresh便是我们的applicationcontext容器的核心启动代码了，现在我们看一下留下注释的这两个方法。首先是registerBeanPostProcessors，看注释可以知道这是在注册BeanPostProcessors用来在bean实例化过程中拦截的一些埋点，我们可以跟进去看一下，最终是在PostProcessorRegistrationDelegate的registerBeanPostProcessors方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码public static void registerBeanPostProcessors(
			ConfigurableListableBeanFactory beanFactory, AbstractApplicationContext applicationContext) {
		// First, register the BeanPostProcessors that implement PriorityOrdered.

		// Next, register the BeanPostProcessors that implement Ordered.
		List<BeanPostProcessor> orderedPostProcessors = new ArrayList<>(orderedPostProcessorNames.size());
		for (String ppName : orderedPostProcessorNames) {
			BeanPostProcessor pp = beanFactory.getBean(ppName, BeanPostProcessor.class);
			orderedPostProcessors.add(pp);
			if (pp instanceof MergedBeanDefinitionPostProcessor) {
				internalPostProcessors.add(pp);
			}
		}
		sortPostProcessors(orderedPostProcessors, beanFactory);
		registerBeanPostProcessors(beanFactory, orderedPostProcessors);

		// Now, register all regular BeanPostProcessors.

		// Finally, re-register all internal BeanPostProcessors.
		// Re-register post-processor for detecting inner beans as ApplicationListeners,
		// moving it to the end of the processor chain (for picking up proxies etc).
		beanFactory.addBeanPostProcessor(new ApplicationListenerDetector(applicationContext));
	}

可以看到他是将先注册实现了PriorityOrdered接口的类，然后是实现了Ordered接口的，接着是其他的，我们可以看到AsyncAnnotationBeanPostProcessor是实现了Ordered接口的，所以会在第二个分支里面注册

从源码里面可以看到是先beanFactory.getBean(ppName, BeanPostProcessor.class);先创建这个实例，然后registerBeanPostProcessors(beanFactory, orderedPostProcessors);，里面实际做的是beanFactory.addBeanPostProcessor(postProcessor);就是把这个实例加到容器中去。那现在这个BeanPostProcessor已经注册完了，我们在看一下它是在哪里起作用的，源码是在refresh里面的finishBeanFactoryInitialization方法里，看注释可以知道这里是实际上初始化非懒加载的单例的，往里面跟，最终是到AbstractAutowireCapableBeanFactory的doCreateBean方法，其实之前的AsyncAnnotationBeanPostProcessor也是通过这个方法创建的


我们可以在这个方法里面看到有这么一句exposedObject = initializeBean(beanName, exposedObject, mbd);，这边就是初始化bean的地方，点进去

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码protected Object initializeBean(final String beanName, final Object bean, @Nullable RootBeanDefinition mbd) {

		Object wrappedBean = bean;
		if (mbd == null || !mbd.isSynthetic()) {
			wrappedBean = applyBeanPostProcessorsBeforeInitialization(wrappedBean, beanName);
		}

			invokeInitMethods(beanName, wrappedBean, mbd);
		if (mbd == null || !mbd.isSynthetic()) {
			wrappedBean = applyBeanPostProcessorsAfterInitialization(wrappedBean, beanName);
		}

		return wrappedBean;
	}

很明显applyBeanPostProcessorsBeforeInitialization，applyBeanPostProcessorsAfterInitialization这两个方法就是调用我们的AsyncAnnotationBeanPostProcessor的方法，实际上我们的后置处理器是在初始化之后做代理的，我们可以跟进去看一下


可以看到这边AsyncAnnotationBeanPostProcessor可以生成一个对象替代原来的对象，也就是所谓的代理，继续往里面看

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
java复制代码    public Object postProcessAfterInitialization(Object bean, String beanName) {
        if (this.advisor != null && !(bean instanceof AopInfrastructureBean)) {
        //这边针对已经代理过的对象，将这个通知加到通知链里面去
            if (bean instanceof Advised) {
                Advised advised = (Advised)bean;
                if (!advised.isFrozen() && this.isEligible(AopUtils.getTargetClass(bean))) {
                    if (this.beforeExistingAdvisors) {
                        advised.addAdvisor(0, this.advisor);
                    } else {
                        advised.addAdvisor(this.advisor);
                    }

                    return bean;
                }
            }

            if (this.isEligible(bean, beanName)) {
            //这边会生成一个代理对象
                ProxyFactory proxyFactory = this.prepareProxyFactory(bean, beanName);
                if (!proxyFactory.isProxyTargetClass()) {
                    this.evaluateProxyInterfaces(bean.getClass(), proxyFactory);
                }

                proxyFactory.addAdvisor(this.advisor);
                this.customizeProxyFactory(proxyFactory);
                return proxyFactory.getProxy(this.getProxyClassLoader());
            } else {
                return bean;
            }
        } else {
            return bean;
        }
    }

这边可以看到是生成了一个代理对象，并携带了通知，这个通知实际上是AsyncAnnotationAdvisor，我们看一下这个类的buildAdvice方法

1
2
3
4
5
6
7
8
java复制代码	protected Advice buildAdvice(
			@Nullable Supplier<Executor> executor, @Nullable Supplier<AsyncUncaughtExceptionHandler> exceptionHandler) {

		AnnotationAsyncExecutionInterceptor interceptor = new AnnotationAsyncExecutionInterceptor(null);
                //这边的两个参数明显就是咱们asyncconfig中的两个方法定义的了
		interceptor.configure(executor, exceptionHandler);
		return interceptor;
	}

原来这个通知本质上是一个拦截器，我们在看一下这个拦截器AnnotationAsyncExecutionInterceptor的invoke方法，实际在他的父类AsyncExecutionInterceptor里

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
java复制代码    @Nullable
    public Object invoke(MethodInvocation invocation) throws Throwable {
        Class<?> targetClass = invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null;
        Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);
        Method userDeclaredMethod = BridgeMethodResolver.findBridgedMethod(specificMethod);
        AsyncTaskExecutor executor = this.determineAsyncExecutor(userDeclaredMethod);
        if (executor == null) {
            throw new IllegalStateException("No executor specified and no default executor set on AsyncExecutionInterceptor either");
        } else {
            Callable<Object> task = () -> {
                try {
                    Object result = invocation.proceed();
                    if (result instanceof Future) {
                        return ((Future)result).get();
                    }
                } catch (ExecutionException var4) {
                    this.handleError(var4.getCause(), userDeclaredMethod, invocation.getArguments());
                } catch (Throwable var5) {
                    this.handleError(var5, userDeclaredMethod, invocation.getArguments());
                }

                return null;
            };
            return this.doSubmit(task, executor, invocation.getMethod().getReturnType());
        }
    }

源码很好理解就是创建了一个异步任务用来执行指定的方法，然后把它交给线程池去执行，这就是@Async注解实际上干的事了。

看到这里，你可能会有疑问，这看起来好像和上面的问题关系不大呀，别急，我们先回到AsyncAnnotationBeanPostProcessor注册的地方去看一下，源码地址PostProcessorRegistrationDelegate#registerBeanPostProcessors，到BeanPostProcessor pp = beanFactory.getBean(ppName, BeanPostProcessor.class);这边，也就是在创建AsyncAnnotationBeanPostProcessor这个对象并将它注册到ioc容器中区，然后后面初始化AsyncServiceImpl的时候就可以将已经创建的实例替换成带有通知的代理对象了，接着我们再往下debug一步，奇怪的事情发生了，源码跳到了咱们之前debug的AbstractAutowireCapableBeanFactory的doCreateBean这个方法的地方，并且此时创建的对象竟然是asyncServiceImpl

咱们想一想，明明这个时候还在创建AsyncAnnotationBeanPostProcessor，为什么中途来创建asyncServiceImpl了，而且如果这个时候就创建这个实体，当它实例化的时候，咱们的后置处理器还没有注册到beanFactory中去，这个时候放到代理对象里面肯定是不携带使异步生效的advisor了，@Async注解也就没有办法生效了，那么它到底为什么会走到这里呢？我们可以看一下方法的调用栈，发现在创建asyncServiceImpl之前先创建了baseConfig这个对象，根据咱们之前看到的源码，这样也是合理的，因为baseConfig中注入了asyncServiceImpl，所以在baseConfig实例化的时候也会先实例化asyncServiceImpl，这段源码在AbstractAutowireCapableBeanFactory的populateBean方法中，是依赖注入的地方，在bean初始化之前，有兴趣的可以看看，在往前看发现又创建了org.springframework.scheduling.annotation.ProxyAsyncConfiguration这个对象，这个对象好像是有点印象，它就是导入AsyncAnnotationBeanPostProcessor的类，其实在spring单例的初始化过程中确实会获取工厂bean来通过反射获取到通过@Bean注入的bean，详细代码可以看ConstructorResolver的instantiateUsingFactoryMethod方法，那么现在情况就比较明朗了，在创建AsyncAnnotationBeanPostProcessor的时候实例化了它的工厂类ProxyAsyncConfiguration，然后不知道怎么回事又实例化了baseConfig，又因为注入关系，所以创建了asyncServiceImpl， 那么现在问题就在于为什么创建ProxyAsyncConfiguration的时候会创建baseConfig对象呢？我们继续看调用栈，如下图

仔细看可以发现baseConfig是通过一个方法setConfigurersautowire进来的，我们打开ProxyAsyncConfiguration这个类的结构图看一下

可以很清楚的看到这个setConfigurers这个方法来自与它的父类AbstractAsyncConfiguration，我们点进去看一下

原来在这里setter注入了baseConfig，到现在问题原因就很明了了，就是因为连续的注入导致AsyncAnnotationBeanPostProcessor创建的时候创建了ProxyAsyncConfiguration，然后又创建了baseConfig，紧接着又创建了asyncServiceImpl，然后在asyncServiceImpl创建完实例初始化之后应用BeanPostProcessor的时候少掉了还尚未注册进容器的AsyncAnnotationBeanPostProcessor，因此使得@Async注解失效了。

  • 问题解决

知道了问题的原因，那么解决方法也就很明显了，我们秩序只需要将注入的asyncServiceImpl改为在方法内部通过容器获取即可

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
java复制代码@Slf4j
@Configuration
public class BaseConfig implements AsyncConfigurer, ApplicationContextAware {

    private static ApplicationContext applicationContext;

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(50);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("defaultExecutor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize(); 
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return ((Throwable var1, Method var2, Object... var3) -> {
            AsyncService bean = applicationContext.getBean(AsyncService.class);
            bean.Print();
            log.info("怎么又是你！");
        });
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        BaseConfig.applicationContext = applicationContext;
    }
}

执行

1
2
3
4
java复制代码2021-01-30 23:56:34.407  INFO 9304 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService
2021-01-30 23:56:34.420  INFO 9304 --- [faultExecutor-1] com.demo.service.impl.AsyncServiceImpl   : current thread id:[17], name:[defaultExecutor-1]
2021-01-30 23:56:34.424  INFO 9304 --- [faultExecutor-1] com.demo.service.impl.AsyncServiceImpl   : 打印。。。
2021-01-30 23:56:34.424  INFO 9304 --- [faultExecutor-1] com.demo.config.BaseConfig               : 怎么又是你！

完美！

本文转载自: 掘金

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

发表于

原创：小姐姐味道（微信公众号ID：xjjdog），欢迎分享，转载请保留出处。

很多接触Docker的同学，都接触过cgroup这个名词。它是Linux上的一项古老的技术，用来实现资源限制，比如CPU、内存等。但有很多同学反映，这项技术有点晦涩，不太好懂。

这就是本篇文章存在的目的，会让你以最简单直观的方式，了解cgroups到底是个什么东西。

接上上篇文章：《5分钟快速了解Docker的底层原理 | namespace篇》

cgroups，是实现docker功能的重要底层设施。如上图，使用cgroups，能够把操作系统的各项资源变成池子，然后通过配置获取相应的资源。

那它是怎么实现的呢？

要注意cgroups这个名词，它有两个特性。首先是c，就是Control的意思，是个动词；第二部分，就是groups，证明它是个组。

  1. 动词的目标

control，用来限制什么呢？除了CPU、内存，还有啥？

使用mount命令，查看当前系统支持的限制目标，它有个专用的名词，叫做子系统。

1
2
3
4
5
6
7
8
9
10
11
12
13
bash复制代码# mount  | grep cgroup
tmpfs on /sys/fs/cgroup type tmpfs (ro,nosuid,nodev,noexec,mode=755)
cgroup on /sys/fs/cgroup/systemd type cgroup (rw,nosuid,nodev,noexec,relatime,xattr,release_agent=/usr/lib/systemd/systemd-cgroups-agent,name=systemd)
cgroup on /sys/fs/cgroup/blkio type cgroup (rw,nosuid,nodev,noexec,relatime,blkio)
cgroup on /sys/fs/cgroup/devices type cgroup (rw,nosuid,nodev,noexec,relatime,devices)
cgroup on /sys/fs/cgroup/cpu,cpuacct type cgroup (rw,nosuid,nodev,noexec,relatime,cpuacct,cpu)
cgroup on /sys/fs/cgroup/net_cls,net_prio type cgroup (rw,nosuid,nodev,noexec,relatime,net_prio,net_cls)
cgroup on /sys/fs/cgroup/cpuset type cgroup (rw,nosuid,nodev,noexec,relatime,cpuset)
cgroup on /sys/fs/cgroup/memory type cgroup (rw,nosuid,nodev,noexec,relatime,memory)
cgroup on /sys/fs/cgroup/freezer type cgroup (rw,nosuid,nodev,noexec,relatime,freezer)
cgroup on /sys/fs/cgroup/pids type cgroup (rw,nosuid,nodev,noexec,relatime,pids)
cgroup on /sys/fs/cgroup/hugetlb type cgroup (rw,nosuid,nodev,noexec,relatime,hugetlb)
cgroup on /sys/fs/cgroup/perf_event type cgroup (rw,nosuid,nodev,noexec,relatime,perf_event)

不同的系统版本，会有一些细微的区别，大体上，子系统的分类包含下面这些：

  • cpu,cpuacct cpu主要限制进程的 cpu 使用率，cpuacct可以统计 cgroups 中的进程的 cpu 使用报告
  • cpuset 可以为 cgroups 中的进程分配单独的 cpu 节点或者内存节点，就像Numa做的那些事情一样
  • blkio 可以限制进程的块设备 io，比如物理设备（磁盘，固态硬盘，USB 等等）
  • devices 控制进程能够访问某些设备
  • net_cls 标记 cgroups 中进程的网络数据包，然后可以使用 tc 模块（traffic control）对数据包进行控制
  • net_prio — 这个子系统用来设计网络流量的优先级
  • freezer 可以挂起或者恢复 cgroups 中的进程。
  • ns 可以使不同 cgroups 下面的进程使用不同的 namespace
  • hugetlb 主要针对于HugeTLB系统进行限制，这是一个大页文件系统。

内容很多，但我们平常关注的大多数就是内存和CPU，这些繁杂的细节，不影响我们理解它的设计原则。

下面就以CPU为例，来看一下子系统的实际表现。

  1. CPU使用限制的例子

首先，我们进入cpu子系统目录。

1
bash复制代码cd /sys/fs/cgroup/cpu

然后，创建一个组名为xjjdog的cgroups，这个名字，就叫做控制组。

1
bash复制代码mkdir xjjdog

这时候，神奇的事情发生了。我们使用ll命令，查看xjjdog目录中的内容，发现系统已经为我们默认生成了一堆文件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
bash复制代码# ll xjjdog/
total 0
-rw-r--r-- 1 root root 0 Jan 28 21:09 cgroup.clone_children
--w--w--w- 1 root root 0 Jan 28 21:09 cgroup.event_control
-rw-r--r-- 1 root root 0 Jan 28 21:09 cgroup.procs
-r--r--r-- 1 root root 0 Jan 28 21:09 cpuacct.stat
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpuacct.usage
-r--r--r-- 1 root root 0 Jan 28 21:09 cpuacct.usage_percpu
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpu.cfs_period_us
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpu.cfs_quota_us
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpu.rt_period_us
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpu.rt_runtime_us
-rw-r--r-- 1 root root 0 Jan 28 21:09 cpu.shares
-r--r--r-- 1 root root 0 Jan 28 21:09 cpu.stat
-rw-r--r-- 1 root root 0 Jan 28 21:09 notify_on_release
-rw-r--r-- 1 root root 0 Jan 28 21:09 tasks

通过控制这些文件里面的数值，就可以对资源进行限制。比如cpu.cfs_quota_us文件，如果我们往里写入100000（十万），那么就证明使用了xjjdog的cgroup，最多能够使用1核的CPU。写入20000，证明最多使用使用1/5核的CPU。

这是因为，cpu.cfs_period_us这个配置文件，默认把1核cpu分成了10万份。

那我们就写入20000试一下。

1
bash复制代码sudo echo 20000 > xjjdog/cpu.cfs_quota_us

我们把当前shell的pid，加入被受控进程列表。

1
bash复制代码echo $$ > xjjdog/tasks

执行完毕之后，再启动一个死循环。

1
bash复制代码while true;do ;done;

重新打开一个shell，使用top观察CPU的使用率。可以发现，我们的死循环，最多只使用了20%的CPU。us保持在20%以下，且不间断的在各个cpu之间切换。

依次试验以下的命令，可以发现CPU的使用，会逐步增加，大体上和我们的限额是相等的。

1
2
3
bash复制代码sudo echo 40000 > xjjdog/cpu.cfs_quota_us
sudo echo 60000 > xjjdog/cpu.cfs_quota_us
sudo echo 100000 > xjjdog/cpu.cfs_quota_us

其他的资源限制，都是类似的思路。那么最重要的工作，就是需要知道cpu.cfs_quota_us这样的字眼，代表的是什么意思，这些对着手册来看是很容易掌握的。比如quota是配额的意思，很明显就是限制资源的使用。

如上图，子系统可以控制多个tasks，把它纳入到控制组之内。我们上篇文章讲到，可以将bash进程，作为docker系统的1号进程，那么同样的，这个1号进程的子进程，都会共享同样的限额配置。

  1. group的意思

浅显的来讲，group就是指的对各种资源进行分组。不同名字的资源，有不同的隔离配置。但它有更多的特性。

比较重要的，是它的层级关系（hierarchy）。这个也比较好理解，它主要是为了简化配置而存在的。

比如我上面的xjjdog目录，对cpu的限制限制在0.5核。这时候，我想要有另外一个应用，对cpu的使用限制在0.5核，同时限制内存1gb，那么就可以直接在xjjdog目录下创建xjjdog0目录，在xjjdog0目录下只配置内存方面的就可以了。

另外，如果你在外层的cpu限额限制了2core，然后在继承的目录里限制了1/5核，那它就只能使用操作系统的2/5核。这也是继承的一个特性。

End

cgroups是2006年诞生的，发起人是Google 的工程师（ Rohit Seth 和 Paul Menage ）。在 2008 年成功合入 Linux 2.6.24 版本中，可以说这项技术是很古老的。cgroups目前已经成为 systemd、Docker、Linux Containers（LXC） 等技术的基础。

像Windows平台的WSL，是没有cgroups功能的，使用mount命令可以验证，这证明了它是不能把docker跑起来的，因为缺乏基础。不过，WSL2已经可以了。

有些同学对docker目前的发展现状有些担心，但当你熟悉了这几个常见的底层原理，读完容器的标准之后，就会发现，上层的实现无论是换成docker也好，换成containerd也罢，都一样！

作者简介：小姐姐味道 (xjjdog)，一个不允许程序员走弯路的公众号。聚焦基础架构和Linux。十年架构，日百亿流量，与你探讨高并发世界，给你不一样的味道。我的个人微信xjjdog0，欢迎添加好友，​进一步交流。​

本文转载自: 掘金

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

发表于

分布式系统

在了解分布式架构之前，我们先来了解下分布式系统。按照维基百科的定义：分布式系统是一组电脑，透过网络相互连接传递消息与通信后并协调它们的行为而形成的系统。组件之间彼此进行交互以实现一个共同的目标。把需要进行大量计算的工程数据分割成小块，由多台计算机分别计算，再上传运算结果后，将结果统一合并得出数据结论的科学。

大概就是这样的一个架构，由多个组件之间彼此交付来完成对外的服务。

那这样的系统有什么好处呢？为什么需要这样的系统呢？

分布式系统前身

分布式系统前身就是集中式的系统，他有一个中心化的节点，可能是一台或者多台机器组成，所有的数据存储、计算都在主机上完成。集中式系统的优点：部署简单，可靠性高，数据一致性强等等。

随着客户量和交易量的不断增长，建立在大型主机上的集中式架构系统性能越来越吃紧，这个时候能采用的手段就是提升硬件配置，比如加内存、扩展磁盘、升级 CPU 等等。这样的扩展方式叫做横向扩展（scale up）

但是硬件并不能无条件的扩展，在扩展到某个点以后，扩展硬件对系统的提升将会非常有限。而且系统硬件的价格也随着水涨船高。随着计算机的发展，这样的架构越来越难适应人们的需求，比如说：

  1. 由于大型主机的复杂性，导致培养一个能够熟练运维大型主机的人的成本很高
  2. 大型主机很贵，一般只有土豪(政府、金融、电信)才能用得 起
  3. 单点问题，一台大型主机出现故障，那么整个系统将处于不 可用状态。而对于大型机的使用群体来说，这种不可用导致的 损失是非常大的
  4. 科技在进步，技术在进步。PC 机性能不断提升，很多企业 放弃大型机改用小型机及普通 PC 来搭建系统架构

分布式架构的意义

  1. 升级单机处理能力的性价比越来越低
    • 单机的处理能力主要依靠 CPU、内存、磁盘。通过更换硬件 做垂直扩展的方式来提升性能，成本会越来越高。
  2. 单机处理能力存在瓶颈
    • 单机处理能力存在瓶颈，CPU、内存都会有自己的性能瓶颈， 也就是说就算你是土豪不惜成本去提升硬件，但是硬件的发 展速度和性能是有限制的。
  3. 稳定性和可用性这两个指标很难达到
    • 单机系统存在可用性和稳定性的问题，这两个指标又是我们 必须要去解决的

分布式架构的演进

一个成熟的大型应用系统架构并不是一开始就设计得很完美的，也不是一开始就具备高性能、高可用、安全性等特性，而且随着业务数据量的增加而逐步完善的。在这个过程中，开发模式，技术架构都会发生比较大的变化。而针对不同业务的系统，会有各自的侧重点，比如电商平台的网站，要解决的是海量商品搜索、下单、支付等问题。通信软件需要解决的是数亿用户的实时消息传输。搜索引擎要解决的是海量数据的搜索。每一个种类的业务都有自己不同的系统架构。

这里通过一个 javaweb 电商应用来模拟一个架构的演变过程。这个模拟过程关注的是数据，访问量提升带来的结构变化，不关注具体的业务点。

假设我们的系统具有如下功能：

  • 用户功能：用户注册和管理。
  • 商品功能：商品展示和管理。
  • 交易功能：创建交易和支付结算。

单应用架构

网站的初期(也可以认为是互联网发展早期)，我们经常会在单机上部署我们的程序和软件。

把所有的应用都部署在同一台机器上，这样就完成了一个简单系统的搭建。

应用服务和数据库分离

随着网站的上线，访问量逐步加大，服务器的负载会越来越高。加入代码层面的优化已经没有办法提高了，在不提升硬件的情况下，增加机器是一个好的方法，投入产出比非常高。这个阶段主要讲的是数据库与应用服务分离，这样不仅提高了单机的负载能力，也提高了容灾能力。

应用服务器集群

继续随着访问量的不断提高。单台机器已经不能满足需求了。假设测试数据库还没有遇到瓶颈，我们可以按照上一阶段的方式继续增加机器。增加应用服务器，通过应用服务器将用户请求分流到各个服务器中，从而提升负载能力，测试多台应用服务器之前还没有直接的交互。

架构发展到这个阶段，各种问题开始慢慢浮现：

  • 用户的请求如何确认转发到那个应用服务器，由谁来做？
  • 用户访问到不同的应用服务器时，如何维护 session ？

读写分离

在把应用服务做集群后，把应用层的性能拉上来了。但是随着业务量的增加，这时候数据库成为了瓶颈。怎么去提高数据库层面的负载呢？有了上面的经验，会自然想到增加服务器。但是如果我们单纯增加一个数据库服务器，后续的请求分别负载到不同的数据库服务器上，那么一定会造成数据库数据不一致的问题(写请求也会被负载到不同的数据库服务器上)。所以这时候一般会先采用读写分离的方式。

发展到该阶段，遇到的问题大多是数据库的问题：

  • 如何保证数据库主从的数据同步；使用 mysql 自带的 master-slave 实现主从复制。
  • 如何选择对应的数据源；通过第三方插件，如 mycat，shardingjdbc；

引入缓存机制减轻数据库压力

随着访问量的提升，逐渐会出现很多用户访问同一个部分内容的情况。对这一部分热点数据，如果每次都去数据库查询，会给读库带来很大的压力。这时候可以引入环境机制，比如 redis。用来做应用的缓存，读请求时，可以先查看缓存中的数据，如果没有再去查询数据库。

使用搜索引擎来缓解数据库压力

数据库做读库的话，对模糊查找效率不是特别好，像电商类的 网站，搜索是非常核心的功能，即便是做了读写分离，这个问题也 不能有效解决。那么这个时候就需要引入搜索引擎了 使用搜索引擎能够大大提高我们的查询速度，但是同时也会带来一 些附加的问题，比如维护索引的构建。

数据库的水平/垂直拆分

到这个时候，用户，商品，交易的数据还在同一个库中，虽然采用了缓存及读写分离的形式。但是随着数据量的增加，数据库仍然是一个瓶颈，因此这里考虑把数据库进行拆分：

  • 垂直拆分：把数据库中不同的业务数据拆分到不同的库
  • 水平拆分：把同一个表中数据拆分为几个表。水平拆分的原因一般是某些业务量已经达到了单个数据库的瓶颈。

应用拆分

随着业务的发展，业务越来越复杂，应用的压力也越来越大。这个时候就可以考虑应用拆分，按照不同的功能拆分为多个子系统(上图中虽然明确标识出了用户、商品，但这只是为了表示 app 的功能，其实在这之前都是只有一个 app。)

到这里以后其实就是来到了微服务架构，但是这个架构中会有比较多的问题：

  • 服务拆分以后，多个服务之间如何通信？
  • 同一服务部署了多个节点，如何保证节点的配置一致性？
  • 服务拆分后，同一个请求可能会落在节点 A 及节点 B ，查看日志非常不方便。
  • 服务的定时任务，多个节点同时运行，如何保证任务不重复执行及漏执行？
  • 分布式的一致性问题。

这些问题都会在后续的文章中一一为大家解决。

总结

我们把集中式架构改造为分布式架构。是因为在业务发展的过程中，集中式的架构会越来越慢，且提升有较高的成本。稳定性及可用性很难保证。我们想要高性能及高可用的架构，因此将架构向分布式架构改造。

本文转载自: 掘金

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

发表于

01、前世今生

你好呀，我是 Guava。

1995 年的时候，我的“公明”哥哥——Java 出生了。经过 20 年的发展，他已经成为世界上最流行的编程语言了，请允许我有失公允的把“之一”给去了。

虽然他时常遭受着各种各样的吐槽，但他始终没有停下前进的脚步。除了他本身的不断进化，围绕着他的大大小小的兄弟们也在不断地更新迭代。我正是在这样的背景下应运而生的，我简单易用，对我大哥是一个非常好的补充，可以说，只要你有使用我哥作为开发语言的项目，几乎都能看到我的身影。

我由 Google 公司开源，目前在 GitHub 上已经有 39.9k 的铁粉了，由此可以证明我的受欢迎程度。

我的身体里主要包含有这些常用的模块：集合 [collections] 、缓存 [caching] 、原生类型支持 [primitives support] 、并发库 [concurrency libraries] 、通用注解 [common annotations] 、字符串处理 [string processing] 、I/O 等。新版的 JDK 中已经直接把我引入了，可想而知我有多优秀，忍不住骄傲了。

这么说吧，学好如何使用我，能让你在编程中变得更快乐，写出更优雅的代码！

02、引入 Guava

如果你要在 Maven 项目使用我的话，需要先在 pom.xml 文件中引入我的依赖。

1
2
3
4
5
xml复制代码<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>30.1-jre</version>
</dependency>

一点要求，JDK 版本需要在 8 以上。

03、基本工具

Doug Lea，java.util.concurrent 包的作者，曾说过一句话：“null 真糟糕”。Tony Hoare，图灵奖得主、快速排序算法的作者，当然也是 null 的创建者，也曾说过类似的话：“null 的使用，让我损失了十亿美元。”鉴于此，我用 Optional 来表示可能为 null 的对象。

代码示例如下所示。

1
2
3
java复制代码Optional<Integer> possible = Optional.of(5);
possible.isPresent(); // returns true
possible.get(); // returns 5

我大哥在 JDK 8 中新增了 Optional 类，显然是从我这借鉴过去的，不过他的和我的有些不同。

  • 我的 Optional 是 abstract 的，意味着我可以有子类对象；我大哥的是 final 的，意味着没有子类对象。
  • 我的 Optional 实现了 Serializable 接口，可以序列化；我大哥的没有。
  • 我的一些方法和我大哥的也不尽相同。

使用 Optional 除了赋予 null 语义，增加了可读性，最大的优点在于它是一种傻瓜式的防护。Optional 迫使你积极思考引用缺失的情况，因为你必须显式地从 Optional 获取引用。

除了 Optional 之外，我还提供了：

  • 参数校验
  • 常见的 Object 方法，比如说 Objects.equals、Objects.hashCode，JDK 7 引入的 Objects 类提供同样的方法，当然也是从我这借鉴的灵感。
  • 更强大的比较器

04、集合

首先我来说一下，为什么需要不可变集合。

  • 保证线程安全。在并发程序中，使用不可变集合既保证线程的安全性，也大大地增强了并发时的效率（跟并发锁方式相比）。
  • 如果一个对象不需要支持修改操作，不可变的集合将会节省空间和时间的开销。
  • 可以当作一个常量来对待，并且集合中的对象在以后也不会被改变。

与 JDK 中提供的不可变集合相比，我提供的 Immutable 才是真正的不可变，我为什么这么说呢？来看下面这个示例。

下面的代码利用 JDK 的 Collections.unmodifiableList(list) 得到一个不可修改的集合 unmodifiableList。

1
2
3
4
5
6
java复制代码List list = new ArrayList();
list.add("雷军");
list.add("乔布斯");

List unmodifiableList = Collections.unmodifiableList(list);
unmodifiableList.add("马云");

运行代码将会出现以下异常：

1
2
3
php复制代码Exception in thread "main" java.lang.UnsupportedOperationException
	at java.base/java.util.Collections$UnmodifiableCollection.add(Collections.java:1060)
	at com.itwanger.guava.NullTest.main(NullTest.java:29)

很好，执行 unmodifiableList.add() 的时候抛出了 UnsupportedOperationException 异常，说明 Collections.unmodifiableList() 返回了一个不可变集合。但真的是这样吗？

你可以把 unmodifiableList.add() 换成 list.add()。

1
2
3
4
5
6
java复制代码List list = new ArrayList();
list.add("雷军");
list.add("乔布斯");

List unmodifiableList = Collections.unmodifiableList(list);
list.add("马云");

再次执行的话，程序并没有报错，并且你会发现 unmodifiableList 中真的多了一个元素。说明什么呢？

Collections.unmodifiableList(…) 实现的不是真正的不可变集合，当原始集合被修改后，不可变集合里面的元素也是跟着发生变化。

我就不会犯这种错，来看下面的代码。

1
2
3
java复制代码List<String> stringArrayList = Lists.newArrayList("雷军","乔布斯");
ImmutableList<String> immutableList = ImmutableList.copyOf(stringArrayList);
immutableList.add("马云");

尝试 immutableList.add() 的时候会抛出 UnsupportedOperationException。我在源码中已经把 add() 方法废弃了。

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码  /**
   * Guaranteed to throw an exception and leave the collection unmodified.
   *
   * @throws UnsupportedOperationException always
   * @deprecated Unsupported operation.
   */
  @CanIgnoreReturnValue
  @Deprecated
  @Override
  public final boolean add(E e) {
    throw new UnsupportedOperationException();
  }

尝试 stringArrayList.add() 修改原集合的时候 immutableList 并不会因此而发生改变。

除了不可变集合以外，我还提供了新的集合类型，比如说：

  • Multiset，可以多次添加相等的元素。当把 Multiset 看成普通的 Collection 时，它表现得就像无序的 ArrayList；当把 Multiset 看作 Map<E, Integer> 时，它也提供了符合性能期望的查询操作。
  • Multimap，可以很容易地把一个键映射到多个值。
  • BiMap，一种特殊的 Map，可以用 inverse() 反转
    BiMap<K, V> 的键值映射；保证值是唯一的，因此 values() 返回 Set 而不是普通的 Collection。

05、字符串处理

字符串表示字符的不可变序列，创建后就不能更改。在我们日常的工作中，字符串的使用非常频繁，熟练的对其操作可以极大的提升我们的工作效率。

我提供了连接器——Joiner，可以用分隔符把字符串序列连接起来。下面的代码将会返回“雷军; 乔布斯”，你可以使用 useForNull(String) 方法用某个字符串来替换 null，而不像 skipNulls() 方法那样直接忽略 null。

1
2
java复制代码Joiner joiner = Joiner.on("; ").skipNulls();
return joiner.join("雷军", null, "乔布斯");

我还提供了拆分器—— Splitter，可以按照指定的分隔符把字符串序列进行拆分。

1
2
3
4
java复制代码Splitter.on(',')
        .trimResults()
        .omitEmptyStrings()
        .split("雷军,乔布斯,,   沉默王二");

06、缓存

缓存在很多场景下都是相当有用的。你应该知道，检索一个值的代价很高，尤其是需要不止一次获取值的时候，就应当考虑使用缓存。

我提供的 Cache 和 ConcurrentMap 很相似，但也不完全一样。最基本的区别是 ConcurrentMap 会一直保存所有添加的元素，直到显式地移除。相对地，我提供的 Cache 为了限制内存占用，通常都设定为自动回收元素。

如果你愿意消耗一些内存空间来提升速度，你能预料到某些键会被查询一次以上，缓存中存放的数据总量不会超出内存容量，就可以使用 Cache。

来个示例你感受下吧。

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
java复制代码@Test
public void testCache() throws ExecutionException, InterruptedException {

    CacheLoader cacheLoader = new CacheLoader<String, Animal>() {
        // 如果找不到元素，会调用这里
        @Override
        public Animal load(String s) {
            return null;
        }
    };
    LoadingCache<String, Animal> loadingCache = CacheBuilder.newBuilder()
        .maximumSize(1000) // 容量
        .expireAfterWrite(3, TimeUnit.SECONDS) // 过期时间
        .removalListener(new MyRemovalListener()) // 失效监听器
        .build(cacheLoader); //
    loadingCache.put("狗", new Animal("旺财", 1));
    loadingCache.put("猫", new Animal("汤姆", 3));
    loadingCache.put("狼", new Animal("灰太狼", 4));

    loadingCache.invalidate("猫"); // 手动失效

    Animal animal = loadingCache.get("狼");
    System.out.println(animal);
    Thread.sleep(4 * 1000);
    // 狼已经自动过去，获取为 null 值报错
    System.out.println(loadingCache.get("狼"));
}

/**
 * 缓存移除监听器
 */
class MyRemovalListener implements RemovalListener<String, Animal> {

    @Override
    public void onRemoval(RemovalNotification<String, Animal> notification) {
        String reason = String.format("key=%s,value=%s,reason=%s", notification.getKey(), notification.getValue(), notification.getCause());
        System.out.println(reason);
    }
}

class Animal {
    private String name;
    private Integer age;

    public Animal(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
}

CacheLoader 中重写了 load 方法，这个方法会在查询缓存没有命中时被调用，我这里直接返回了 null，其实这样会在没有命中时抛出 CacheLoader returned null for key 异常信息。

MyRemovalListener 作为缓存元素失效时的监听类，在有元素缓存失效时会自动调用 onRemoval 方法，这里需要注意的是这个方法是同步方法，如果这里耗时较长，会阻塞直到处理完成。

LoadingCache 就是缓存的主要操作对象了，常用的就是其中的 put 和 get 方法了。

07、尾声

上面介绍了我认为最常用的功能，作为 Google 公司开源的 Java 开发核心库，个人觉得实用性还是很高的（不然呢？嘿嘿嘿）。引入到你的项目后不仅能快速的实现一些开发中常用的功能，而且还可以让代码更加的优雅简洁。

我觉得适用于每一个 Java 项目，至于其他的一些功能，比如说散列、事件总线、数学运算、反射，就等待你去发掘了。

另外，我为你准备了一份中文版的官方文档（部分目录见上图），一共 190 页，写得非常详细，强烈建议你通过下面的方式下载一份放在桌面上！

链接:pan.baidu.com/s/1hrfaArC3… 密码:684a

推荐阅读：

自学java，学多久可以自己找到工作？

看完谷歌学长的刷题笔记，我决定 2021 年手撕这101道 Leetcode 算法题

如果内容对你有所帮助的话，请给我点个赞吧，笔芯~，你最美你最帅，你 2021 年发大财。

本文转载自: 掘金

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

发表于

引言

随着互联网应用体量不断的增加，为了对应大量用户访问的体验效果，及提供应用的可用性，微服务应运而生，微服务使应用的业务具有更高的内聚性，同时对整体应用进行服务化的解耦；微服务之间除了通过异步的MQ进行通信，常用的还有基于同步的RPC通知机制，比如基于HTTP的GRPC, BRPC及在国内使用比较广泛的Dubbo。一致在使用DUBOO，对其中的功能组件有大致的了解，本计划看一下源码的，一致没有抽出时间，最近终于忙里抽闲，一探Dubbo芳容。

目录

概要框架设计

dubbo框架主要包括序列化，消息层，传输层，协议层。序列化主要是请求消息和响应消息的序列化，比如基于Javad的ObjectOut/InputStream序列化、基于JSON的序列化。消息层提供消费者调用服务请求消息、服务提供方处理
结果响应消息的编解码；传输层主要建立消费者和服务者的通信通道，传输服务请求响应数据，比如基于Netty和Mina的，默认为Netty；协议层首先是基于相关协议将服提供者，和消费者通过export暴露出去，即注册器Registry中，消费者通过Registry订阅响应的服务提供者，消费者发现有服务
提供者，则与服务提供者建立连接，注册协议有基于Zookeeper，Redis等，在注册协议中还有一个注册器目录服务，用于提供消费者和服务者列表，及根据负载均衡策略选择服务者。服务提供者接受的消费者的服务请求后，根据相关协议，调用相应的Invoker服务。 消费者和服务者的RPC调用协议，实际在DubboProtocol中，协议首先导出服务，消费者发送RPC请求，调用Exporter服务容器中的Invoker。

源码分析

Dubbo框架设计源码解读第一篇（服务和引用bean初始化）

dubbo框架主要包括消息层，传输层，协议层。消息层提供消费者调用服务请求消息、服务提供方处理
结果响应消息的编解码；传输层主要建立消费者和服务者的通信通道，传输服务请求响应数据；协议层首先是基于相关协议将服
提供者，和消费者通过export暴露出去，即注册的Registry中，消费者通过Registry订阅响应的服务提供者，消费者发现有服务
提供者，则与服务提供者建立连接。服务提供者接受的消费者的服务请求后，根据相关协议，调用响应的Invoker服务。

ServiceAnnotationBeanPostProcessor主要做的事情是扫描 应用先的Service注解bean，并构造ServiceBean，注册到bean注册器中。导出服务实际委托给相应的协议RegistryProtocol

ReferenceBean后处理主要扫描Reference注解的bean，并构造ReferenceBean，ReferenceBean通过Invoker去调用服务提供者的服务，Invoker为服务的包装类。实际通过相应的协议创建。

应用协议

注册器协议

Dubbo框架设计源码解读二（注册器，服务注册，订阅）

注册协议，导出服务主要有注册服务和订阅服务。注册服务，实际是将基于Dubbo协议的服务URL写到ZK上，如何在注册的过程中，由于Dubbo自身机制导致的注册失败，将加入的失败注册集，并有定时钟，进行重试注册。订阅服务，监听服务提供者的节点路径。消费者注册到ZK上的订阅服务节点上，具体的订阅委托给目录服务。

注册目录服务依赖于注册器，消费者从注册器获取服务提供者，实际为从注册目录服务获取服务列表（zk注册器为，服务节点下的提供者），并根据路由策略，选择可用的服务提供者Invoker。注册器目录处理提供服务路由，同时监听服务的变化。如果注册器节点信息存在变化，则重新刷新服务，建立服务Invoker索引。

Dubbo协议

Dubbo框架设计源码解读三（Dubbo协议，服务导出，引用）

Dubbo协议是消费者和服务者通信的基础，包括服务的调用。注册器协议中，有如下一个功能，导出服务到本地， 实际委托的相应的协议，比如Dubbo协议DubboProtocol的export操作。注册器目录，当前监听注册器节点变化是，重新索引服务，在转换URL为Invoker，实际委托的相应的协议，比如Dubbo协议DubboProtocol的 refer操作。 dubbo协议的导出服务，实际上创建一个服务Server，根据dubbo协议配置，可为NettyServer，或MinaServer。默认为NettyServer。

数据传输器Transport

Dubbo框架设计源码解读四(Dubbo基于Netty的传输器Transport)

Netty服务端是基于经典的bootstrap，事件，worker, 编解码器，消息处理器的实现。netty处理器实际为一个共享的SimpleChannelHandler， 所有操作委托内部通道处理器DecodeHandler。DecodeHandler在接受消息后，解码相应的消息，将交由内部的HeaderExchangeHandler处理。HeaderExchangeHandler， 所有的操作委托给内部的ExchangeHander，实际为DubboProtocol的中ExchangeHandleAdater。

消费者调用服务提供者实际上发送的一个Invocation消息，服务端接受到消息，根据Invoker上下文，从Dubbo协议的Exporter容器中获取对应的Invoker，调用相关服务，将调用结果返回给消费者。

netty客户端也没有多少新鲜的动心，编解码器，Netty客户端处理器NettyClientHandler。NettyClientHandler实际为一个共享的ChannelDuplexHandler，所有操作委托内部通道处理器DecodeHandler。DecodeHandler在接受消息后，解码相应的消息，将交由内部的HeaderExchangeHandler处理。HeanderExchangeHandler，HeanderExchangeHandler所有的操作委托给内部的ExchangeHander，实际为DubboProtocol的中ExchangeHandleAdater。一个请求，主要包括请求Id，版本，及数据及Invocation。服务响应，主要包含消息id，消息版本，状态，相应结果，及错误信息，如果有的话。

消息编解码

Dubbo框架设计源码解读五(消息编解码器)

NettyCodecAdapter为编解码器的适配，内部编码器实际为ByteToMessageDecoder， 内部的编码操作委托给内部编解码器，根据SPI机制，实际为DubboCodec；内部解码器实际为ByteToMessageDecoder，解码操作委托给DubboCodec。

编码器编码主要是针对请求Request和响应Response，编码首先写头部，针对请求主要有魔数、序列化标志，请求id，然后是请求数据,主要有RpcInvocation的服务URL，版本信息，服务方法，方法参数类型，方法参数类型通过ReflectUtils进行编码，最后还有方法参数；针对响应，首先写头部，魔数，请求id，然后是响应数据。需要注意在编码请求和响应的时候，有一部分是Attament，这部分是可以扩展的地方。序列化器，有FastJsonSerialization，JavaSerialization, 默认为JavaSerialization。JavaSerialization的序列化和反序列化实际是基于ObjectOutput/InputStream。

解码消息首先解码消息头部魔数等数据，然后是请求id，如果是消费者请求则解码出消息体，实际为DecodeableRpcInvocation，包括方面名，参数以及参数值；如果是请求响应，解码出返回结果，实际为DecodeableRpcResult，包含服务响应数据。

附

Dubbo类图

参考文献

dubbo offical site

dubbo github

dubbo github vt

incubator-dubbo-spring-boot-project github vt

本文转载自: 掘金

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

发表于

环境准备

服务器名

IP

说明

MySQL

192.168.223.135

部署MySQL数据库和Nginx

Nacos

192.168.223.137

部署Nacos集群

资源有限，MySQL 部署了一台机器，Nginx 和 Nacos 集群部署在了另一台机器。如果在生产环境部署，可以按照自己的需求调整。

配置步骤

下载地址：github.com/alibaba/nac…

将压缩包拷贝到对应部署 Nacos 的机器上

  1. MySQL 数据库配置

MySQL安装教程

安装好 MySQL 以后，需要初始化 MySQL 数据库，数据库初始化文件在压缩包 conf 文件下的 nacos-mysql.sql，在对应的数据库环境下导入 SQL 文件

1
2
3
4
5
shell复制代码# 进入MySQL终端
mysql -u root -p123456
mysql> create database nacos_config;
mysql> use nacos_config;
mysql> source /root/nacos-mysql.sql
  1. application.properties 配置

在 nacos 的解压目录 nacos/ 的 conf 目录下，有配置文件 application.properties，修改 conf/application.properties 文件，增加支持 MySQL 数据源配置

1
2
3
4
5
6
ini复制代码spring.datasource.platform=mysql

db.num=1
db.url.0=jdbc:mysql://192.168.223.135:3306/nacos_config?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true
db.user=root
db.password=123456
  1. 配置集群配置文件

在 nacos 的解压目录 nacos/ 的 conf 目录下，有配置文件 cluster.conf，请每行配置成ip:port。（请配置3个或3个以上节点）

1
2
bash复制代码cp cluster.conf.example cluster.conf
vim cluster.conf

image.png

  1. 编辑 Nacos 的启动脚本 startup.sh，使它能够接受不同的启动端口

修改前

image.png

修改后

image.png

  1. 配置 Nginx 作为负载均衡器

Nginx安装教程

在 nginx.conf 文件#gzip on;下方添加如下内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
ini复制代码upstream cluster {
    server 192.168.223.137:3333;
    server 192.168.223.137:4444;
    server 192.168.223.137:5555;
}
server {
    listen       1111;
    server_name  localhost;
    location / {
        #root front;
        #index index.htm;
        proxy_pass http://cluster;
    }
}
  1. 启动测试

启动nacos集群

sh startup.sh -p 3333
sh startup.sh -p 4444
sh startup.sh -p 5555
ps -ef | grep nacos | grep -v grep | wc -l

启动nginx

/usr/local/nginx/sbin/nginx
ps -ef | grep nginx

浏览器访问

http://192.168.223.135:1111/nacos

image.pngimage.png

新增一个配置进行测试查看是否存入数据库

image.png

image.png

在 nacos-spring-cloud-provider-example 中将 application.properties 中服务注册的地址修改为 spring.cloud.nacos.discovery.server-addr=192.168.223.135:1111 进行测试

image.png

本文转载自: 掘金

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

0%