严格来说，本文并不算是 Spring Boot 中的知识点，但是很多学过 SpringMVC 的小伙伴，对于 @ControllerAdvice 却并不熟悉，Spring Boot 和 SpringMVC 一脉相承，@ControllerAdvice 在 Spring Boot 中也有广泛的使用场景，因此本文我们就来聊一聊这个问题。

@ControllerAdvice ，很多初学者可能都没有听说过这个注解，实际上，这是一个非常有用的注解，顾名思义，这是一个增强的 Controller。使用这个 Controller ，可以实现三个方面的功能：

1. 全局异常处理
2. 全局数据绑定
3. 全局数据预处理

灵活使用这三个功能，可以帮助我们简化很多工作，需要注意的是，这是 SpringMVC 提供的功能，在 Spring Boot 中可以直接使用，下面分别来看。

全局异常处理

使用 @ControllerAdvice 实现全局异常处理，只需要定义类，添加该注解即可定义方式如下：

1
2
3
4
5
6
7
8
9
10

复制代码@ControllerAdvice
public class MyGlobalExceptionHandler {
    @ExceptionHandler(Exception.class)
    public ModelAndView customException(Exception e) {
        ModelAndView mv = new ModelAndView();
        mv.addObject("message", e.getMessage());
        mv.setViewName("myerror");
        return mv;
    }
}

在该类中，可以定义多个方法，不同的方法处理不同的异常，例如专门处理空指针的方法、专门处理数组越界的方法…，也可以直接向上面代码一样，在一个方法中处理所有的异常信息。

@ExceptionHandler 注解用来指明异常的处理类型，即如果这里指定为 NullpointerException，则数组越界异常就不会进到这个方法中来。

全局数据绑定

全局数据绑定功能可以用来做一些初始化的数据操作，我们可以将一些公共的数据定义在添加了 @ControllerAdvice 注解的类中，这样，在每一个 Controller 的接口中，就都能够访问导致这些数据。

使用步骤，首先定义全局数据，如下：

1
2
3
4
5
6
7
8
9
10

复制代码@ControllerAdvice
public class MyGlobalExceptionHandler {
    @ModelAttribute(name = "md")
    public Map<String,Object> mydata() {
        HashMap<String, Object> map = new HashMap<>();
        map.put("age", 99);
        map.put("gender", "男");
        return map;
    }
}

使用 @ModelAttribute 注解标记该方法的返回数据是一个全局数据，默认情况下，这个全局数据的 key 就是返回的变量名，value 就是方法返回值，当然开发者可以通过 @ModelAttribute 注解的 name 属性去重新指定 key。

定义完成后，在任何一个Controller 的接口中，都可以获取到这里定义的数据：

1
2
3
4
5
6
7
8
9
10

复制代码@RestController
public class HelloController {
    @GetMapping("/hello")
    public String hello(Model model) {
        Map<String, Object> map = model.asMap();
        System.out.println(map);
        int i = 1 / 0;
        return "hello controller advice";
    }
}

全局数据预处理

考虑我有两个实体类，Book 和 Author，分别定义如下：

1
2
3
4
5
6
7
8
9
10

复制代码public class Book {
    private String name;
    private Long price;
    //getter/setter
}
public class Author {
    private String name;
    private Integer age;
    //getter/setter
}

此时，如果我定义一个数据添加接口，如下：

1
2
3
4
5

复制代码@PostMapping("/book")
public void addBook(Book book, Author author) {
    System.out.println(book);
    System.out.println(author);
}

这个时候，添加操作就会有问题，因为两个实体类都有一个 name 属性，从前端传递时 ，无法区分。此时，通过 @ControllerAdvice 的全局数据预处理可以解决这个问题

解决步骤如下:

1.给接口中的变量取别名

1
2
3
4
5

复制代码@PostMapping("/book")
public void addBook(@ModelAttribute("b") Book book, @ModelAttribute("a") Author author) {
    System.out.println(book);
    System.out.println(author);
}

2.进行请求数据预处理

在 @ControllerAdvice 标记的类中添加如下代码:

1
2
3
4
5
6
7
8

复制代码@InitBinder("b")
public void b(WebDataBinder binder) {
    binder.setFieldDefaultPrefix("b.");
}
@InitBinder("a")
public void a(WebDataBinder binder) {
    binder.setFieldDefaultPrefix("a.");
}

@InitBinder(“b”) 注解表示该方法用来处理和Book和相关的参数,在方法中,给参数添加一个 b 前缀,即请求参数要有b前缀.

3.发送请求

请求发送时,通过给不同对象的参数添加不同的前缀,可以实现参数的区分.

总结

这就是松哥给大伙介绍的 @ControllerAdvice 的几个简单用法，这些点既可以在传统的 SSM 项目中使用，也可以在 Spring Boot + Spring Cloud 微服务中使用，欢迎大家有问题一起讨论。

关注公众号【江南一点雨】，专注于 Spring Boot+微服务以及前后端分离等全栈技术，定期视频教程分享，关注后回复 Java ，领取松哥为你精心准备的 Java 干货！

本文转载自: 掘金

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

作者 | 晓土 阿里巴巴高级工程师

姊妹篇阅读推荐：《云原生时代，分布式系统设计必备知识图谱（内含22个知识点）》

导读：本文力求从分布式基础理论、架构设计模式、工程应用、部署运维、业界方案这几大方面，介绍基于 MSA（微服务架构）的分布式知识体系大纲，从而对 SOA 到 MSA 进化有着立体的认识；从概念上和工具应用上更近一步了解微服务分布式的本质，身临其境的感受如何搭建全套微服务架构的过程。

关注“阿里巴巴云原生”公众号，回复“分布”，即可下载分布式系统及其知识体系清晰大图！

随着移动互联网的发展和智能终端的普及，计算机系统早就从单机独立工作过渡到多机器协作，集群按照分布式理论构建出庞大复杂的应用服务，在分布式的基础上正进行一场云原生的技术革命，彻底打破传统的开发方式，解放了新一代的生产力。

分布式系统知识体系大图

关注“阿里巴巴云原生”公众号，回复“分布”，即可下载分布式系统及其知识体系清晰大图！

基础理论

SOA 到 MSA 的进化

SOA 面向服务架构

由于业务发展到一定程度后，需要对服务进行解耦，进而把一个单一的大系统按逻辑拆分成不同的子系统，通过服务接口来通讯。面向服务的设计模式，最终需要总线集成服务，而且大部分时候还共享数据库，出现单点故障时会导致总线层面的故障，更进一步可能会把数据库拖垮，所以才有了更加独立的设计方案的出现。

MSA 微服务架构

微服务是真正意义上的独立服务，从服务入口到数据持久层，逻辑上都是独立隔离的，无需服务总线来接入，但同时也增加了整个分布式系统的搭建和管理难度，需要对服务进行编排和管理，所以伴随着微服务的兴起，微服务生态的整套技术栈也需要无缝接入，才能支撑起微服务的治理理念。

节点与网络

节点

传统的节点也就是一台单体的物理机，所有的服务都揉进去包括服务和数据库；随着虚拟化的发展，单台物理机往往可以分成多台虚拟机，实现资源利用的最大化，节点的概念也变成单台虚拟机上面服务；近几年容器技术逐渐成熟后，服务已经彻底容器化，也就是节点只是轻量级的容器服务。总体来说，节点就是能提供单位服务的逻辑计算资源的集合。

网络

分布式架构的根基就是网络，不管是局域网还是公网，没有网络就无法把计算机联合在一起工作，但是网络也带来了一系列的问题。网络消息的传播有先后，消息丢失和延迟是经常发生的事情，我们定义了三种网络工作模式：

• 同步网络
  • 节点同步执行
  • 消息延迟有限
  • 高效全局锁
• 半同步网络
  • 锁范围放宽
• 异步网络
  • 节点独立执行
  • 消息延迟无上限
  • 无全局锁
  • 部分算法不可行

常用网络传输层有两大协议的特点简介：

• TCP 协议
  • 首先 tcp 协议传输可靠，尽管其他的协议可以更快传输
  • tcp 解决重复和乱序问题
• UDP 协议
  • 常量数据流
  • 丢包不致命

时间与顺序

时间

慢速物理时空中，时间独自在流淌着，对于串行的事务来说，很简单的就是跟着时间的脚步走就可以，先来后到的发生。而后我们发明了时钟来刻画以往发生的时间点，时钟让这个世界井然有序。但是对于分布式世界来说，跟时间打交道着实是一件痛苦的事情。

分布式世界里面，我们要协调不同节点之间的先来后到关系，不同节点本身承认的时间又各执己见，于是我们创造了网络时间协议（NTP）试图来解决不同节点之间的标准时间，但是 NTP 本身表现并不尽如人意，所以我们又构造出了逻辑时钟，最后改进为向量时钟：

• NTP 的一些缺点，无法完全满足分布式下并发任务的协调问题
  • 节点间时间不同步
  • 硬件时钟漂移
  • 线程可能休眠
  • 操作系统休眠
  • 硬件休眠

• 逻辑时钟
  • 定义事件先来后到
  • t’ = max(t, t_msg + 1)

• 向量时钟
  • t_i’ = max(t_i, t_msg_i)
• 原子钟

顺序

有了衡量时间的工具，解决顺序问题自然就是水到渠成了。因为整个分布式的理论基础就是如何协商不同节点的一致性问题，而顺序则是一致性理论的基本概念，所以前文我们才需要花时间介绍衡量时间的刻度和工具。

一致性理论

说到一致性理论，我们必须看一张关于一致性强弱对系统建设影响的对比图：

该图对比了不同一致性算法下的事务、性能、错误、延迟的平衡。

强一致性 ACID

单机环境下我们对传统关系型数据库有苛刻的要求，由于存在网络的延迟和消息丢失，ACID 便是保证事务的原则，这四大原则甚至我们都不需要解释出来就耳熟能详了：

• Atomicity：原子性，一个事务中的所有操作，要么全部完成，要么全部不完成，不会结束在中间某个环节；
• Consistency：一致性，在事务开始之前和事务结束以后，数据库的完整性没有被破坏；
• Isolation：隔离性，数据库允许多个并发事务同时对其数据进行读写和修改的能力，隔离性可以防止多个事务并发执行时，由于交叉执行而导致数据的不一致；
• Durabilit：事务处理结束后，对数据的修改就是永久的，即便系统故障也不会丢失。

分布式一致性 CAP

分布式环境下，我们无法保证网络的正常连接和信息的传送，于是发展出了 CAP/FLP/DLS 这三个重要的理论：

• CAP：分布式计算系统不可能同时确保一致性（Consistency）、可用性（Availablity）和分区容忍性（Partition）；
• FLP：在异步环境中，如果节点间的网络延迟没有上限，只要有一个恶意的节点存在，就没有算法能在有限的时间内达成共识；
• DLS：
  • 在一个部分同步网络的模型（也就是说：网络延时有界限但是我们并不知道在哪里）下运行的协议可以容忍 1/3 任意（换句话说，拜占庭）错误；
  • 在一个异步模型中的确定性的协议（没有网络延时上限）不能容错（不过这个论文没有提起随机化算法可以容忍 1/3 的错误）；
  • 同步模型中的协议（网络延时可以保证小于已知 d 时间），可以令人吃惊的达到 100% 容错，虽然对 1/2 的节点出错可以发生的情况有所限制。

弱一致性 BASE

多数情况下，其实我们也并非一定要求强一致性，部分业务可以容忍一定程度的延迟一致，所以为了兼顾效率，发展出来了最终一致性理论 BASE。BASE 是指基本可用（Basically Available）、软状态（ Soft State）、最终一致性（ Eventual Consistency）：

• 基本可用（Basically Available）：基本可用是指分布式系统在出现故障的时候，允许损失部分可用性，即保证核心可用；
• 软状态（Soft State）：软状态是指允许系统存在中间状态，而该中间状态不会影响系统整体可用性。分布式存储中一般一份数据至少会有三个副本，允许不同节点间副本同步的延时就是软状态的体现；
• 最终一致性（Eventual Consistency）：最终一致性是指系统中的所有数据副本经过一定时间后，最终能够达到一致的状态。弱一致性和强一致性相反，最终一致性是弱一致性的一种特殊情况。

一致性算法

分布式架构的核心就在于一致性的实现和妥协，那么如何设计一套算法来保证不同节点之间的通信和数据达到无限趋向一致性，就非常重要了。保证不同节点在充满不确定性网络环境下能达成相同副本的一致性是非常困难的，业界对该课题也做了大量的研究。

首先我们要了解一致性的大前提原则 (CALM):
CALM 原则的全称是 Consistency and Logical Monotonicity ，主要描述的是分布式系统中单调逻辑与一致性的关系，它的内容如下，参考 consistency as logical monotonicity。

• 在分布式系统中，单调的逻辑都能保证 “最终一致性”，这个过程中不需要依赖中心节点的调度；
• 任意分布式系统，如果所有的非单调逻辑都有中心节点调度，那么这个分布式系统就可以实现最终“一致性”。

然后再关注分布式系统的数据结构 CRDT(Conflict-Free Replicated Data Types)：
我们了解到分布式一些规律原则之后，就要着手考虑如何来实现解决方案，一致性算法的前提是数据结构，或者说一切算法的根基都是数据结构，设计良好的数据结构加上精妙的算法可以高效的解决现实的问题。经过前人不断的探索，我们得知分布式系统被广泛采用的数据结构 CRDT。
参考《谈谈 CRDT》,A comprehensive study of Convergent and Commutative Replicated Data Types

• 基于状态（state-based）：即将各个节点之间的 CRDT 数据直接进行合并，所有节点都能最终合并到同一个状态，数据合并的顺序不会影响到最终的结果；
• 基于操作（operation-based）：将每一次对数据的操作通知给其他节点。只要节点知道了对数据的所有操作（收到操作的顺序可以是任意的），就能合并到同一个状态。

了解数据结构后，我们需要来关注一下分布式系统的一些重要的***协议***HATs(Highly Available Transactions)，ZAB(Zookeeper Atomic Broadcast)：
参考《高可用事务》，《ZAB 协议分析》

最后要学习的是业界主流的一致性算法 ：
说实话具体的算法我也还没完全搞懂，一致性算法是分布式系统最核心本质的内容，这部分的发展也会影响架构的革新，不同场景的应用也催生不同的算法。

• Paxos：《优雅的Paxos算法》
• Raft ：《Raft 一致性算法》
• Gossip：《Gossip Visualization》

这一节我们说完分布式系统里面核心理论基础，如何达成不同节点之间的数据一致性，下面我们将会讲到目前都有哪些主流的分布式系统。

场景分类

文件系统

单台计算机的存储始终有上限，随着网络的出现，多台计算机协作存储文件的方案也相继被提出来。最早的分布式文件系统其实也称为网络文件系统，第一个文件服务器在 1970 年代被发展出来。在 1976 年迪吉多公司设计出 File Access Listener（FAL），而现代分布式文件系统则出自赫赫有名的 Google 的论文，《The Google File System》奠定了分布式文件系统的基础。现代主流分布式文件系统参考《分布式文件系统对比》,下面列举几个常用的文件系统：

• HDFS
• FastDFS
• Ceph
• mooseFS

数据库

数据库当然也属于文件系统，主数据增加了事务、检索、擦除等高级特性，所以复杂度又增加了，既要考虑数据一致性也得保证足够的性能。传统关系型数据库为了兼顾事务和性能的特性，在分布式方面的发展有限，非关系型数据库摆脱了事务的强一致性束缚，达到了最终一致性的效果，从而有了飞跃的发展，NoSql(Not Only Sql) 也产生了多个架构的数据库类型，包括 KV、列式存储、文档类型等。

• 列式存储：Hbase
• 文档存储：Elasticsearch，MongoDB
• KV 类型：Redis
• 关系型：Spanner

计算

分布式计算系统构建在分布式存储的基础上，充分发挥分布式系统的数据冗余灾备，多副本高效获取数据的特性，进而并行计算，把原本需要长时间计算的任务拆分成多个任务并行处理，从而提高了计算效率。分布式计算系统在场景上分为离线计算、实时计算和流式计算。

• 离线：Hadoop
• 实时：Spark
• 流式：Storm，Flink/Blink

缓存

缓存作为提升性能的利器无处不在，小到 CPU 缓存架构，大到分布式应用存储。分布式缓存系统提供了热点数据的随机访问机制，大大了提升了访问时间，但是带来的问题是如何保证数据的一致性，引入分布式锁来解决这个问题，主流的分布式存储系统基本就是 Redis 了。

• 持久化：Redis
• 非持久化：Memcache

消息

分布式消息队列系统是消除异步带来的一系列复杂步骤的一大利器，在多线程高并发场景下，我们常常需要谨慎设计业务代码，来保证多线程并发情况下不出现资源竞争导致的死锁问题。而消息队列以一种延迟消费的模式将异步任务都存到队列，然后再逐个消化。

• Kafka
• RabbitMQ
• RocketMQ
• ActiveMQ

监控

分布式系统从单机到集群的形态发展，复杂度也大大提高，所以对整个系统的监控也是必不可少。

• Zookeeper

应用

分布式系统的核心模块就是在应用如何处理业务逻辑，应用直接的调用依赖于特定的协议来通信，有基于 RPC 协议的，也有基于通用的 HTTP 协议。

• HSF
• Dubbo

日志

错误对应分布式系统是家常便饭，而且我们设计系统的时候，本身就需要把容错作为普遍存在的现象来考虑。那么当出现故障的时候，快速恢复和排查故障就显得非常重要了。分布式日志采集存储和检索则可以给我们提供有力的工具来定位请求链路中出现问题的环节。

• 日志采集：flume
• 日志存储：ElasticSearch/Solr，SLS
• 日志定位：Zipkin

账本

前文我们提到所谓分布式系统，是迫于单机的性能有限，而堆硬件却又无法无休止的增加，单机堆硬件最终也会遇到性能增长曲线的瓶颈。于是我们才采用了多台计算机来干同样的活，但是这样的分布式系统始终需要中心化的节点来监控或者调度系统的资源，即使该中心节点也可能是多节点组成。区块链则是真正的区中心化分布式系统，系统里面只有 P2P 网络协议各自通信，没有真正意义的中心节点，彼此按照区块链节点的算力、权益等机制来协调新区块的产生。

• 比特币
• 以太坊

设计模式

上节我们列举了不同场景下不同分布式系统架构扮演的角色和实现的功能，本节我们更进一步归纳分布式系统设计的时候是如何考虑架构设计的、不同设计方案直接的区别和侧重点、不同场景需要选择合作设计模式，来减少试错的成本，设计分布式系统需要考虑以下的问题。

可用性

可用性是系统运行和工作的时间比例，通常以正常运行时间的百分比来衡量。它可能受系统错误、基础架构问题、恶意攻击和系统负载的影响。分布式系统通常为用户提供服务级别协议（SLA），因此应用程序必须设计为最大化可用性。

• 健康检查：系统实现全链路功能检查，外部工具定期通过公开端点访问系统
• 负载均衡：使用队列起到削峰作用，作为请求和服务之间的缓冲区，以平滑间歇性的重负载
• 节流：限制应用级别、租户或整个服务所消耗资源的范围

数据管理

数据管理是分布式系统的关键要素，并影响大多数质量的属性。由于性能，可扩展性或可用性等原因，数据通常托管在不同位置和多个服务器上，这可能带来一系列挑战。例如，必须维护数据一致性，并且通常需要跨不同位置同步数据。

• 缓存：根据需要将数据从数据存储层加载到缓存
• CQRS(Command Query Responsibility Segregation)： 命令查询职责分离
• 事件溯源：仅使用追加方式记录域中完整的系列事件
• 索引表：在经常查询引用的字段上创建索引
• 物化视图：生成一个或多个数据预填充视图
• 拆分：将数据拆分为水平的分区或分片

设计与实现

良好的设计包括诸如组件设计和部署的一致性、简化管理和开发的可维护性、以及允许组件和子系统用于其他应用程序和其他方案的可重用性等因素。在设计和实施阶段做出的决策对分布式系统和服务质量和总体拥有成本产生巨大影响。

• 代理：反向代理
• 适配器： 在现代应用程序和遗留系统之间实现适配器层
• 前后端分离： 后端服务提供接口供前端应用程序调用
• 计算资源整合：将多个相关任务或操作合并到一个计算单元中
• 配置分离：将配置信息从应用程序部署包中移出到配置中心
• 网关聚合：使用网关将多个单独的请求聚合到一个请求中
• 网关卸载：将共享或专用服务功能卸载到网关代理
• 网关路由：使用单个端点将请求路由到多个服务
• 领导人选举：通过选择一个实例作为负责管理其他实例管理员，协调分布式系统的云
• 管道和过滤器：将复杂的任务分解为一系列可以重复使用的单独组件
• 边车：将应用的监控组件部署到单独的进程或容器中，以提供隔离和封装
• 静态内容托管：将静态内容部署到 CDN，加速访问效率

消息

分布式系统需要一个连接组件和服务的消息传递中间件，理想情况是以松散耦合的方式，以便最大限度地提高可伸缩性。异步消息传递被广泛使用，并提供许多好处，但也带来了诸如消息排序，幂等性等挑战

• 竞争消费者：多线程并发消费
• 优先级队列： 消息队列分优先级，优先级高的先被消费

管理与监控

分布式系统在远程数据中心运行，无法完全控制基础结构，这使管理和监视比单机部署更困难。应用必须公开运行时信息，管理员可以使用这些信息来管理和监视系统，以及支持不断变化的业务需求和自定义，而无需停止或重新部署应用。

性能与扩展

性能表示系统在给定时间间隔内执行任何操作的响应性，而可伸缩性是系统处理负载增加而不影响性能或容易增加可用资源的能力。分布式系统通常会遇到变化的负载和活动高峰，特别是在多租户场景中，几乎是不可能预测的。相反，应用应该能够在限制范围内扩展以满足需求高峰，并在需求减少时进行扩展。可伸缩性不仅涉及计算实例，还涉及其他元素，如数据存储、消息队列等。

弹性

弹性是指系统能够优雅地处理故障并从故障中恢复。分布式系统通常是多租户，使用共享平台服务、竞争资源和带宽，通过 Internet 进行通信，以及在商用硬件上运行，意味着出现瞬态和更永久性故障的可能性增加。为了保持弹性，必须快速有效地检测故障并进行恢复。

• 隔离：将应用程序的元素隔离到池中，以便在其中一个失败时，其他元素将继续运行
• 断路器：处理连接到远程服务或资源时可能需要不同时间修复的故障
• 补偿交易：撤消一系列步骤执行的工作，这些步骤共同定义最终一致的操作
• 健康检查：系统实现全链路功能检查，外部工具定期通过公开端点访问系统
• 重试：通过透明地重试先前失败的操作，使应用程序在尝试连接到服务或网络资源时处理预期的临时故障

安全

安全性是系统能够防止在设计使用之外的恶意或意外行为，并防止泄露或丢失信息。分布式系统在受信任的本地边界之外的 Internet 上运行，通常向公众开放，并且可以为不受信任的用户提供服务。必须以保护应用程序免受恶意攻击，限制仅允许对已批准用户的访问，并保护敏感数据。

• 联合身份：将身份验证委派给外部身份提供商
• 看门人： 通过使用专用主机实例来保护应用程序和服务，该实例充当客户端与应用程序或服务之间的代理，验证和清理请求，并在它们之间传递请求和数据
• 代客钥匙：使用为客户端提供对特定资源或服务的受限直接访问的令牌或密钥

工程应用

前文我们介绍了分布式系统的核心理论，面临的一些难题和解决问题的折中思路，罗列了现有主流分布式系统的分类，而且归纳了建设分布式系统的一些方法论，那么接下来我们将从工程角度来介绍真刀真枪搭建分布式系统包含的内容和步骤。

资源调度

巧妇难为无米之炊，我们一切的软件系统都是构建在硬件服务器的基础上。从最开始的物理机直接部署软件系统，到虚拟机的应用，最后到了资源上云容器化，硬件资源的使用也开始了集约化的管理。本节对比的是传统运维角色对应的职责范围，在 devops 环境下，开发运维一体化，我们要实现的也是资源的灵活高效使用。

弹性伸缩

过去软件系统随着用户量增加需要增加机器资源的话，传统的方式就是找运维申请机器，然后部署好软件服务接入集群，整个过程依赖的是运维人员的人肉经验，效率低下而且容易出错。微服务分布式则无需人肉增加物理机器，在容器化技术的支撑下，我们只需要申请云资源，然后执行容器脚本即可。

• 应用扩容：用户激增需要对服务进行扩展，包括自动化扩容，峰值过后的自动缩容
• 机器下线：对于过时应用，进行应用下线，云平台收回容器宿主资源
• 机器置换：对于故障机器，可供置换容器宿主资源，服务自动启动，无缝切换

网络管理

有了计算资源后，另外最重要的就是网络资源了。在现有的云化背景下，我们几乎不会直接接触到物理的带宽资源，而是直接由云平台统一管理带宽资源。我们需要的是对网络资源的最大化应用和有效的管理。

• 域名申请：应用申请配套域名资源的申请，多套域名映射规则的规范
• 域名变更：域名变更统一平台管理
• 负载管理：多机应用的访问策略设定
• 安全外联：基础访问鉴权，拦截非法请求
• 统一接入：提供统一接入的权限申请平台，提供统一的登录管理

故障快照

在系统故障的时候我们第一要务是系统恢复，同时保留案发现场也是非常重要的，资源调度平台则需要有统一的机制保存好故障现场。

• 现场保留：内存分布，线程数等资源现象的保存，如 JavaDump 钩子接入
• 调试接入：采用字节码技术无需入侵业务代码，可以供生产环境现场日志打点调试

流量调度

在我们建设好分布式系统后，最先受到考验的关口就是网关了，进而我们需要关注系统流量的情况，也就是如何对流量的管理，我们追求的是在系统可容纳的流量上限内，把资源留给最优质的流量使用、把非法恶意的流量挡在门外，这样节省成本的同时确保系统不会被冲击崩溃。

负载均衡

负载均衡是我们对服务如何消化流量的通用设计，通常分为物理层的底层协议分流的硬负载均衡和软件层的软负载。负载均衡解决方案已经是业界成熟的方案，我们通常会针对特定业务在不同环境进行优化，常用有如下的负载均衡解决方案

• 交换机
• F5
• LVS/ALI-LVS
• Nginx/Tengine
• VIPServer/ConfigServer

网关设计

负载均衡首当其冲的就是网关，因为中心化集群流量最先打到的地方就是网关了，如果网关扛不住压力的话，那么整个系统将不可用。

• 高性能：网关设计第一需要考虑的是高性能的流量转发，网关单节点通常能达到上百万的并发流量
• 分布式：出于流量压力分担和灾备考虑，网关设计同样需要分布式
• 业务筛选：网关同设计简单的规则，排除掉大部分的恶意流量

流量管理

• 请求校验：请求鉴权可以把多少非法请求拦截，清洗
• 数据缓存：多数无状态的请求存在数据热点，所以采用 CDN 可以把相当大一部分的流量消费掉

流控控制

剩下的真实流量我们采用不同的算法来分流请求。

• 流量分配

+ 计数器
+ 队列
+ 漏斗
+ 令牌桶
+ 动态流控

• 流量限制在流量激增的时候，通常我们需要有限流措施来防止系统出现雪崩，那么就需要预估系统的流量上限，然后设定好上限数，但流量增加到一定阈值后，多出来的流量则不会进入系统，通过牺牲部分流量来保全系统的可用性。

+ 限流策略
+ QPS 粒度
+ 线程数粒度
+ RT 阈值
+ 限流工具 - Sentinel

服务调度

所谓打铁还需自身硬，流量做好了调度管理后，剩下的就是服务自身的健壮性了。分布式系统服务出现故障是常有的事情，甚至我们需要把故障本身当做是分布式服务的一部分。

注册中心

我们网络管理一节中介绍了网关，网关是流量的集散地，而注册中心则是服务的根据地。

• 状态类型：第一好应用服务的状态，通过注册中心就可以检测服务是否可用
• 生命周期：应用服务不同的状态组成了应用的生命周期

版本管理

• 集群版本：集群不用应用有自身对应的版本号，由不同服务组成的集群也需要定义大的版本号
• 版本回滚：在部署异常的时候可以根据大的集群版本进行回滚管理

服务编排

服务编排的定义是：通过消息的交互序列来控制各个部分资源的交互。参与交互的资源都是对等的，没有集中的控制。微服务环境下服务众多我们需要有一个总的协调器来协议服务之间的依赖，调用关系，K8s 则是我们的不二选择。

• K8s
• Spring Cloud
  • HSF
  • ZK+Dubbo

服务控制

前面我们解决了网络的健壮性和效率问题，这节介绍的是如何使我们的服务更加健壮。

• 发现资源管理那节我们介绍了从云平台申请了容器宿主资源后，通过自动化脚本就可以启动应用服务，启动后服务则需要发现注册中心，并且把自身的服务信息注册到服务网关，即是网关接入。注册中心则会监控服务的不同状态，做健康检查，把不可用的服务归类标记。

+ 网关接入
+ 健康检查

• 降级：当用户激增的时候，我们首先是在流量端做手脚，也就是限流。当我们发现限流后系统响应变慢了，有可能导致更多的问题时，我们也需要对服务本身做一些操作。服务降级就是把当前不是很核心的功能关闭掉，或者不是很要紧的准确性放宽范围，事后再做一些人工补救。

+ 降低一致性约束
+ 关闭非核心服务
+ 简化功能

• 熔断：当我们都做了以上的操作后，还是觉得不放心，那么就需要再进一步操心。熔断是对过载的一种自身保护，犹如我们开关跳闸一样。比如当我们服务不断对数据库进行查询的时候，如果业务问题造成查询问题，这是数据库本身需要熔断来保证不会被应用拖垮，并且访问友好的信息，告诉服务不要再盲目调用了。

+ 闭合状态
+ 半开状态
+ 断开状态
+ 熔断工具- Hystrix

• 幂等：我们知道，一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。那么就需要对单次操作赋予一个全局的 id 来做标识，这样多次请求后我们可以判断来源于同个客户端，避免出现脏数据。

+ 全局一致性 ID
+ Snowflake

数据调度

数据存储最大的挑战就是数据冗余的管理，冗余多了效率变低而且占用资源，副本少了起不到灾备的作用，我们通常的做法是把有转态的请求，通过转态分离，转化为无状态请求。

状态转移

分离状态至全局存储，请求转换为无状态流量，比如我们通常会将登陆信息缓存至全局 redis 中间件，而不需要在多个应用中去冗余用户的登陆数据。

分库分表

数据横向扩展。

分片分区

多副本冗余。

自动化运维

我们从资源申请管理的时候就介绍到 devops 的趋势，真正做到开发运维一体化则需要不同的中间件来配合完成。

配置中心

全局配置中心按环境来区分，统一管理，减少了多处配置的混乱局面。

• switch
• diamend

部署策略

微服务分布式部署是家常便饭，如何让我们的服务更好地支撑业务发展，稳健的部署策略是我们首先需要考虑的，如下的部署策略适合不同业务和不同的阶段。

• 停机部署
• 滚动部署
• 蓝绿部署
• 灰度部署
• A/B 测试

作业调度

任务调度是系统必不可少的一个环节，传统的方式是在 Linux 机器上配置 crond 定时任务或者直接在业务代码里面完成调度业务，现在则是成熟的中间件来代替。

• SchedulerX
• Spring 定时任务

应用管理

运维工作中很大一部分时间需要对应用进行重启，上下线操作，还有日志清理。

• 应用重启
• 应用下线
• 日志清理

容错处理

既然我们知道分布式系统故障是家常便饭，那么应对故障的方案也是不可或缺的环节。通常我们有主动和被动的方式来处理：

• 主动是在错误出现的时候，我们试图再试试几次，说不定就成功了，成功的话就可以避免了该次错误
• 被动方式是错误的事情已经发生了，为了挽回，我们只是做时候处理，把负面影响降到最小

重试设计

重试设计的关键在于设计好重试的时间和次数，如果超过重试次数，或是一段时间，那么重试就没有意义了。开源的项目 spring-retry 可以很好地实现我们重试的计划。

事务补偿

事务补偿符合我们最终一致性的理念。补偿事务不一定会将系统中的数据返回到原始操作开始时其所处的状态。 相反，它补偿操作失败前由已成功完成的步骤所执行的工作。补偿事务中步骤的顺序不一定与原始操作中步骤的顺序完全相反。 例如，一个数据存储可能比另一个数据存储对不一致性更加敏感，因而补偿事务中撤销对此存储的更改的步骤应该会首先发生。对完成操作所需的每个资源采用短期的基于超时的锁并预先获取这些资源，这样有助于增加总体活动成功的可能性。 仅在获取所有资源后才应执行工作。 锁过期之前必须完成所有操作。

全栈监控

由于分布式系统是由众多机器共同协作的系统，而且网络也无法保证完全可用，所以我们需要建设一套对各个环节都能监控的系统，这样我们才能从底层到业务各个层面进行监控，出现意外的时候可以及时修复故障，避免更多的问题出现。

基础层

基础层面是对容器资源的监测，包含各个硬件指标的负载情况

• CPU、IO、内存、线程、吞吐

中间件

分布式系统接入了大量的中间件平台，中间件本身的健康情况也需要监控。

应用层

• 性能监控：应用层面的需要对每个应用服务的实时指标（qps，rt），上下游依赖等进行监控
• 业务监控：除了应用本身的监控程度，业务监控也是保证系统正常的一个环节，通过设计合理的业务规则，对异常的情况做报警设置

监控链路

• zipkin/eagleeye
• sls
• goc
• Alimonitor

故障恢复

当故障已经发生后，我们第一个要做的就是马上消除故障，确保系统服务正常可用，这个时候通常做回滚操作。

应用回滚

应用回滚之前需要保存好故障现场，以便排查原因。

基线回退

应用服务回滚后，代码基线也需要 revert 到前一版本。

版本回滚

整体回滚需要服务编排，通过大版本号对集群进行回滚。

性能调优

性能优化是分布式系统的大专题，涉及的面非常广，这块简直可以单独拿出来做一个系列来讲，本节就先不展开。本身我们做服务治理的过程也是在性能的优化过程。
参考《高并发编程知识体系》

分布式锁

缓存是解决性能问题的一大利器，理想情况下，每个请求不需要额外计算就立刻能获取到结果时最快。小到 CPU 的三级缓存，大到分布式缓存，缓存无处不在，分布式缓存需要解决的就是数据的一致性，这个时候我们引入了分布式锁的概念，如何处理分布式锁的问题将决定我们获取缓存数据的效率。

高并发

多线程编程模式提升了系统的吞吐量，但也同时带来了业务的复杂度。

异步

事件驱动的异步编程是一种新的编程模式，摒弃了多线程的复杂业务处理问题，同时能够提升系统的响应效率。

总结

最后总结一下，如果有可能的话，请尝试使用单节点方式而不是分布式系统。分布式系统伴随着一些失败的操作，为了处理灾难性故障，我们使用备份；为了提高可靠性，我们引入了冗余。

分布式系统本质就是一堆机器的协同，而我们要做的就是搞出各种手段来然机器的运行达到预期。这么复杂的系统，需要了解各个环节、各个中间件的接入，是一个非常大的工程。庆幸的是，在微服务背景下，多数基础性的工作已经有人帮我们实现了。前文所描述的分布式架构，在工程实现了是需要用到分布式三件套 (Docker+K8S+Srping Cloud) 基本就可以构建出来了。

分布式架构核心技术分布图如下：

原图来源：dzone.com/articles/de…

分布式技术栈使用中间件：

原图来源：dzone.com/articles/de…

“ 阿里巴巴云原生微信公众号（ID：Alicloudnative）关注微服务、Serverless、容器、Service Mesh等技术领域、聚焦云原生流行技术趋势、云原生大规模的落地实践，做最懂云原生开发者的技术公众号。”

本文转载自: 掘金

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

前言

最近有个需求解析一个订单文件，并且说明文件可达到千万条数据，每条数据大概在20个字段左右，每个字段使用逗号分隔，需要尽量在半小时内入库。

思路

1.估算文件大小

因为告诉文件有千万条，同时每条记录大概在20个字段左右，所以可以大致估算一下整个订单文件的大小，方法也很简单使用FileWriter往文件中插入一千万条数据，查看文件大小，经测试大概在1.5G左右；

2.如何批量插入

由上可知文件比较大，一次性读取内存肯定不行，方法是每次从当前订单文件中截取一部分数据，然后进行批量插入，如何批次插入可以使用**insert(…)values(…),(…)**的方式，经测试这种方式效率还是挺高的；

3.数据的完整性

截取数据的时候需要注意，需要保证数据的完整性，每条记录最后都是一个换行符，需要根据这个标识保证每次截取都是整条数，不要出现半条数据这种情况；

4.数据库是否支持批次数据

因为需要进行批次数据的插入，数据库是否支持大量数据写入，比如这边使用的mysql，可以通过设置max_allowed_packet来保证批次提交的数据量；

5.中途出错的情况

因为是大文件解析，如果中途出现错误，比如数据刚好插入到900w的时候，数据库连接失败，这种情况不可能重新来插一遍，所有需要记录每次插入数据的位置，并且需要保证和批次插入的数据在同一个事务中，这样恢复之后可以从记录的位置开始继续插入。

实现

1.准备数据表

这里需要准备两张表分别是：订单状态位置信息表，订单表；

1
2
3
4
5
6
7
8
9
10
11

复制代码CREATE TABLE `file_analysis` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `file_type` varchar(255) NOT NULL COMMENT '文件类型 01:类型1，02:类型2',
  `file_name` varchar(255) NOT NULL COMMENT '文件名称',
  `file_path` varchar(255) NOT NULL COMMENT '文件路径',
  `status` varchar(255) NOT NULL COMMENT '文件状态  0初始化；1成功；2失败：3处理中',
  `position` bigint(20) NOT NULL COMMENT '上一次处理完成的位置',
  `crt_time` datetime NOT NULL COMMENT '创建时间',
  `upd_time` datetime NOT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8

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

复制代码CREATE TABLE `file_order` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `file_id` bigint(20) DEFAULT NULL,
  `field1` varchar(255) DEFAULT NULL,
  `field2` varchar(255) DEFAULT NULL,
  `field3` varchar(255) DEFAULT NULL,
  `field4` varchar(255) DEFAULT NULL,
  `field5` varchar(255) DEFAULT NULL,
  `field6` varchar(255) DEFAULT NULL,
  `field7` varchar(255) DEFAULT NULL,
  `field8` varchar(255) DEFAULT NULL,
  `field9` varchar(255) DEFAULT NULL,
  `field10` varchar(255) DEFAULT NULL,
  `field11` varchar(255) DEFAULT NULL,
  `field12` varchar(255) DEFAULT NULL,
  `field13` varchar(255) DEFAULT NULL,
  `field14` varchar(255) DEFAULT NULL,
  `field15` varchar(255) DEFAULT NULL,
  `field16` varchar(255) DEFAULT NULL,
  `field17` varchar(255) DEFAULT NULL,
  `field18` varchar(255) DEFAULT NULL,
  `crt_time` datetime NOT NULL COMMENT '创建时间',
  `upd_time` datetime NOT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10000024 DEFAULT CHARSET=utf8

2.配置数据库包大小

1
2
3
4
5
6
7
8
9
10
11

复制代码mysql> show VARIABLES like '%max_allowed_packet%';
+--------------------------+------------+
| Variable_name            | Value      |
+--------------------------+------------+
| max_allowed_packet       | 1048576    |
| slave_max_allowed_packet | 1073741824 |
+--------------------------+------------+
2 rows in set

mysql> set global max_allowed_packet = 1024*1024*10;
Query OK, 0 rows affected

通过设置max_allowed_packet，保证数据库能够接收批次插入的数据包大小；不然会出现如下错误：

1
2
3
4
5

复制代码Caused by: com.mysql.jdbc.PacketTooBigException: Packet for query is too large (4980577 > 1048576). You can change this value on the server by setting the max_allowed_packet' variable.
	at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3915)
	at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2598)
	at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2778)
	at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2834)

3.准备测试数据

1
2
3
4
5
6
7
8
9

复制代码	public static void main(String[] args) throws IOException {
		FileWriter out = new FileWriter(new File("D://xxxxxxx//orders.txt"));
		for (int i = 0; i < 10000000; i++) {
			out.write(
					"vaule1,vaule2,vaule3,vaule4,vaule5,vaule6,vaule7,vaule8,vaule9,vaule10,vaule11,vaule12,vaule13,vaule14,vaule15,vaule16,vaule17,vaule18");
			out.write(System.getProperty("line.separator"));
		}
		out.close();
	}

使用FileWriter遍历往一个文件里插入1000w条数据即可，这个速度还是很快的，不要忘了在每条数据的后面添加换行符(\n\r)；

4.截取数据的完整性

除了需要设置每次读取文件的大小，同时还需要设置一个参数，用来每次获取一小部分数据，从这小部分数据中获取换行符(\n\r)，如果获取不到一直累加直接获取为止，这个值设置大小大致同每条数据的大小差不多合适，部分实现如下：

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

复制代码ByteBuffer byteBuffer = ByteBuffer.allocate(buffSize); // 申请一个缓存区
long endPosition = batchFileSize + startPosition - buffSize;// 子文件结束位置

long startTime, endTime;
for (int i = 0; i < count; i++) {
	startTime = System.currentTimeMillis();
	if (i + 1 != count) {
		int read = inputChannel.read(byteBuffer, endPosition);// 读取数据
		readW: while (read != -1) {
			byteBuffer.flip();// 切换读模式
			byte[] array = byteBuffer.array();
			for (int j = 0; j < array.length; j++) {
				byte b = array[j];
				if (b == 10 || b == 13) { // 判断\n\r
					endPosition += j;
					break readW;
				}
			}
			endPosition += buffSize;
			byteBuffer.clear(); // 重置缓存块指针
			read = inputChannel.read(byteBuffer, endPosition);
		}
	} else {
		endPosition = fileSize; // 最后一个文件直接指向文件末尾
	}
    ...省略，更多可以查看Github完整代码...
}

如上代码所示开辟了一个缓冲区，根据每行数据大小来定大概在200字节左右，然后通过遍历查找换行符(\n\r)，找到以后将当前的位置加到之前的结束位置上，保证了数据的完整性；

5.批次插入数据

通过**insert(…)values(…),(…)**的方式批次插入数据，部分代码如下：

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

复制代码// 保存订单和解析位置保证在一个事务中
		SqlSession session = sqlSessionFactory.openSession();
		try {
			long startTime = System.currentTimeMillis();
			FielAnalysisMapper fielAnalysisMapper = session.getMapper(FielAnalysisMapper.class);
			FileOrderMapper fileOrderMapper = session.getMapper(FileOrderMapper.class);
			fileOrderMapper.batchInsert(orderList);

			// 更新上次解析到的位置，同时指定更新时间
			fileAnalysis.setPosition(endPosition + 1);
			fileAnalysis.setStatus("3");
			fileAnalysis.setUpdTime(new Date());
			fielAnalysisMapper.updateFileAnalysis(fileAnalysis);
			session.commit();
			long endTime = System.currentTimeMillis();
			System.out.println("===插入数据花费:" + (endTime - startTime) + "ms===");
		} catch (Exception e) {
			session.rollback();
		} finally {
			session.close();
		}
        ...省略，更多可以查看Github完整代码...

如上代码在一个事务中同时保存批次订单数据和文件解析位置信息，batchInsert通过使用mybatis的****标签来遍历订单列表，生成values数据；

总结

以上展示了部分代码，完整的代码可以查看Github地址中的batchInsert模块，本地设置每次截取的文件大小为2M，经测试1000w条数据(大小1.5G左右)插入mysql数据库中，大概花费时间在20分钟左右，当然可以通过设置截取的文件大小，花费的时间也会相应的改变。

完整代码

Github

本文转载自: 掘金

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

作者 | 莫源 阿里巴巴技术专家

一、背景

监控和日志是大型分布式系统的重要基础设施，监控可以帮助开发者查看系统的运行状态，而日志可以协助问题的排查和诊断。

在 Kubernetes 中，监控和日志属于生态的一部分，它并不是核心组件，因此大部分的能力依赖上层的云厂商的适配。Kubernetes 定义了介入的接口标准和规范，任何符合接口标准的组件都可以快速集成。

二、监控

监控类型

先看一下监控，从监控类型上划分，在 K8s 中可以分成四个不同的类型：

1.资源监控

比较常见的像 CPU、内存、网络这种资源类的一个指标，通常这些指标会以数值、百分比的单位进行统计，是最常见的一个监控方式。这种监控方式在常规的监控里面，类似项目 zabbix telegraph，这些系统都是可以做到的。

2.性能监控

性能监控指的就是 APM 监控，也就是说常见的一些应用性能类的监控指标的检查。通常是通过一些 Hook 的机制在虚拟机层、字节码执行层通过隐式调用，或者是在应用层显示注入，获取更深层次的一个监控指标，一般是用来应用的调优和诊断的。比较常见的类似像 jvm 或者 php 的 Zend Engine，通过一些常见的 Hook 机制，拿到类似像 jvm 里面的 GC 的次数，各种内存代的一个分布以及网络连接数的一些指标，通过这种方式来进行应用的性能诊断和调优。

3.安全监控

安全监控主要是对安全进行的一系列的监控策略，类似像越权管理、安全漏洞扫描等等。

4.事件监控

事件监控是 K8s 中比较另类的一种监控方式。之前的文章为大家介绍了在 K8s 中的一个设计理念，就是基于状态机的一个状态转换。从正常的状态转换成另一个正常的状态的时候，会发生一个 normal 的事件，而从一个正常状态转换成一个异常状态的时候，会发生一个 warning 的事件。通常情况下，warning 的事件是我们比较关心的，而事件监控就是可以把 normal 的事件或者是 warning 事件离线到一个数据中心，然后通过数据中心的分析以及报警，把相应的一些异常通过像钉钉或者是短信、邮件的方式进行暴露，弥补常规监控的一些缺陷和弊端。

Kubernetes 的监控演进

在早期，也就是 1.10 以前的 K8s 版本。大家都会使用类似像 Heapster 这样的组件来去进行监控的采集，Heapster 的设计原理其实也比较简单。

首先，我们在每一个 Kubernetes 上面有一个包裹好的 cadvisor，这个 cadvisor 是负责数据采集的组件。当 cadvisor 把数据采集完成，Kubernetes 会把 cadvisor 采集到的数据进行包裹，暴露成相应的 API。在早期的时候，实际上是有三种不同的 API：

• 第一种是 summary 接口；
• 第二种是 kubelet 接口；
• 第三种是 Prometheus 接口。

这三种接口，其实对应的数据源都是 cadvisor，只是数据格式有所不同。而在 Heapster 里面，其实支持了 summary 接口和 kubelet 两种数据采集接口，Heapster 会定期去每一个节点拉取数据，在自己的内存里面进行聚合，然后再暴露相应的 service，供上层的消费者进行使用。在 K8s 中比较常见的消费者，类似像 dashboard，或者是 HPA-Controller，它通过调用 service 获取相应的监控数据，来实现相应的弹性伸缩，以及监控数据的一个展示。

这个是以前的一个数据消费链路，这条消费链路看上去很清晰，也没有太多的一个问题，那为什么 Kubernetes 会将 Heapster 放弃掉而转换到 metrics-service 呢？其实这个主要的一个动力来源是由于 Heapster 在做监控数据接口的标准化。为什么要做监控数据接口标准化呢？

• 第一点在于客户的需求是千变万化的，比如说今天用 Heapster 进行了基础数据的一个资源采集，那明天的时候，我想在应用里面暴露在线人数的一个数据接口，放到自己的接口系统里进行数据的一个展现，以及类似像 HPA 的一个数据消费。那这个场景在 Heapster 下能不能做呢？答案是不可以的，所以这就是 Heapster 自身扩展性的弊端；

• 第二点是 Heapster 里面为了保证数据的离线能力，提供了很多的 sink，而这个 sink 包含了类似像 influxdb、sls、钉钉等等一系列 sink。这个 sink 主要做的是把数据采集下来，并且把这个数据离线走，然后很多客户会用 influxdb 做这个数据离线，在 influxdb 上去接入类似像 grafana 监控数据的一个可视化的软件，来实践监控数据的可视化。

但是后来社区发现，这些 sink 很多时候都是没有人来维护的。这也导致整个 Heapster 的项目有很多的 bug，这个 bug 一直存留在社区里面，是没有人修复的，这个也是会给社区的项目的活跃度包括项目的稳定性带来了很多的挑战。

基于这两点原因，K8s 把 Heapster 进行了 break 掉，然后做了一个精简版的监控采集组件，叫做 metrics-server。

上图是 Heapster 内部的一个架构。大家可以发现它分为几个部分，第一个部分是 core 部分，然后上层是有一个通过标准的 http 或者 https 暴露的这个 API。然后中间是 source 的部分，source 部分相当于是采集数据暴露的不同的接口，然后 processor 的部分是进行数据转换以及数据聚合的部分。最后是 sink 部分，sink 部分是负责数据离线的，这个是早期的 Heapster 的一个应用的架构。那到后期的时候呢，K8s 做了这个监控接口的一个标准化，逐渐就把 Heapster 进行了裁剪，转化成了 metrics-server。

目前 0.3.1 版本的 metrics-server 大致的一个结构就变成了上图这样，是非常简单的：有一个 core 层、中间的 source 层，以及简单的 API 层，额外增加了 API Registration 这层。这层的作用就是它可以把相应的数据接口注册到 K8s 的 API server 之上，以后客户不再需要通过这个 API 层去访问 metrics-server，而是可以通过这个 API 注册层，通过 API server 访问 API 注册层，再到 metrics-server。这样的话，真正的数据消费方可能感知到的并不是一个 metrics-server，而是说感知到的是实现了这样一个 API 的具体的实现，而这个实现是 metrics-server。这个就是 metrics-server 改动最大的一个地方。

Kubernetes 的监控接口标准

在 K8s 里面针对于监控，有三种不同的接口标准。它将监控的数据消费能力进行了标准化和解耦，实现了一个与社区的融合，社区里面主要分为三类。

第一类 Resource Metrice

对应的接口是 metrics.k8s.io，主要的实现就是 metrics-server，它提供的是资源的监控，比较常见的是节点级别、pod 级别、namespace 级别、class 级别。这类的监控指标都可以通过 metrics.k8s.io 这个接口获取到。

第二类 Custom Metrics

对应的 API 是 custom.metrics.k8s.io，主要的实现是 Prometheus。它提供的是资源监控和自定义监控，资源监控和上面的资源监控其实是有覆盖关系的，而这个自定义监控指的是：比如应用上面想暴露一个类似像在线人数，或者说调用后面的这个数据库的 MySQL 的慢查询。这些其实都是可以在应用层做自己的定义的，然后并通过标准的 Prometheus 的 client，暴露出相应的 metrics，然后再被 Prometheus 进行采集。

而这类的接口一旦采集上来也是可以通过类似像 custom.metrics.k8s.io 这样一个接口的标准来进行数据消费的，也就是说现在如果以这种方式接入的 Prometheus，那你就可以通过 custom.metrics.k8s.io 这个接口来进行 HPA，进行数据消费。

第三类 External Metrics

External Metrics 其实是比较特殊的一类，因为我们知道 K8s 现在已经成为了云原生接口的一个实现标准。很多时候在云上打交道的是云服务，比如说在一个应用里面用到了前面的是消息队列，后面的是 RBS 数据库。那有时在进行数据消费的时候，同时需要去消费一些云产品的监控指标，类似像消息队列中消息的数目，或者是接入层 SLB 的 connection 数目，SLB 上层的 200 个请求数目等等，这些监控指标。

那怎么去消费呢？也是在 K8s 里面实现了一个标准，就是 external.metrics.k8s.io。主要的实现厂商就是各个云厂商的 provider，通过这个 provider 可以通过云资源的监控指标。在阿里云上面也实现了阿里巴巴 cloud metrics adapter 用来提供这个标准的 external.metrics.k8s.io 的一个实现。

Promethues - 开源社区的监控“标准”

接下来我们来看一个比较常见的开源社区里面的监控方案，就是 Prometheus。Prometheus 为什么说是开源社区的监控标准呢？

• 一是因为首先 Prometheus 是 CNCF 云原生社区的一个毕业项目。然后第二个是现在有越来越多的开源项目都以 Prometheus 作为监控标准，类似说我们比较常见的 Spark、Tensorflow、Flink 这些项目，其实它都有标准的 Prometheus 的采集接口。

• 第二个是对于类似像比较常见的一些数据库、中间件这类的项目，它都有相应的 Prometheus 采集客户端。类似像 ETCD、zookeeper、MySQL 或者说 PostgreSQL，这些其实都有相应的这个 Prometheus 的接口，如果没有的，社区里面也会有相应的 exporter 进行接口的一个实现。

那我们先来看一下 Prometheus 整个的大致一个结构。

上图是 Prometheus 采集的数据链路，它主要可以分为三种不同的数据采集链路。

• 第一种，是这个 push 的方式，就是通过 pushgateway 进行数据采集，然后数据线到 pushgateway，然后 Prometheus 再通过 pull 的方式去 pushgateway 去拉数据。这种采集方式主要应对的场景就是你的这个任务可能是比较短暂的，比如说我们知道 Prometheus，最常见的采集方式是拉模式，那带来一个问题就是，一旦你的数据声明周期短于数据的采集周期，比如我采集周期是 30s，而我这个任务可能运行 15s 就完了。这种场景之下，可能会造成有些数据漏采。对于这种场景最简单的一个做法就是先通过 pushgateway，先把你的 metrics push下来，然后再通过 pull 的方式从 pushgateway 去拉数据，通过这种方式可以做到，短时间的不丢作业任务。

• 第二种是标准的 pull 模式，它是直接通过拉模式去对应的数据的任务上面去拉取数据。

• 第三种是 Prometheus on Prometheus，就是可以通过另一个 Prometheus 来去同步数据到这个 Prometheus。

这是三种 Prometheus 中的采集方式。那从数据源上面，除了标准的静态配置，Prometheus 也支持 service discovery。也就是说可以通过一些服务发现的机制，动态地去发现一些采集对象。在 K8s 里面比较常见的是可以有 Kubernetes 的这种动态发现机制，只需要配置一些 annotation，它就可以自动地来配置采集任务来进行数据采集，是非常方便的。

etheus 提供了一个外置组件叫 Alentmanager，它可以将相应的报警信息通过邮件或者短信的方式进行数据的一个告警。在数据消费上面，可以通过上层的 API clients，可以通过 web UI，可以通过 Grafana 进行数据的展现和数据的消费。

总结起来 Prometheus 有如下五个特点：

• 第一个特点就是简介强大的接入标准，开发者只需要实现 Prometheus Client 这样一个接口标准，就可以直接实现数据的一个采集；
• 第二种就是多种的数据采集、离线的方式。可以通过 push 的方式、 pull 的方式、Prometheus on Prometheus的方式来进行数据的采集和离线；
• 第三种就是和 K8s 的兼容；
• 第四种就是丰富的插件机制与生态；
• 第五个是 Prometheus Operator 的一个助力，Prometheus Operator 可能是目前我们见到的所有 Operator 里面做的最复杂的，但是它里面也是把 Prometheus 这种动态能力做到淋漓尽致的一个 Operator，如果在 K8s 里面使用 Prometheus，比较推荐大家使用 Prometheus Operator 的方式来去进行部署和运维。

kube-eventer - Kubernetes 事件离线工具

最后，我们给大家介绍一个 K8s 中的事件离线工具叫做 kube-eventer。kube-eventer 是阿里云容器服务开源出的一个组件，它可以将 K8s 里面，类似像 pod eventer、node eventer、核心组件的 eventer、crd 的 eventer 等等一系列的 eventer，通过 API sever 的这个 watch 机制离线到类似像 SLS、Dingtalk、kafka、InfluxDB，然后通过这种离线的机制进行一个时间的审计、监控和告警，我们现在已经把这个项目开源到 GitHub 上了，大家有兴趣的话可以来看一下这个项目。

那上面这张图其实就是 Dingtalk 的一个报警图。可以看见里面有一个 warning 的事件，这个事件是在 kube-system namespace 之下，具体的这个 pod，大致的一个原因是这个 pod 重启失败了，然后大致 reason 就是 backoff，然后具体发生事件是什么时间。可以通过这个信息来做到一个 Checkups。

三、日志

日志的场景

接下来给大家来介绍一下在 K8s 里面日志的一个部分。首先我们来看一下日志的场景，日志在 K8s 里面主要分为四个大的场景：

1. 主机内核的日志

• 第一个是主机内核的日志，主机内核日志可以协助开发者进行一些常见的问题与诊断，比如说网栈的异常，类似像我们的 iptables mark，它可以看到有 controller table 这样的一些 message；

• 第二个是驱动异常，比较常见的是一些网络方案里面有的时候可能会出现驱动异常，或者说是类似 GPU 的一些场景，驱动异常可能是比较常见的一些错误；

• 第三个就是文件系统异常，在早期 docker 还不是很成熟的场景之下，overlayfs 或者是 AUFS，实际上是会经常出现问题的。在这些出现问题后，开发者是没有太好的办法来去进行监控和诊断的。这一部分，其实是可以主机内核日志里面来查看到一些异常；

• 再往下是影响节点的一些异常，比如说内核里面的一些 kernel panic，或者是一些 OOM，这些也会在主机日志里面有相应的一些反映。

2. Runtime 的日志

第二个是 runtime 的日志，比较常见的是 Docker 的一些日志，我们可以通过 docker 的日志来排查类似像删除一些 Pod Hang 这一系列的问题。

3. 核心组件的日志

第三个是核心组件的日志，在 K8s 里面核心组件包含了类似像一些外置的中间件，类似像 etcd，或者像一些内置的组件，类似像 API server、kube-scheduler、controller-manger、kubelet 等等这一系列的组件。而这些组件的日志可以帮我们来看到整个 K8s 集群里面管控面的一个资源的使用量，然后以及目前运行的一个状态是否有一些异常。

还有的就是类似像一些核心的中间件，如 Ingress 这种网络中间件，它可以帮我们来看到整个的一个接入层的一个流量，通过 Ingress 的日志，可以做到一个很好的接入层的一个应用分析。

4. 部署应用的日志

最后是部署应用的日志，可以通过应用的日志来查看业务层的一个状态。比如说可以看业务层有没有 500 的请求？有没有一些 panic？有没有一些异常的错误的访问？那这些其实都可以通过应用日志来进行查看的。

日志的采集

首先我们来看一下日志采集，从采集位置是哪个划分，需要支持如下三种：

• 首先是宿主机文件，这种场景比较常见的是说我的这个容器里面，通过类似像 volume，把日志文件写到了宿主机之上。通过宿主机的日志轮转的策略进行日志的轮转，然后再通过我的宿主机上的这个 agent 进行采集；
• 第二种是容器内有日志文件，那这种常见方式怎么处理呢，比较常见的一个方式是说我通过一个 Sidecar 的 streaming 的 container，转写到 stdout，通过 stdout 写到相应的 log-file，然后再通过本地的一个日志轮转，然后以及外部的一个 agent 采集；

• 第三种我们直接写到 stdout，这种比较常见的一个策略，第一种就是直接我拿这个 agent 去采集到远端，第二种我直接通过类似像一些 sls 的标准 API 采集到远端。

那社区里面其实比较推荐的是使用 **Fluentd **的一个采集方案，Fluentd 是在每一个节点上面都会起相应的 agent，然后这个 agent 会把数据汇集到一个 Fluentd 的一个 server，这个 server 里面可以将数据离线到相应的类似像 elasticsearch，然后再通过 kibana 做展现；或者是离线到 influxdb，然后通过 Grafana 做展现。这个其实是社区里目前比较推荐的一个做法。

四、总结

最后给大家做一下今天课程的总结，以及给大家介绍一下在阿里云上面监控和日志的最佳实践。在课程开始的时候，给大家介绍了监控和日志并不属于 K8s 里面的核心组件，而大部分是定义了一个标准的一个接口方式，然后通过上层的这个云厂商进行各自的一个适配。

阿里云容器服务监控体系

监控体系组件介绍

首先，我先给大家来介绍一下在阿里云容器服务里面的监控体系，这张图实际上是监控的一个大图。

右侧的四个产品是和监控日志相关比较紧密的四个产品：

sls

第一个是 SLS，就是日志服务，那刚才我们已经提到了在 K8s 里面日志分为很多种不同的采集，比如说有核心组件的日志、接入层的日志、还有应用的日志等等。在阿里云容器服务里面，可以通过 API server 采集到审计的日志，然后可以通过类似像 service mesh 或者 ingress controller 采集到接入层的日志，然后以及相应的应用层采集到应用的日志。

有了这条数据链路之后，其实还不够。因为数据链路只是帮我们做到了一个数据的离线，我们还需要做上层的数据的展现和分析。比如说像审计，可以通过审计日志来看到今天有多少操作、有多少变更、有没有攻击、系统有没有异常。这些都可以通过审计的 Dashboard 来查看。

ARMS

第二个就是应用的一个性能监控。性能监控上面，可以通过这个 ARMS 这样的产品来去进行查看。ARMS 目前支持的 JAVA、PHP 两种语言，可以通过 ARMS 来做应用的一个性能诊断和问题的一个调优。

AHAS

第三个是比较特殊的叫 AHAS。AHAS 是一个架构感知的监控，我们知道在 K8s 里面，很多时候都是通过一些微服的架构进行部署的。微服带来的问题就是组件会变的非常多，组件的副本处也会变的很多。这会带来一个在拓扑管理上面的一个复杂性。

如果我们想要看一个应用在 K8s 中流量的一个走向，或者是针对流量异常的一个排查，其实没有一个很好的可视化是很复杂的。AHAS 的一个作用就是通过网络栈的一个监控，可以绘制出整个 K8s 中应用的一个拓扑关系，然后以及相应的资源监控和网络的带宽监控、流量的监控，以及异常事件的一个诊断。任何如果有架构拓扑感知的一个层面，来实现另一种的监控解决方案。

Cloud Monitor

最后是 Cloud Monitor，也就是基础的云监控。它可以采集标准的 Resource Metrics Monitoring，来进行监控数据的一个展现，可以实现 node、pod 等等监控指标的一个展现和告警。

阿里云增强的功能

这一部分是阿里云在开源上做的增强。首先是 metrics-server，文章开始提到了 metrics-server 做了很多的一个精简。但是从客户的角度来讲，这个精简实际上是把一些功能做了一个裁剪，这将会带来很多不便。比如说有很多客户希望将监控数据离线到类似像 SLS 或者是 influxdb，这种能力实际上用社区的版本是没有办法继续来做的，这个地方阿里云继续保留了常见的维护率比较高的 sink，这是第一个增强。

然后是第二个增强，因为在 K8s 里面整合的一个生态的发展并不是以同样的节奏进行演进的。比如说 Dashboard 的发布，并不是和 K8s 的大版本进行匹配的。比如 K8s 发了 1.12，Dashboard 并不会也发 1.12 的版本，而是说它会根据自己的节奏来去发布，这样会造成一个结果就是说以前依赖于 Heapster 的很多的组件在升级到 metrics-server 之后就直接 break 掉，阿里云在 metrics-server 上面做了完整的 Heapster 兼容，也就是说从目前 K8s 1.7 版本一直到 K8s 1.14 版本，都可以使用阿里云的 metrics-server，来做到完整的监控组件的消费的一个兼容。

还有就是 eventer 和 npd，上面提到了 kube-eventer 这个组件。然后在 npd 上面，我们也做了很多额外的增强，类似像增加了很多监控和检测项，类似像 kernel Hang、npd 的一个检测、出入网的监控、snat 的一个检测。然后还有类似像 fd 的 check，这些其实都是在 npd 里面的一些监控项，阿里云做了很多的增强。然后开发者可以直接部署 npd 的一个 check，就可以实现节点诊断的一个告警，然后并通过 eventer 离线上的 kafka 或者是 Dingtalk。

再往上是 Prometheus 生态，Prometheus 生态里面，在存储层可以让开发者对接，阿里云的 HiTSDB 以及 InfluxDB，然后在采集层提供了优化的 node-exporter，以及一些场景化监控的 exporter，类似像 Spark、TensorFlow、Argo 这类场景化的 exporter。还有就是针对于 GPU，阿里云做了很多额外的增强，类似于像支持 GPU 的单卡监控以及 GPU share 的监控，然后在 Prometheus 上面，我们连同 ARMS 团队推出了托管版的 Prometheus，开发者可以使用开箱即用的 helm chats，不需要部署 Prometheus server，就可以直接体验到 Prometheus 的一个监控采集能力。

阿里云容器服务日志体系

在日志上面，阿里云做了哪些增强呢？首先是采集方式上，做到了完整的一个兼容。可以采集 pod log 日志、核心组件日志、docker engine 日志、kernel 日志，以及类似像一些中间件的日志，都收集到 SLS。收集到 SLS 之后，我们可以通过数据离线到 OSS，离线到 Max Compute，做一个数据的离线和归档，以及离线预算。

然后还有是对于一些数据的实时消费，我们可以到 Opensearch、可以到 E-Map、可以到 Flink，来做到一个日志的搜索和上层的一个消费。在日志展现上面，我们可以既对接开源的 Grafana，也可以对接类似像 DataV，去做数据展示，实现一个完整的数据链路的采集和消费。

本文总结

• 首先主要为大家介绍了监控，其中包括：四种容器场景下的常见的监控方式；Kubernetes 的监控演进和接口标准；两种常用的来源的监控方案；
• 在日志上我们主要介绍了四种不同的场景，介绍了 Fluentd 的一个采集方案；
• 最后向大家介绍了一下阿里云日志和监控的一个最佳实践。

“ 阿里巴巴云原生微信公众号（ID：Alicloudnative）关注微服务、Serverless、容器、Service Mesh等技术领域、聚焦云原生流行技术趋势、云原生大规模的落地实践，做最懂云原生开发者的技术公众号。”

本文转载自: 掘金

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

在上篇文章中，我们详细分析了doCreateBean()中的第5步：属性填充，本文接着分析doCreateBean()的第6步——初始化 bean 实例对象

本文转自公众号：芋道源码

首先回顾下CreateBean的主流程：

1. 如果是单例模式，从factoryBeanInstanceCache 缓存中获取BeanWrapper 实例对象并删除缓存
2. 调用 createBeanInstance() 实例化 bean
3. 后置处理
4. 单例模式的循环依赖处理
5. 属性填充
6. 初始化 bean 实例对象
7. 依赖检查
8. 注册bean的销毁方法

一、初始化

Spring在对Bean进行属性填充之后，会对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
25
26
27
28
29
30
31
32
33
34
35
36
37
38

复制代码//AbstractAutowireCapableBeanFactory.java

protected Object initializeBean(final String beanName, final Object bean, @Nullable RootBeanDefinition mbd) {
	//JDK的安全机制验证权限
	if (System.getSecurityManager() != null) {
		// <1> 激活 Aware 方法，对特殊的 bean 处理：Aware、BeanClassLoaderAware、BeanFactoryAware
		AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
			invokeAwareMethods(beanName, bean);
			return null;
		}, getAccessControlContext());
	}
	else {
		// <1> 激活 Aware 方法，对特殊的 bean 处理：Aware、BeanClassLoaderAware、BeanFactoryAware
		invokeAwareMethods(beanName, bean);
	}

	Object wrappedBean = bean;
	// <2> 后置处理器，before
	if (mbd == null || !mbd.isSynthetic()) {
		wrappedBean = applyBeanPostProcessorsBeforeInitialization(wrappedBean, beanName);
	}

	// <3> 激活用户自定义的 init 方法
	try {
		invokeInitMethods(beanName, wrappedBean, mbd);
	}
	catch (Throwable ex) {
		throw new BeanCreationException(
				(mbd != null ? mbd.getResourceDescription() : null),
				beanName, "Invocation of init method failed", ex);
	}
	// <2> 后置处理器，after
	if (mbd == null || !mbd.isSynthetic()) {
		wrappedBean = applyBeanPostProcessorsAfterInitialization(wrappedBean, beanName);
	}

	return wrappedBean;
	}

初始化 bean 的方法其实就是三个步骤的处理，而这三个步骤主要还是根据用户设定的来进行初始化，这三个过程为：

• <1> 激活 Aware 方法。
• <2> 后置处理器。
• <3> 自定义的 init 方法。

1.1、Aware

Aware ，英文翻译是意识到的，感知的。Spring 提供了诸多 Aware 接口，用于辅助 Spring Bean 以编程的方式调用 Spring 容器，通过实现这些接口，可以增强 Spring Bean 的功能。

Spring 提供了如下系列的 Aware 接口：

LoadTimeWeaverAware：加载Spring Bean时织入第三方模块，如AspectJ

BeanClassLoaderAware：加载Spring Bean的类加载器

BootstrapContextAware：资源适配器BootstrapContext，如JCA,CCI

ResourceLoaderAware：底层访问资源的加载器

BeanFactoryAware：声明BeanFactory

PortletConfigAware：PortletConfig

PortletContextAware：PortletContext

ServletConfigAware：ServletConfig

ServletContextAware：ServletContext

MessageSourceAware：国际化

ApplicationEventPublisherAware：应用事件

NotificationPublisherAware：JMX通知

BeanNameAware：声明Spring Bean的名字

Aware比较复杂，后面会专门学习一下这块内容，这里就不多说了。

1.2、后置处理器

BeanPostProcessor 在前面介绍 bean 加载的过程曾多次遇到，

它的作用是：

如果我们想要在 Spring 容器完成 Bean 的实例化，配置和其他的初始化后添加一些自己的逻辑处理，那么请使用该接口，这个接口给与了用户充足的权限去更改或者扩展 Spring，是我们对 Spring 进行扩展和增强处理一个必不可少的接口。

applyBeanPostProcessorsBeforeInitialization() 方法，代码如下：

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

复制代码// AbstractAutowireCapableBeanFactory.java

@Override
public Object applyBeanPostProcessorsBeforeInitialization(Object existingBean, String beanName)
        throws BeansException {
    Object result = existingBean;
    // 遍历 BeanPostProcessor 数组
    for (BeanPostProcessor processor : getBeanPostProcessors()) {
        // 处理
        Object current = processor.postProcessBeforeInitialization(result, beanName);
        // 返回空，则返回 result
        if (current == null) {
            return result;
        }
        // 修改 result
        result = current;
    }
    return result;
}

applyBeanPostProcessorsAfterInitialization() 方法，代码如下：

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

复制代码// AbstractAutowireCapableBeanFactory.java

@Override
public Object applyBeanPostProcessorsAfterInitialization(Object existingBean, String beanName)
        throws BeansException {
    Object result = existingBean;
    // 遍历 BeanPostProcessor
    for (BeanPostProcessor processor : getBeanPostProcessors()) {
        // 处理
        Object current = processor.postProcessAfterInitialization(result, beanName);
        // 返回空，则返回 result
        if (current == null) {
            return result;
        }
        // 修改 result
        result = current;
    }
    return result;
}

其逻辑就是通过 getBeanPostProcessors() 方法，获取定义的 BeanPostProcessor ，然后分别调用其 postProcessBeforeInitialization() 和 postProcessAfterInitialization() 方法，进行自定义的业务处理。

1.3、自定义init方法

在xml中有一个< bean >标签的配置， init-method 方法，是可以让我们在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
25
26
27
28
29
30
31
32
33
34

复制代码protected void invokeInitMethods(String beanName, final Object bean, @Nullable RootBeanDefinition mbd)
        throws Throwable {
    // 首先会检查是否是 InitializingBean ，如果是的话需要调用 afterPropertiesSet()
    boolean isInitializingBean = (bean instanceof InitializingBean);
    if (isInitializingBean && (mbd == null || !mbd.isExternallyManagedInitMethod("afterPropertiesSet"))) {
        if (logger.isTraceEnabled()) {
            logger.trace("Invoking afterPropertiesSet() on bean with name '" + beanName + "'");
        }
        if (System.getSecurityManager() != null) { // 安全模式
            try {
                AccessController.doPrivileged((PrivilegedExceptionAction<Object>) () -> {
                    // <1> 属性初始化的处理
                    ((InitializingBean) bean).afterPropertiesSet();
                    return null;
                }, getAccessControlContext());
            } catch (PrivilegedActionException pae) {
                throw pae.getException();
            }
        } else {
            // <1> 属性初始化的处理
            ((InitializingBean) bean).afterPropertiesSet();
        }
    }

    if (mbd != null && bean.getClass() != NullBean.class) {
        String initMethodName = mbd.getInitMethodName();
        if (StringUtils.hasLength(initMethodName) &&
                !(isInitializingBean && "afterPropertiesSet".equals(initMethodName)) &&
                !mbd.isExternallyManagedInitMethod(initMethodName)) {
            // <2> 激活用户自定义的初始化方法
            invokeCustomInitMethod(beanName, bean, mbd);
        }
    }
}

首先，检查是否为 InitializingBean 。如果是的话，需要执行 afterPropertiesSet() 方法，因为我们除了可以使用 init-method 来自定初始化方法外，还可以实现 InitializingBean 接口。接口仅有一个 afterPropertiesSet() 方法。

两者的执行先后顺序是先 <1> 的 #afterPropertiesSet() 方法，后 <2> 的 init-method 对应的方法。

本文转载自: 掘金

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

项目背景

公司内部封装了NLP通用算法的GRPC服务，比如文本情感识别、文本分类、实体识别等， 提供给大数据等其他部门实时调用。

RPC 服务的调用日志，通过Filebeat、Logstash 实时发送到Elasticsearch，现在需要通过对日志的调用情况实时统计分析，判断调用情况是否出现异常，并对异常情况能够实时告警。

业务场景分析

使用ES 的watcher 插件创建一个threshold alert， 设置预警规则，当一个时间周期内的数据量达到阈值， 就进行告警。

这种方案优点是，设置起来比较简单，只要设置query的条件，以及阈值就可以快速完成。
缺点是，这种方案比较适合粗粒度，并且阈值明确的情况下设置。

设想一下，如果要做到分钟级实时监控，并且对异常点进行告警通知，如果用规则来设置，很难满足这种场景。
针对这种场景，很容易想到，使用Machine Learning （ML）的方法，根据以往的数据变化趋势，来主动发现异常点，那就简单多了。

针对这种需求， 有很多解决方案，比如，把日志实时发送到KAFKA，再用FlinkML 来做机器学习，训练异常检测模型， 不过多讨论这种方案。
如题， 这里介绍一种更简单的方案，使用 ES Stack 提供的 ML 模块, 很容在ES Stack上快速体验AIOps。

实践经验

Requirements

• ELK 6.4.3
• Xpack Machine Learning

日志结构

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

复制代码{
  "@timestamp": "2019-10-09T09:10:22.547Z",
  "@version": "1",
  "call_day": "20191009",
  "call_hour": "2019100909",
  "call_minute": "201910090910",
  "call_time": "10-09 09:10:21.714317",
  "cost": 0.004031,
  "method": "Nlp.ContentCategory",
  "month": "10",
  "service_name": "mg",
  "timestamp": "2019-10-09T09:10:22.547Z"
}

• @timestamp 日志写入时间，后面的时序统计就是依据这个时间字段
• cost 调用方法消耗时间，可以利用这个字段的平均值来预测方法耗时是否异常
• method 调用的方法
• service_name 调用的服务名称

目的

• 使用历史日志数据，训练异常检测模型，并找出异常的点
• 使用训练出来的模型，检测后面的实时日志数据，发现新的异常点，并进行告警

步骤

创建Job

有四种创建ML Job的模式：

1. 单指标模式：使用单一指标作为模型特征参数
2. 多指标模式：使用多指标作为模型特征参数
3. 种群分析模式：种群分析，找出不合群的点
4. 高级模式：可以灵活的设置各种模型指标

为了简单起见，这里使用单一指标模式来创建ML Job

• 选择时间段，尽量选择足够长的时间段，提供足够多的数据喂给ML模型
• Aggregate： 聚合统计的方法，Count、High Count、Low Count、Mean、High Mean、Low Mean等，这里选择 Count方法， 表示统计特定时间窗口数据量，如果检测点的数据量超过模型计算值上下边界值的，都算作是异常点
• Bucket Span 统计的时间窗口大小

配置告警阈值

针对检测结果的严重程度进行邮件告警

查看检测结果

通过Anomaly Explorer 可以很容易找到异常的点

同时，通过Metric Viewer也很直观的查看异常周边的变化趋势，下面举例介绍一下，2019-09-19异常点的情况

从上图可以看出，在10:00/09:00/08:00/07:00 分别检测出了不同程度的异常，
其中severity值的大小，表明了异常的严重程度，severity值越大，说明这个异常越偏离正常情况。

probability是事情发生的概率，值越小，这个事件发生的可能性越小，从而说明如果这个事件发生了，肯定是不寻常的，跟上面severity反应的情况也是自洽的。

typical 是典型值，就是根据以往数据学习的模型，预测这个时间段的数据量是应该是多少

actual 是实际值，是这个时间段的统计的真实数据量，如果跟typical差异越大，出现异常的可能性就越大

产生结果解释

上述的检测结果来自真实业务场景，在7点到10点之间的业务调用方出现异常，导致在7点到10点之间的RPC服务日志产生量减少。ML 通过对以往数据的特征的学习，对这个时间段的日志量进行预测，发现实际结果与预测结果不符合。 在10点以后，业务调用方恢复正常，预测的结果跟实际调用量也是相吻合，在最终的结果图上也是反映回到正常曲线。

Bucket Span 调优

Bucket （桶） ，在进行实时数据流分析的时候，使用桶来区分指定时间窗口的数据，桶的大小，反映了每个窗口的数据量的大小， 也就是每次给ML喂的数据量的大小。

上面的结果是对桶大小进行调优后的效果，选择的Bucket Span 是5min， 最初选择的Bucket Span是15min， 没有把异常点很好的检测出来。

上图桶大小是15min，从图上可以看出来，没有把7点这个异常点检测出来，同时异常的重要程度不够。

设置1min的效果，同样不能把异常点检测很好的出来，而且受到噪音的干扰比较大，检测到很多严重程度很低的小的异常点。

调优参考：

• 桶的大小不宜设置太小，如果太小，虽然喂给ML的数据变多，但是在很小的时间段内，数据的变化比较频繁，产生大量的噪音，不利于模型的学习。反之，如果桶的大小设置太大了，将会丢掉很多细节信息，而且可能导致喂给ML的数据量不足，甚至把某个异常点给丢掉了。桶的大小要根据业务场景来，如上的场景，想尽快发现调用的异常，在数据量比较充足的情况下，设置5min是比较合适的。当然，这也不是绝对的情况，需要不断地调整这个值的大小，找到合适的值。
• ES ML 提供了一键预估（Estimate bucket span) Bucket大小，但是这个功能并不是很智能，不能完全依赖，还需要自己不断调试
• 尽量给ML喂更多充足的数据，数据量太少，ML不能很好的拟合数据，预测的结果也是不准确的
• 排除掉一些已知的“异常”数据，比如，周末的数据调用量下降很多，如果也直接喂给模型的话，会对预测结果产生干扰。好在，ES 提供了calendar功能，可以把特定时间段的移除，这个时间段的数据不算做异常。

• 使用多指标模式，选择更多的影响因子，使训练出来的模型更加健壮
• 由于数据不断变化，持续把更多更新的数据喂给模型，使其能够学习到新的特征，以适应新的数据

总结

一个完善的AIOps体系还有很多工作要做，上述只是整个体系中一小块，好在ES Stack现在越来越完善，下一步可以考虑把APM、Machine Learning整合起来，使用APM 对系统性能进行监控，使用ML对系统的各项指标进行建模，预测机器的使用趋势，并做到动态伸缩。

本文转载自: 掘金

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

作者简介

万汨，饿了么资深开发工程师。iOS，Go，Java均有涉猎。目前主攻大数据开发。喜欢骑行、爬山。

前言：针对“附近的人”这一位置服务领域的应用场景，常见的可使用PG、MySQL和MongoDB等多种DB的空间索引进行实现。而Redis另辟蹊径，结合其有序队列zset以及geohash编码，实现了空间搜索功能，且拥有极高的运行效率。本文将从源码角度对其算法原理进行解析，并推算查询时间复杂度。

要提供完整的“附近的人”服务，最基本的是要实现“增”、“删”、“查”的功能。以下将分别进行介绍，其中会重点对查询功能进行解析。

操作命令

自Redis 3.2开始，Redis基于geohash和有序集合提供了地理位置相关功能。
Redis Geo模块包含了以下6个命令：

• GEOADD: 将给定的位置对象（纬度、经度、名字）添加到指定的key;
• GEOPOS: 从key里面返回所有给定位置对象的位置（经度和纬度）;
• GEODIST: 返回两个给定位置之间的距离;
• GEOHASH: 返回一个或多个位置对象的Geohash表示;
• GEORADIUS: 以给定的经纬度为中心，返回目标集合中与中心的距离不超过给定最大距离的所有位置对象;
• GEORADIUSBYMEMBER: 以给定的位置对象为中心，返回与其距离不超过给定最大距离的所有位置对象。

其中，组合使用GEOADD和GEORADIUS可实现“附近的人”中“增”和“查”的基本功能。要实现微信中“附近的人”功能，可直接使用GEORADIUSBYMEMBER命令。其中“给定的位置对象”即为用户本人，搜索的对象为其他用户。不过本质上，GEORADIUSBYMEMBER = GEOPOS + GEORADIUS，即先查找用户位置再通过该位置搜索附近满足位置相互距离条件的其他用户对象。

以下会从源码角度入手对GEOADD和GEORADIUS命令进行分析，剖析其算法原理。

Redis geo操作中只包含了“增”和“查”的操作，并没有专门的“删除”命令。主要是因为Redis内部使用有序集合(zset)保存位置对象，可用zrem进行删除。

在Redis源码geo.c的文件注释中，只说明了该文件为GEOADD、GEORADIUS和GEORADIUSBYMEMBER的实现文件（其实在也实现了另三个命令）。从侧面看出其他三个命令为辅助命令。

GEOADD

使用方式

1

sql复制代码GEOADD key longitude latitude member [longitude latitude member ...]

将给定的位置对象（纬度、经度、名字）添加到指定的key。

其中，key为集合名称，member为该经纬度所对应的对象。在实际运用中，当所需存储的对象数量过多时，可通过设置多key(如一个省一个key)的方式对对象集合变相做sharding，避免单集合数量过多。

成功插入后的返回值：

1

bash复制代码(integer) N

其中N为成功插入的个数。

源码分析

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

scss复制代码/* GEOADD key long lat name [long2 lat2 name2 ... longN latN nameN] */
void geoaddCommand(client *c) {

//参数校验
    /* Check arguments number for sanity. */
    if ((c->argc - 2) % 3 != 0) {
        /* Need an odd number of arguments if we got this far... */
        addReplyError(c, "syntax error. Try GEOADD key [x1] [y1] [name1] "
                         "[x2] [y2] [name2] ... ");
        return;
    }

//参数提取Redis
    int elements = (c->argc - 2) / 3;
    int argc = 2+elements*2; /* ZADD key score ele ... */
    robj **argv = zcalloc(argc*sizeof(robj*));
    argv[0] = createRawStringObject("zadd",4);
    argv[1] = c->argv[1]; /* key */
    incrRefCount(argv[1]);

//参数遍历+转换
    /* Create the argument vector to call ZADD in order to add all
     * the score,value pairs to the requested zset, where score is actually
     * an encoded version of lat,long. */
    int i;
    for (i = 0; i < elements; i++) {
        double xy[2];

    //提取经纬度
        if (extractLongLatOrReply(c, (c->argv+2)+(i*3),xy) == C_ERR) {
            for (i = 0; i < argc; i++)
                if (argv[i]) decrRefCount(argv[i]);
            zfree(argv);
            return;
        }
    
    //将经纬度转换为52位的geohash作为分值 & 提取对象名称
        /* Turn the coordinates into the score of the element. */
        GeoHashBits hash;
        geohashEncodeWGS84(xy[0], xy[1], GEO_STEP_MAX, &hash);
        GeoHashFix52Bits bits = geohashAlign52Bits(hash);
        robj *score = createObject(OBJ_STRING, sdsfromlonglong(bits));
        robj *val = c->argv[2 + i * 3 + 2];

    //设置有序集合的对象元素名称和分值
        argv[2+i*2] = score;
        argv[3+i*2] = val;
        incrRefCount(val);
    }

//调用zadd命令，存储转化好的对象
    /* Finally call ZADD that will do the work for us. */
    replaceClientCommandVector(c,argc,argv);
    zaddCommand(c);
}

通过源码分析可以看出Redis内部使用有序集合(zset)保存位置对象，有序集合中每个元素都是一个带位置的对象，元素的score值为其经纬度对应的52位的geohash值。

double类型精度为52位；

geohash是以base32的方式编码，52bits最高可存储10位geohash值，对应地理区域大小为0.6*0.6米的格子。换句话说经Redis geo转换过的位置理论上会有约0.3*1.414=0.424米的误差。

算法小结

简单总结下GEOADD命令都干了啥：

1、参数提取和校验；

2、将入参经纬度转换为52位的geohash值（score）；

3、调用ZADD命令将member及其对应的score存入集合key中。

GEORADIUS

使用方式

1

css复制代码GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [ASC|DESC] [COUNT count] [STORE key] [STORedisT key]

以给定的经纬度为中心，返回目标集合中与中心的距离不超过给定最大距离的所有位置对象。

范围单位：m | km | ft | mi –> 米 | 千米 | 英尺 | 英里

额外参数：

• WITHDIST：在返回位置对象的同时，将位置对象与中心之间的距离也一并返回。距离的单位和用户给定的范围单位保持一致。

• WITHCOORD：将位置对象的经度和维度也一并返回。

• WITHHASH：以 52 位有符号整数的形式，返回位置对象经过原始 geohash 编码的有序集合分值。这个选项主要用于底层应用或者调试，实际中的作用并不大。

• ASC|DESC：从近到远返回位置对象元素 | 从远到近返回位置对象元素。

• COUNT count：选取前N个匹配位置对象元素。（不设置则返回所有元素）

• STORE key：将返回结果的地理位置信息保存到指定key。

• STORedisT key：将返回结果离中心点的距离保存到指定key。

由于 STORE 和 STORedisT 两个选项的存在，GEORADIUS 和 GEORADIUSBYMEMBER 命令在技术上会被标记为写入命令，从而只会查询（写入）主实例，QPS过高时容易造成主实例读写压力过大。
为解决这个问题，在 Redis 3.2.10 和 Redis 4.0.0 中，分别新增了 GEORADIUS_RO 和 GEORADIUSBYMEMBER_RO两个只读命令。

不过，在实际开发中笔者发现 在java package Redis.clients.jedis.params.geo 的 GeoRadiusParam 参数类中并不包含 STORE 和 STORedisT 两个参数选项，在调用georadius时是否真的只查询了主实例，还是进行了只读封装。感兴趣的朋友可以自己研究下。

成功查询后的返回值：

不带WITH限定，返回一个member list，如：

1

css复制代码["member1","member2","member3"]

带WITH限定，member list中每个member也是一个嵌套list，如：

1
2
3

css复制代码[	["member1", distance1, [longitude1, latitude1]]
	["member2", distance2, [longitude2, latitude2]]
]

源码分析

此段源码较长，看不下去的可直接看中文注释，或直接跳到小结部分

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

ini复制代码/* GEORADIUS key x y radius unit [WITHDIST] [WITHHASH] [WITHCOORD] [ASC|DESC]
 *                               [COUNT count] [STORE key] [STORedisT key]
 * GEORADIUSBYMEMBER key member radius unit ... options ... */
void georadiusGeneric(client *c, int flags) {
    robj *key = c->argv[1];
    robj *storekey = NULL;
    int stoRedist = 0; /* 0 for STORE, 1 for STORedisT. */

//根据key获取有序集合
    robj *zobj = NULL;
    if ((zobj = lookupKeyReadOrReply(c, key, shared.null[c->resp])) == NULL ||
        checkType(c, zobj, OBJ_ZSET)) {
        return;
    }

//根据用户输入（经纬度/member）确认中心点经纬度
    int base_args;
    double xy[2] = { 0 };
    if (flags & RADIUS_COORDS) {
		……
    }

//获取查询范围距离
    double radius_meters = 0, conversion = 1;
    if ((radius_meters = extractDistanceOrReply(c, c->argv + base_args - 2,
                                                &conversion)) < 0) {
        return;
    }

//获取可选参数 （withdist、withhash、withcoords、sort、count）
    int withdist = 0, withhash = 0, withcoords = 0;
    int sort = SORT_NONE;
    long long count = 0;
    if (c->argc > base_args) {
        ... ...
    }

//获取 STORE 和 STORedisT 参数
    if (storekey && (withdist || withhash || withcoords)) {
        addReplyError(c,
            "STORE option in GEORADIUS is not compatible with "
            "WITHDIST, WITHHASH and WITHCOORDS options");
        return;
    }

//设定排序
    if (count != 0 && sort == SORT_NONE) sort = SORT_ASC;

//利用中心点和半径计算目标区域范围
    GeoHashRadius georadius =
        geohashGetAreasByRadiusWGS84(xy[0], xy[1], radius_meters);

//对中心点及其周围8个geohash网格区域进行查找，找出范围内元素对象
    geoArray *ga = geoArrayCreate();
    membersOfAllNeighbors(zobj, georadius, xy[0], xy[1], radius_meters, ga);

//未匹配返空
    /* If no matching results, the user gets an empty reply. */
    if (ga->used == 0 && storekey == NULL) {
        addReplyNull(c);
        geoArrayFree(ga);
        return;
    }

//一些返回值的设定和返回
    ……
    geoArrayFree(ga);
}

上文代码中最核心的步骤有两个，一是“计算中心点范围”，二是“对中心点及其周围8个geohash网格区域进行查找”。对应的是geohashGetAreasByRadiusWGS84和membersOfAllNeighbors两个函数。我们依次来看：

• 计算中心点范围：

// geohash_helper.c

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

ini复制代码GeoHashRadius geohashGetAreasByRadiusWGS84(double longitude, double latitude,
                                           double radius_meters) {
    return geohashGetAreasByRadius(longitude, latitude, radius_meters);
}

//返回能够覆盖目标区域范围的9个geohashBox
GeoHashRadius geohashGetAreasByRadius(double longitude, double latitude, double radius_meters) {
//一些参数设置
    GeoHashRange long_range, lat_range;
    GeoHashRadius radius;
    GeoHashBits hash;
    GeoHashNeighbors neighbors;
    GeoHashArea area;
    double min_lon, max_lon, min_lat, max_lat;
    double bounds[4];
    int steps;

//计算目标区域外接矩形的经纬度范围（目标区域为：以目标经纬度为中心，半径为指定距离的圆）
    geohashBoundingBox(longitude, latitude, radius_meters, bounds);
    min_lon = bounds[0];
    min_lat = bounds[1];
    max_lon = bounds[2];
    max_lat = bounds[3];

//根据目标区域中心点纬度和半径，计算带查询的9个搜索框的geohash精度（位）
//这里用到latitude主要是针对极地的情况对精度进行了一些调整（纬度越高，位数越小）
    steps = geohashEstimateStepsByRadius(radius_meters,latitude);

//设置经纬度最大最小值：-180<=longitude<=180, -85<=latitude<=85
    geohashGetCoordRange(&long_range,&lat_range);
    
//将待查经纬度按指定精度（steps）编码成geohash值
    geohashEncode(&long_range,&lat_range,longitude,latitude,steps,&hash);
    
//将geohash值在8个方向上进行扩充，确定周围8个Box（neighbors）
    geohashNeighbors(&hash,&neighbors);
    
//根据hash值确定area经纬度范围
    geohashDecode(long_range,lat_range,hash,&area);

//一些特殊情况处理
    ……

//构建并返回结果    
    radius.hash = hash;
    radius.neighbors = neighbors;
    radius.area = area;
    return radius;
}

• 对中心点及其周围8个geohash网格区域进行查找:

// geo.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

ini复制代码//在9个hashBox中获取想要的元素
int membersOfAllNeighbors(robj *zobj, GeoHashRadius n, double lon, double lat, double radius, geoArray *ga) {
    GeoHashBits neighbors[9];
    unsigned int i, count = 0, last_processed = 0;
    int debugmsg = 0;

//获取9个搜索hashBox
    neighbors[0] = n.hash;
    ……
    neighbors[8] = n.neighbors.south_west;

//在每个hashBox中搜索目标点
    for (i = 0; i < sizeof(neighbors) / sizeof(*neighbors); i++) {
        if (HASHISZERO(neighbors[i])) {
            if (debugmsg) D("neighbors[%d] is zero",i);
            continue;
        }

	//剔除可能的重复hashBox (搜索半径>5000KM时可能出现)
        if (last_processed &&
            neighbors[i].bits == neighbors[last_processed].bits &&
            neighbors[i].step == neighbors[last_processed].step)
        {
            continue;
        }

	//搜索hashBox中满足条件的对象    
        count += membersOfGeoHashBox(zobj, neighbors[i], ga, lon, lat, radius);
        last_processed = i;
    }
    return count;
}


int membersOfGeoHashBox(robj *zobj, GeoHashBits hash, geoArray *ga, double lon, double lat, double radius) {
//获取hashBox内的最大、最小geohash值（52位）
    GeoHashFix52Bits min, max;
    scoresOfGeoHashBox(hash,&min,&max);

//根据最大、最小geohash值筛选zobj集合中满足条件的点
    return geoGetPointsInRange(zobj, min, max, lon, lat, radius, ga);
}


int geoGetPointsInRange(robj *zobj, double min, double max, double lon, double lat, double radius, geoArray *ga) {

//搜索Range的参数边界设置（即9个hashBox其中一个的边界范围）
    zrangespec range = { .min = min, .max = max, .minex = 0, .maxex = 1 };
    size_t origincount = ga->used;
    sds member;

//搜索集合zobj可能有ZIPLIST和SKIPLIST两种编码方式，这里以SKIPLIST为例，逻辑是一样的
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        ……
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
        zset *zs = zobj->ptr;
        zskiplist *zsl = zs->zsl;
        zskiplistNode *ln;

	//获取在hashBox范围内的首个元素（跳表数据结构，效率可比拟于二叉查找树），没有则返0
        if ((ln = zslFirstInRange(zsl, &range)) == NULL) {
            /* Nothing exists starting at our min.  No results. */
            return 0;
        }

	//从首个元素开始遍历集合
        while (ln) {
            sds ele = ln->ele;
		//遍历元素超出range范围则break
            /* Abort when the node is no longer in range. */
            if (!zslValueLteMax(ln->score, &range))
                break;
		//元素校验（计算元素与中心点的距离）
            ele = sdsdup(ele);
            if (geoAppendIfWithinRadius(ga,lon,lat,radius,ln->score,ele)
                == C_ERR) sdsfree(ele);
            ln = ln->level[0].forward;
        }
    }
    return ga->used - origincount;
}

int geoAppendIfWithinRadius(geoArray *ga, double lon, double lat, double radius, double score, sds member) {
    double distance, xy[2];

//解码错误, 返回error
    if (!decodeGeohash(score,xy)) return C_ERR; /* Can't decode. */

//最终距离校验(计算球面距离distance看是否小于radius)
    if (!geohashGetDistanceIfInRadiusWGS84(lon,lat, xy[0], xy[1],
                                           radius, &distance))
    {
        return C_ERR;
    }

//构建并返回满足条件的元素
    geoPoint *gp = geoArrayAppend(ga);
    gp->longitude = xy[0];
    gp->latitude = xy[1];
    gp->dist = distance;
    gp->member = member;
    gp->score = score;
    return C_OK;
}

算法小结

抛开众多可选参数不谈，简单总结下GEORADIUS命令是怎么利用geohash获取目标位置对象的：

1、参数提取和校验；

2、利用中心点和输入半径计算待查区域范围。这个范围参数包括满足条件的最高的geohash网格等级(精度) 以及 对应的能够覆盖目标区域的九宫格位置；（后续会有详细说明）

3、对九宫格进行遍历，根据每个geohash网格的范围框选出位置对象。进一步找出与中心点距离小于输入半径的对象，进行返回。

直接描述不太好理解，我们通过如下两张图在对算法进行简单的演示：

令左图的中心为搜索中心，绿色圆形区域为目标区域，所有点为待搜索的位置对象，红色点则为满足条件的位置对象。

在实际搜索时,首先会根据搜索半径计算geohash网格等级（即右图中网格大小等级），并确定九宫格位置（即红色九宫格位置信息）；再依次查找计算九宫格中的点（蓝点和红点）与中心点的距离，最终筛选出距离范围内的点（红点）。

算法分析

为什么要用这种算法策略进行查询，或者说这种策略的优势在哪，让我们以问答的方式进行分析说明。

• 为什么要找到满足条件的最高的geohash网格等级？为什么用九宫格？

这其实是一个问题，本质上是对所有的元素对象进行了一次初步筛选。 在多层geohash网格中，每个低等级的geohash网格都是由4个高一级的网格拼接而成（如图）。

换句话说，geohash网格等级越高，所覆盖的地理位置范围就越小。 当我们根据输入半径和中心点位置计算出的能够覆盖目标区域的最高等级的九宫格（网格）时，就已经对九宫格外的元素进行了筛除。 这里之所以使用九宫格，而不用单个网格，主要原因还是为了避免边界情况，尽可能缩小查询区域范围。试想以0经纬度为中心，就算查1米范围，单个网格覆盖的话也得查整个地球区域。而向四周八个方向扩展一圈可有效避免这个问题。

• 如何通过geohash网格的范围框选出元素对象？效率如何？

首先在每个geohash网格中的geohash值都是连续的，有固定范围。所以只要找出有序集合中，处在该范围的位置对象即可。以下是有序集合的跳表数据结构：

其拥有类似二叉查找树的查询效率，操作平均时间复杂性为O(log(N))。且最底层的所有元素都以链表的形式按序排列。所以在查询时，只要找到集合中处在目标geohash网格中的第一个值，后续依次对比即可，不用多次查找。 九宫格不能一起查，要一个个遍历的原因也在于九宫格各网格对应的geohash值不具有连续性。只有连续了，查询效率才会高，不然要多做许多距离运算。

综上，我们从源码角度解析了Redis Geo模块中 “增（GEOADD）” 和 “查（GEORADIUS）” 的详细过程。并可推算出Redis中GEORADIUS查找附近的人功能，时间复杂度为：O(N+log(M))。其中N为九宫格范围内的位置元素数量（要算距离）；M是指定层级格子的数量，log(M)是跳表结构中找到每个格子首元素的时间复杂度（这个过程一般会进行9次）。结合Redis本身基于内存的存储特性，在实际使用过程中有非常高的运行效率。

Reference

Redis命令参考

geohash

Redis中ZSET数据结构skiplist

阅读博客还不过瘾？

欢迎大家扫二维码通过添加群助手，加入交流群，讨论和博客有关的技术问题，还可以和博主有更多互动

博客转载、线下活动及合作等问题请邮件至 yidong.zheng@ele.me 进行沟通

本文转载自: 掘金

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

在上篇文章中，我们详细分析了doCreateBean()中的第2步：实例化bean，本文接着分析doCreateBean()的第4步“循环依赖处理”,也就是populateBean()方法。

首先回顾下CreateBean的主流程：

1. 如果是单例模式，从factoryBeanInstanceCache 缓存中获取BeanWrapper 实例对象并删除缓存
2. 调用 createBeanInstance() 实例化 bean
3. 后置处理
4. 单例模式的循环依赖处理
5. 属性填充
6. 初始化 bean 实例对象
7. 依赖检查
8. 注册bean的销毁方法

本章我们主要分析第4步：

一、循环依赖是什么？

循环依赖，其实就是循环引用，就是两个或者两个以上的 bean 互相引用对方，最终形成一个闭环，如 A 依赖 B，B 依赖 C，C 依赖 A。如下图所示：

Spring中的循环依赖，其实就是一个死循环的过程，在初始化 A 的时候发现依赖了 B，这时就会去初始化 B，然后又发现 B 依赖 C，跑去初始化 C，初始化 C 的时候发现依赖了 A，则又会去初始化 A，依次循环永不退出，除非有终结条件。

一般来说，Spring 循环依赖的情况有两种：

构造器的循环依赖。
field 属性的循环依赖。
对于构造器的循环依赖，Spring 是无法解决的，只能抛出 BeanCurrentlyInCreationException 异常表示循环依赖，所以下面我们分析的都是基于 field 属性的循环依赖。

在前文 Spring Ioc源码分析 之 Bean的加载（三）：各个 scope 的 Bean 创建 中提到，Spring 只解决 scope 为 singleton 的循环依赖。对于scope 为 prototype 的 bean ，Spring 无法解决，直接抛出 BeanCurrentlyInCreationException 异常。

为什么 Spring 不处理 prototype bean 呢？其实如果理解 Spring 是如何解决 singleton bean 的循环依赖就明白了。这里先留个疑问，我们先来看下 Spring 是如何解决 singleton bean 的循环依赖的。

二、解决singleton循环依赖

在AbstractBeanFactory 的 doGetBean()方法中，我们根据BeanName去获取Singleton 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
25
26
27
28
29

复制代码//DefaultSingletonBeanRegistry.java

@Nullable
protected Object getSingleton(String beanName, boolean allowEarlyReference) {
    // 从一级缓存缓存 singletonObjects 中加载 bean
    Object singletonObject = this.singletonObjects.get(beanName);
    // 缓存中的 bean 为空，且当前 bean 正在创建
    if (singletonObject == null && isSingletonCurrentlyInCreation(beanName)) {
        // 加锁
        synchronized (this.singletonObjects) {
            // 从 二级缓存 earlySingletonObjects 中获取
            singletonObject = this.earlySingletonObjects.get(beanName);
            // earlySingletonObjects 中没有，且允许提前创建
            if (singletonObject == null && allowEarlyReference) {
                // 从 三级缓存 singletonFactories 中获取对应的 ObjectFactory
                ObjectFactory<?> singletonFactory = this.singletonFactories.get(beanName);
                if (singletonFactory != null) {
                    //从单例工厂中获取bean
                    singletonObject = singletonFactory.getObject();
                    // 添加到二级缓存
                    this.earlySingletonObjects.put(beanName, singletonObject);
                    // 从三级缓存中删除
                    this.singletonFactories.remove(beanName);
                }
            }
        }
    }
    return singletonObject;
}

这段代码涉及的3个关键的变量，分别是3个级别的缓存，定义如下：

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

复制代码	/** Cache of singleton objects: bean name --> bean instance */
	//单例bean的缓存 一级缓存
	private final Map<String, Object> singletonObjects = new ConcurrentHashMap<>(256);

	/** Cache of singleton factories: bean name --> ObjectFactory */
	//单例对象工厂缓存 三级缓存
	private final Map<String, ObjectFactory<?>> singletonFactories = new HashMap<>(16);

	/** Cache of early singleton objects: bean name --> bean instance */
	//预加载单例bean缓存 二级缓存
	//存放的 bean 不一定是完整的
	private final Map<String, Object> earlySingletonObjects = new HashMap<>(16);

getSingleton()的逻辑比较清晰：

• 首先，尝试从一级缓存singletonObjects中获取单例Bean。
• 如果获取不到，则从二级缓存earlySingletonObjects中获取单例Bean。
• 如果仍然获取不到，则从三级缓存singletonFactories中获取单例BeanFactory。
• 最后，如果从三级缓存中拿到了BeanFactory，则通过getObject()把Bean存入二级缓存中，并把该Bean的三级缓存删掉。

2.1、三级缓存

看到这里可能会有些疑问，这3个缓存怎么就解决了singleton循环依赖了呢？

先别着急，我们现在分析了获取缓存的代码，再来看下存储缓存的代码。
在 AbstractAutowireCapableBeanFactory 的 doCreateBean() 方法中，有这么一段代码：

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

复制代码// AbstractAutowireCapableBeanFactory.java

boolean earlySingletonExposure = (mbd.isSingleton() // 单例模式
        && this.allowCircularReferences // 允许循环依赖
        && isSingletonCurrentlyInCreation(beanName)); // 当前单例 bean 是否正在被创建
if (earlySingletonExposure) {
    if (logger.isTraceEnabled()) {
        logger.trace("Eagerly caching bean '" + beanName +
                "' to allow for resolving potential circular references");
    }
    // 为了后期避免循环依赖，提前将创建的 bean 实例加入到三级缓存 singletonFactories 中
    addSingletonFactory(beanName, () -> getEarlyBeanReference(beanName, mbd, bean));
}

这段代码就是put三级缓存singletonFactories的地方，其核心逻辑是，当满足以下3个条件时，把bean加入三级缓存中：

• 单例
• 允许循环依赖
• 当前单例Bean正在创建

addSingletonFactory(String beanName, ObjectFactory<?> singletonFactory) 方法，代码如下：

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

复制代码// DefaultSingletonBeanRegistry.java

protected void addSingletonFactory(String beanName, ObjectFactory<?> singletonFactory) {
	Assert.notNull(singletonFactory, "Singleton factory must not be null");
	synchronized (this.singletonObjects) {
		if (!this.singletonObjects.containsKey(beanName)) {
			this.singletonFactories.put(beanName, singletonFactory);
			this.earlySingletonObjects.remove(beanName);
			this.registeredSingletons.add(beanName);
		}
	}
}

从这段代码我们可以看出，singletonFactories 这个三级缓存才是解决 Spring Bean 循环依赖的关键。同时这段代码发生在 createBeanInstance(...) 方法之后，也就是说这个 bean 其实已经被创建出来了，但是它还没有完善（没有进行属性填充和初始化），但是对于其他依赖它的对象而言已经足够了（已经有内存地址了，可以根据对象引用定位到堆中对象），能够被认出来了。

2.2、一级缓存

到这里我们发现三级缓存 singletonFactories 和 二级缓存 earlySingletonObjects 中的值都有出处了，那一级缓存在哪里设置的呢？在类 DefaultSingletonBeanRegistry 中，可以发现这个 addSingleton(String beanName, Object singletonObject) 方法，代码如下：

1
2
3
4
5
6
7
8
9
10
11

复制代码// DefaultSingletonBeanRegistry.java

protected void addSingleton(String beanName, Object singletonObject) {
	synchronized (this.singletonObjects) {
	        //添加至一级缓存，同时从二级、三级缓存中删除。
		this.singletonObjects.put(beanName, singletonObject);
		this.singletonFactories.remove(beanName);
		this.earlySingletonObjects.remove(beanName);
		this.registeredSingletons.add(beanName);
	}
}

该方法是在 #doGetBean(…) 方法中，处理不同 scope 时，如果是 singleton调用的，如下图所示：

也就是说，一级缓存里面是完整的Bean。
小结：

• 一级缓存里面是完整的Bean,是当一个Bean完全创建后才put
• 三级缓存是不完整的BeanFactory,是当一个Bean在new之后就put(没有属性填充、初始化)
• 二级缓存是对三级缓存的易用性处理，只不过是通过getObject()方法从三级缓存的BeanFactory中取出Bean

总结

现在我们再来回顾下Spring解决单例循环依赖的方案：

• Spring 在创建 bean 的时候并不是等它完全完成，而是在创建过程中将创建中的 bean 的 ObjectFactory 提前曝光（即加入到 singletonFactories 三级缓存中）。
• 这样，一旦下一个 bean 创建的时候需要依赖 bean ，则从三级缓存中获取。

举个栗子：

比如我们团队里要报名参加活动，你不用上来就把你的生日、性别、家庭信息什么的全部填完，你只要先报个名字，统计下人数就行，之后再慢慢完善你的个人信息。

核心思想：提前暴露，先用着

最后来描述下就上面那个循环依赖 Spring 解决的过程：

• 首先 A 完成初始化第一步并将自己提前曝光出来（通过 三级缓存 将自己提前曝光），在初始化的时候，发现自己依赖对象 B，此时就会去尝试 get(B)，这个时候发现 B 还没有被创建出来
• 然后 B 就走创建流程，在 B 初始化的时候，同样发现自己依赖 C，C 也没有被创建出来
• 这个时候 C 又开始初始化进程，但是在初始化的过程中发现自己依赖 A，于是尝试 get(A)，这个时候由于 A 已经添加至缓存中（三级缓存 singletonFactories ），通过 ObjectFactory 提前曝光，所以可以通过 ObjectFactory#getObject() 方法来拿到 A 对象，C 拿到 A 对象后顺利完成初始化，然后将自己添加到一级缓存中
• 回到 B ，B 也可以拿到 C 对象，完成初始化，A 可以顺利拿到 B 完成初始化。到这里整个链路就已经完成了初始化过程了

最后，为什么多例模式不能解决循环依赖呢？

因为多例模式下每次new() Bean都不是一个，如果按照这样存到缓存中，就变成单例了。

参考：

cmsblogs.com/?p=todo （小明哥）

本文转载自: 掘金

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

撰写我对java后端开发工程师选择方向的想法，写给在java后端选择转方向的人

背景

看到一些java开发工程师，对java后端薪酬太悲观了。认为换去大数据领域就会高工资。觉得java后端没有前途。我从事java后端开发，对大数据领域工作有些了解，但不深入。本文描述一下我对java后端和是否转大数据开发的个人见解。

目的

1. 分析大数据领域分类
2. 分析大数据工作工资高的原因
3. 分析造成觉得java后端开发不够前景的原因
4. java后端转大数据工作做什么
5. 转去大数据领域的各类方向与java后端比较衡量

点赞再看，关注公众号：【地藏思维】给大家分享互联网场景设计与架构设计方案
掘金：地藏Kelvin juejin.cn/user/104639…

一、大数据领域工作我认为分4类

二、大数据领域工资高的原因

大家看到大数据工资高，其实是大数据领域包含了建模或者算法工程师那部分。高工资的就只有推荐算法、用户画像、风控建模、自然语言这些工作，职位为算法或者建模工程师。

然而大数据领域的大部分工作，都是上图表中，第1、2类的工作，如：etl、爬虫、实时离线流式计算，es、顶多就机器学习。即使这些工作也只是工程级的应用（换句话说就是写业务代码，搬砖），如果工资高也是有架构能力（提升spark性能之类），而不是大数据应用开发。

三、分析造成觉得java后端开发不够前景的原因

有人觉得java后端开发工资低，没有前景，没有适应时代。

第一、大数据时代很久了，很早就开始招大数据了，不是需求火爆的状态，如安卓工程师一开始火，如现在做的人多了，像安卓变多了，大数据的应用开发就不像2014年刚开始的时候那么高工资了，但是大数据中算法、建模工程师依然高薪，那种要求高质量高的工作都是10个人里面只有1个会的那种。

第二、很多java后端开发都是业务开发，写好业务没bug渡过一天又一天，没有遇到好项目或者没有自主学习，导致做了很久的java开发工程师，都是做业务，写CRUD、redis、mq等，会写代码是一回事，但是有没有好的技术方案就是另外一回事。

四、Java后端转大数据工作做什么

java换去做大数据其实只能做etl、爬虫、实时离线流式计算，es、顶多就机器学习这些工程级的应用，也就换套工具写业务代码，换套工具搬砖而已。

因为Java开发人员多数是使用、应用程度，而不是研究程度，所以Java工程师转大数据很少有人会做到第3、4类的工作，如果做第3、4类估计是重新开始了。

其实第1、2类这些工作薪酬跟java后端没什么区别，毕竟两个领域都有纯业务搬砖和自带技术体系的人。

这些大数据工程级应用（第1、2类），也有架构组，如同java后端一样，也有业务架构和基础架构。其实如果积累经验java后端和这些大数据晋升我认为是一样的。

举例

假如表中的第2类，大数据工程级应用做spark、hadoop，一种是做应用开发，如双11在页面显示华为、小米等品牌实时出货量多少，就用实时流式计算。
另一种属于架构工作，如开发个数坊系统（也叫数据仓库、DataWareHouse）出来让大数据应用开发同事在上面做 OLAP。这些架构组的人，一般需要对hadoop、spark、presto源码有过研究，或许会在上面二次开发，或者进行性能优化工作。
前者是换套工具搬砖，后者是架构组。如同java也有些业务代码和架构设计。

五、转去大数据领域的各类方向与java后端比较衡量

考虑方向

• 要么转做大数据架构，如研究spark、hadoop、presto，搞个数坊系统（又叫DataWareHouse、数据仓库）、shuffle调优等，毕竟属于架构组，工资会高一点。
• 要么转做推荐算法、用户画像、建模/算法类。而这部分工作都是有要求的，算法过硬、研究生、985、211 、数学专业，这些工作也会更高。数据挖掘与分析不止会mathot、spark streaming，还有SAS/SPSS 。
• 如果转做大数据应用做实时流式计算、离线流式计算、es分词统计，其实是相当于业务码农，如果有java后端开发经验的话，这种那还不如在java后端继续深耕，毕竟换去做大数据应用开发深耕也是一样的。

考虑晋升机会

• 考虑另一部分，能晋升到领导位置的，一般是伴随公司成长的核心员工。公司成长，开始是业务，一般都是java后端业务代码。等到中期、后期做报表才会用上大数据业务开发（第1、2类），有性能问题就会有架构组，再后期才到推荐算法这些让app更好体验的东西，如淘宝首页推荐。所以业务架构在前期就比较容易晋升。
• 等公司成长起来了，公司有钱自然就会招很好的算法、建模工程师做真正有价值的部分。
  而实时流式计算、elastic-search这些业务码农，也只是搬砖，现在做的人像安卓一样多了，就不像2014年刚开始的时候那么高工资了。

考虑所在城市的岗位数量

如第3、4类工作，岗位比较少，换公司换工作是否方便，有些公司如：中国移动 的第3类大数据工作就有外包出去，不是正式编制。
画好跳槽路线，因为转行第一间不一定是你的终点，所以要看其他的更上流的企业的要求是否能匹配自己。

BackUp作用

• 多学大数据只是防止当前公司业务停止，没有业务开发时，java后端开发工程师可能被裁员掉，学大数据和前端React.js类只是对于java后端开发另谋活路的backup。因为有些职位就希望你全栈，但现在很多都前后端分离的。
• 而被淘汰掉的java后端只是写业务代码，用用redis、mq。
• java后端人人都会写，java后端技术领域还是很广的，但有没有写出好的技术方案就另外一回事。

总结

大数据、前端页面开发对于java后端开发工程师来讲，我觉得了解就可以了，知道有解决办法，不必每个领域都精通，况且没办法每个领域都精通。

如果后端开发转去做大数据、项目经理、产品经理岗位，估计都是java后端技术没做上去（本身不喜欢做程序员的也有可能），或者是只会做纯业务代码这些被淘汰掉了，所以就换领域了，还有转hr的。
不过同级别的java后端开发和产品经理薪资确实有差距，估计一两千。

我觉得大数据工程级应用开发（第1、2类）和Java后端开发薪资就没什么差距，以前java后端能转大数据应用开发，是因为那时候还缺人，现在不缺人了，要招都是招有真实经验的。

如果你从事java后端开发几年了，要转大数据领域，相当于你有一个升高级java开发工程师的机会，还是选择中级大数据应用开发工程师的机会，反正都是写业务代码的。

如果你的条件过硬，如985/211学历、数学专业、算法研究经验，如果要转算法/建模工程师就早点转，大数据领域高工资的就是这类人。

如果java后端开发工作经验4以上年了，没有硬性条件，建议继续深入后端学习。

如果java后端开发工作一两年，你想怎么转都可以。

如想了解薪酬，可以在招聘网站搜大数据工程师（一般就是指第1、2类的），和算法工程师、风控建模工程师、推荐算法工程师、用户画像工程师。我所知道有个风控建模经理三万多。

欢迎留言跟我讨论

欢迎关注

我的公众号 ：地藏思维

掘金：地藏Kelvin

简书：地藏Kelvin

CSDN：地藏Kelvin

我的Gitee： 地藏Kelvin gitee.com/dizang-kelv…

本文转载自: 掘金

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

前言

前段时间做某个项目，由于涉及到tcp/udp方面的知识比较多，于是就索性趁热打铁，写个工具来强化相关知识。另外由于并非十分擅长Golang，所以也顺便再了解下Golang吧。

简介

flynet 是一款Golang语言编写的命令行工具，目前支持的功能包括：

• Http代理
• 本地Socks5代理
• C/S模式的Socks5代理，支持TCP/UDP方式
• 内网穿透
• …
  项目目前分为clien端和sever端，除http、本地socks5代理两端都支持外，其余功能需要两端配合使用。

使用方式

安装

Windows、linux用户可以直接在Releases页面下载对应的版本即可，其他平台可自行下载源码编译。

Windows中命令行进入到相应目录，.\win-client.exe ...或 .\win-server.exe ...

Linux中同样的， ./linux-server ...或./linux-client ...

在下文中皆以server ...或client ...表示。

尝试运行后，如果输出如下信息表示成功：

1
2
3
4
5
6
7
8
9
10
11

复制代码Usage: flynet [options]
  -M, --mode        choose which mode to run. the mode must be one of['http', 'socks5',
                    'socks5-tcp', 'socks5-udp', 'forward']
  -L, --listen      choose which port(s) to listen or forward
  -S, --server      the server address client connect to
  -V, --verbose     output detail info
  -l, --log         output detail info to log file
  -H, --help        show detail usage

Mail bug reports and suggestions to <asche910@gmail.com>
or github: https://github.com/asche910/flynet

Http代理

http代理直接在本机上开启Http代理，client和server都支持，命令如下：

1

复制代码server -M http -L 8848

或

1

复制代码client -M http -L 8848

表示在本机8848端口上开启了Http代理服务，如果没有任何信息输出则表示启动成功，毕竟linux的一大哲学就是：

没有消息就是好消息

当然如果还是想看到消息的话，可以在后面加上 -V或--verbose参数，这样的话就会输出很多消息了。或者也可以加上-l或--log参数来启动日志文件，会在运行目录下生成一个 flynet.log文件。

本地Socks5代理

本机上开启socks5代理的话，也是非常简单的，client和server都支持，命令如下：

1

复制代码server -M socks5  -L 8848

或

1

复制代码client -M socks5 -L 8848

这就表示在本机8848端口上开启了socks5代理，然后Chrome配合SwitchyOmega就可以很好的上网了。

C/S模式的Socks5代理-TCP

前面的那个是在本地上的socks5代理，这个则是client和server相互配合的socks5代理，并且中间是以tcp协议传输。用途的话，自由发挥吧。使用方法如下：

服务端

1

复制代码server -M socks5-tcp -L 8888

客户端

1

复制代码client -M socks5-tcp -L 8848 -S asche.top:8888

这里的例子是假设我服务器域名为 asche.top，然后客户端在8848端口开启了socks5代理，然后流量是以TCP的方式转发到了服务器的8888端口上，交由服务器去请求相应的目标网站，再把请求结果返回给客户端。如果可以，中间流量再进行加密，保证了传输的安全性。

C/S模式的Socks5代理-UDP

这个和上面tcp那个非常相似，不同的是这个使用UDP报文进行传输。毕竟UDP在某些方面有它自身的优势，而且某些重要的协议主要使用udp传输，比如DNS协议。下面来介绍具体用法：

服务端

1

复制代码server -M socks5-udp -L 53

客户端

1

复制代码client -M socks5-udp -L 8848 -S asche.top:53

这里同样以域名asche.top、端口53为例，客户端在8848端口开启了socks5代理，然后所有流量通过udp方式传输到服务端的53端口上，服务端收到后解析请求，然后将所有请求发至目标网站，再将结果以udp方式返回到客户端。同样的是中间传输也进行了加密。

内网穿透

内网穿透，即NAT穿透，网络连接时术语，计算机是局域网内时，外网与内网的计算机节点需要连接通信，有时就会出现不支持内网穿透。就是说映射端口，能让外网的电脑找到处于内网的电脑，提高下载速度

简单点说就是让外网能够访问到内网中的机器。这里该工具所做的就是将内网的某个端口映射到服务器的某个端口中去，这样通过访问服务器的某个端口就可以间接的访问到内网中的端口了。方法如下：

服务端

1

复制代码server -M forward -L 8888 8080

客户端

1

复制代码server -M forward -L 80 -S asche.top:8888

同样假设服务器域名为asche.top, 这样所完成的就是将客户端的80端口映射到了服务端的8080端口上，中间的数据传输是通过服务端监听8888来完成的。然后我们访问asche.top:8080看到的内容应该就是客户端80端口上的内容了。

结语

项目目前功能也比较局限，日后应该会加上更多功能。另外地址位于 flynet, 还望大家多多支持！

本文转载自: 掘金

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