Flink-1.11 7天训练营

发表于 2020-07-13 | 分类于 flink

https://flink-learning.org.cn/

CDC：在数据迁移场景非常实用

PyFink：提升对Pandas UDF的支持，扩展python在支持分布式上的能力

1.1、流批的本质

有界的数据是批，无界的数据是流

海量数据处理三辆马车

GFS

BigTable

MapReduce

对应的开源框架

Hadoop

流与批的区别：业务延时的不同

流是批的特例，Flink（Native-Streaming）

批是流的特例，Spark（Micro-Batching）

1.2、流计算的核心问题

问题：

延问题
更新撤回：
容错续跑：机器故障以后，可以继续跑
透明升级：某个Job升级以后，可以继续前面的任务跑
乱序问题：数据到达顺序问题
正确性问题：
部署问题
弹性扩容问题

延时问题

Native Stream + Early-Fire

更新撤回

告诉下游节点上一次计算结果无效

使用”-“标记上一次计算结果失效

容错续跑

透明升级

乱序问题

数据产生时间 EventTime

数据进入流计算引擎的时间 Watermark

正确性问题

数据是否有丢失
数据是否只参与了一次计算

1.3、Flink应用场景

事件驱动型应用

社交：关注事件引发的后续操作

网购：恶意差评事件 -> 封号

金融：欺诈 -> 反欺诈

数据分析型应用

数据管道型应用

数据清洗

1.4、怎样理解流批一体融合？

用户、运行、运维三个维度看

第二章 Stream Process 在 Flink 的实现

2.1、并行处理和编程范式

任务划分类似有向无环图（DAG）的

2.2、DataStream API 概览

批处理的数据分区类似于洗牌，将相关的牌放在一起

流处理分区是随着数据的到来动态完成的

2.3、状态和时间

时间的使用案例：

运行结果：

作业框架，基于框架进行开发

【Stream Processing with Apache Flink】
讲师：崔星灿，Apache Flink Committer
时间：7月14日 20:00-21:00（UTC+8）
【作业内容】
- 准备环境：能clone和运行源码中的streaming-examples
- 下载地址：https://github.com/flink-china/Flink-Geek-Training/blob/master/Training2.java

第三章运行时架构

3.1 、Runtime 总览

三大核心组件

3.2、 JobMaster — 作业的控制中心

Lost Leadership：JobMaster出问题了，异常终止

Eager：服务于流式处理作业，节省调度花费的时间；
Lazy from sources：服务于批处理作业，延迟下游资源的调度，避免空转浪费资源；
（WIP）Piplined region based：类似Lazy from sources，既能减少调度时间，又能避免空转浪费资源；

出错恢复策略与 Pipiline 有关，出错后，会重新pipline里相关的任务

单点重启：只重启出错的节点，接下来的版本会牺牲数据一致性来实现

3.3、TaskExecutor — 任务的运行容器

3.4、ResourceManager — 资源管理中心

第四章流计算的状态和容错机制

4.1、流计算中的状态

状态的种类

4.2、全局一致性快照

第三张图，后发生的事件包含在快照中，而先发生的事件没有包含在快照中，所以不是全局一致性快照

全局一致性快照的实现方法

异步全局一致性快照算法

Chandy-Lamport 算法流程

实线代表：开始快照

虚线代表：结束快照

A 是自己内部发生的事件，与其他进程没有交互，我们认为是P1自己发给自己的消息

4.3、Flink的容错机制

如何保证 Exactly once

需要一个可回退的 source

复杂场景 —— 多流输入

降低一致性要求

通过异步快照减少停顿

4.4、Flink的状态管理

定义一个状态

本地状态后端 JVM

形成快照的过程需要序列化

优点：

缺点：

RocksDB 作为状态后端

第5章 Flink SQL/Table API 介绍与实战

5.1、Flink SQL

声明式：用户只需要表达他需要什么，不需要关心怎么计算的

批流统一：SQL很容易做批处理，不管输入是静态的批数据，还是动态的流数据，结果是相同的

sql/python/scala/java 的api 被翻译成 Logical Plan，Logical Plan 被优化器优化成 Physical Plan，然后翻译成 Transformations 的DAG，再交给 JobGraph 执行

完整的类型系统：数据的精度
TopN：
高效流式去重：在明细层去重，交给汇总层的时候才会精确
维表关联：Mysql、hive、hbase
cdc：对接canal等 binlog的数据
内聚函数：超过230个内置函数
MiniBash：
多种解热点手段：
完整的批处理支持
Python Table API：多语言支持
Hive集成：读写hive，支持hive sql 语法

5.3 实战Demo

前提条件：需要基础的 SQL 知识

准备环境：你将需要一个至少 8 GB 内存和安装了 Docker 的笔记本电脑。

下载地址：https://www.docker.com/get-started

代码：https://github.com/wuchong/flink-sql-demo

6.1、PyFlink简介

6.2、PyFlink 功能介绍

Python Table API

Python UDF

Python UDF 架构

向量化 Python UDF

Python UDF Metrics

Python UDF 执行优化

6.4、PyFlink 下一步规划

第七章 Flink Ecosystems 连接外部生态系统

7.1、连接外部系统

通过DDL创建的表是如何被使用的：

7.2、常用Connector

Kafka Connector

Elasticsearch Connector

FileSystem Connector

Hive Connector

DataGen Connector

Print Connector

BlackHole Connector

可以做性能测试，数据来了以后不做任何处理，直接丢弃

7.3、示例 & Demo

kafka交互

写入es

hive交互

未命名

发表于 2020-07-12

k8s简介和安装

k8s核心概念：Pod、控制器、Services、网络通信

k8s 资源清单及编写、Pod生命周期

k8s 存储类型

k8s 调度器及HELM

k8s 集群安全机制

k8s简介和安装

1、k8s简介

IaaS：Infrastructure as a Service

PaaS：Platform as a Service（类似：阿里云、腾讯云）

SaaS：Software as a Service（类似：腾讯文档、有道云笔记等产品）

软件部署演进路线

3、基础知识

4、安装配置

系统要求：

64位操作系统 lInux 内核3.10以上，建立4.4以上内核（Ubuntu/Centos7+），3台

cpu至少2核，最好4核

内存最少2G，推荐8G

etcd 3.0版本

docker 18.03 版本及以上

Flannel

Kubernetes 1.18.5

内核升级

1 2	cat /etc/redhat-release uname -r

k8s核心概念：Pod、控制器、Services、网络通信

kubernetes 对象

基本对象

Pod

网络

存储

高级对象

扩容缩容

1	kubectl scale deployment nginx-deploy --replicas=5

查看节点分布

1	kubectl get po -o wide

自动扩容缩容

1	kubectl autoscale deployment nginx-deploy --min=3 --max=7 --cpu-percent=60

滚动升级

1 2	docker pull nginx:1.18.0 docker pull nginx:1.19.1

kubectl delete deployment nginx-deploy
kubectl get deployment
kubectl get pod
kubectl create -f nginx-deploy.yaml
kubectl describe deployment nginx-deploy

1 2	kubectl set image deployment/nginx-deploy nginx=docker.io/nginx:1.19.1 kubectl get rs

回滚

1
2
3

kubectl rollout undo deployment/nginx-depoly
# 回滚到某个版本
kubectl rollout undo deployment/nginx-depoly --to-reversion=2

AB测试

DaemonSet

确保集群中的每个pod有且只有一个副本

Filebeat、Logstash、Flume （agent）

Promethues node Exporter 监控每台机器的CPU、内存指标

k8s 资源清单及编写、Pod生命周期

1 2	kubectl api-versions kubectl api-resouces

查看label

修改label（修改后，又重新创建了一个新pod，补齐3个nginx）

查看每个对象有哪些属性

kubectl explain deployment
kubectl explain deployment.status
kubectl explain pod
kubectl explain kind
kubectl explain apiVersion
kubectl explain metadata

编写yaml文件 nginx-daemon.yaml

kind: DaemonSet
apiVersion: apps/v1
metadata:
  name: nginx-daemon
spec:
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx
        ports:
        - containerPort: 80

Job

kind: Job
apiVersion: batch/v1
metadata:
  name: job-demo
spec:
  template:
    metadata:
      name: job-demo
    spec:
      containers:
      - name: job-demo
        image: docker.io/busybox:latest
        command: ["echo", "'scale=10; 4*a(1)'", "|", "bc", "-l"]
      restartPolicy: Never

Pause 是根容器

CronJob

Pod的生命周期

ImagePullBackOff ，镜像拉取失败，采用退避算法重新拉取镜像

pending
running
succeeded
failed
unknown

Init Container 按顺序执行

Main Container

探针：存活检测

启动延迟5秒，检测周期3秒

nginx 一般都是80端口，去检测8008端口，是不存在的，会认为容器启动失败，会再启动一个容器

第二个例子：

启动后先休眠60秒，移除a文件，再休眠

检测a文件是否存在，不存在则删除pod，再创建新的Pod

postStart

preStop

探针：读取检测

读取检测，如果能读取到，则代表容器达到了ready状态，可以供外界访问，否则就被认为不可访问

Service

代理模式

暴露固定的ip，端口，供外界访问

1
2
3

kubectl create -f nginx-deploy.yaml
kubectl get deploy
kubectl expose deploy nginx-deploy --port=8000 --target-port=80

删除暴露

1	kubectl delete

域名格式

k8s 转发模式

linux 内核态，用户态

userspace

k8s 1.4 之前使用 iptables

ipvs（ipv server）虚拟ip地址服务 VIP 在1.8-beta版本里加入使用，在1.14版本里成为正式的转发协议

iptables 协助维护内核netfilter中的各种路由

Namespace

做资源隔离

6种命名空间

挂载文件系统：mount
进程：pid
网络Network：不同namespace之间的网络是不通的
IPC 进程间通信
UTS：隔离主机名和 NIS域名
User：

Cgroup

cgroup：linux 2.6以后出现的，google工程师导入linux内核的

基于 cpuset 开发

作用：做资源管控

cpuset

linux操作系统启动以后，会mount一个cpu的子系统，我们可以对cpu的子系统做各种各样的操作和定制。这样我们就可以管控每个应用程序使用资源的多少了

# 安装工具管理 cgroup
yum install -y libcgroup-tools
cgcreate -g cpu:/test
cgget
cgset

虚拟设备对就相当于虚拟网卡之间的桥

# 列出所有网络命名空间
ip netns list
ip netns add test1
ip a
# 定义虚拟设备对
ip link add veth0 type veth peer name veth1
# 将veth1转给test1
ip link set veth1 netns test1
# 指定test1 的网段
ip netns exec test1 ip addr add 10.0.5.5/24 dev veth1
ip netns exec test1 ip link set veth1 up
ip netns exec test1 ip link set lo up
ip netns exec test1 ip addr show
# 
ip addr add 10.0.5.6/24 dev veth0
ip link set veth0 up

k8s 各个节点之所以能互相访问是因为每个节点上都部署了 kube-proxy

上图中的架构，kube-proxy 太繁忙了，所以可以优化成iptables转发

但iptables转发规则太多的时候又不利于维护，所以又引入了ipvs （VIP server）

yum search ipvs
yum install -y ipvsadm.x86_64
ipvsadm -Ln
ipvsadm --clear

cm - ConfigMap 配置中心，涉及k8s的很多配置，都是用configMap来做的

通过配置可以选择使用 iptables 还是 ipvs

服务类型

32659 是宿主机ip，指向容器的8000端口

在yaml文件中可指定端口

DNS

ingress

k8s 存储类型

configMap

pv/pvc

pv：Persistence Volume

pvc：Persistence Volume Claim（声明）

pv 支持 filesystem 和 blockDevice

1	kubectl get pv -n devops

如果pvc没有指定pv，则按照最小匹配原则匹配PV

访问模式

NFS：以linux方式搭建
RWO (Read-Write-Once)：单个节点读写
ROX (Read-Only-Many)：多个节点只读
RWX（Read-Write-Many）：多个节点读写

回收策略

Retain：手动回收
Recycle：基本擦除（rm -rf /somevolume/*）
Delete：和会提供商有关

PV生命周期

Avaliable：空闲没有被绑定
Bound：已经被绑定
Release：已经被释放，但没有重新声明
Failed：回收失败

k8s 生命周期

Pending
ContainerCreating
Running

k8s 调度器及HELM

亲和力

节点亲和力

硬亲和：requiredDuringSchedulingIgnoredDuringExecution。如果节点label改了，服务会被驱逐

软亲和：perferredDuringSchedulingIgnoredDuringExecution。如果节点label改了，服务还能迁移到其他节点

pod 亲和力

驱逐（排除）

污点节点taint

让pod部署不到这个节点上

NoSchedule：Pod 不会被分配到该节点
PerferNoSchedule：尝试不要将pod分配到该节点
NoExecute：Pod肯定不会被分配到该节点，部署在该节点的pod会被驱逐

加污点

去除污点

容忍tolerations

容忍是对于pod来说的

k8s 集群安全机制

configMap 配置k8s 程序配置非加密

secret

ServiceAccount token

opaque passwork 密钥

走出微服务误区：避免从单体到分布式单体

发表于 2020-07-08 | 分类于 micro-service

作者 | 敖小剑

最近，社区频繁出现各种对微服务的质疑和反思声音，甚至放弃微服务回归单体。鉴于此，本文从“分布式单体”问题出发，介绍通过引入非侵入式方案和引入 Event/EDA 来走出微服务实践误区——从单体到微服务，最后却沦为分布式单体。

1、回顾：从单体到微服务到 Function

在过去几年间，微服务架构成为业界主流，很多公司开始采用微服务，并将原有的单体应用迁移到微服务架构。从架构上说，微服务和单体之间最大的变化在于微服务架构下应用的粒度被“拆小”：将所有业务逻辑都集中在一起的单体应用，按照领域模型拆分为多个内聚而自治的“更小”应用。而 Function 则在拆分上更进一步，拆分粒度变成“单个操作”，基于 Function 逐渐演进出现了 FaaS 形态和 Serverless 架构。

在微服务和 Serverless 喧嚣中，业界逐渐出现很多质疑和反对的声音：越来越多的人发现，当他们兴冲冲的迁移单体应用到微服务和 Serverless 架构后，收益却并没有期望中的那么理想。而最近，也出现了一些对微服务的各种质疑、反思的声音，甚至放弃微服务回归单体。举例，我在Infoq 中国网站简单搜索关键字“微服务”，前三页中就出现了如下内容：

我们为什么停用微服务？

这些公司为什么放弃微服务？

https://www.infoq.cn/article/KSzctluch2ijbRbKYBgO

致传统企业朋友：不够痛就别微服务，有坑

https://www.infoq.cn/article/Nd0RofAUp0WtlvlQArbu

微服务带来的心理阴影

Uber 团队放弃微服务改用宏服务，网友评论炸锅了

为什么 Segment 从微服务回归单体

无论是支持还是反对微服务的声音，大多都是着眼于组织架构（康威定律，对应用和代码的 ownership）、微服务拆分（粒度大小，如何识别领域模型和业务边界）、分布式事务（跨多个微服务调用时维持一致性），工具（自动化构建、部署、可测试性、监控、分布式链路跟踪、CI/CD），数据库分离（避免多个微服务，尤其是领域模型外的微服务共享数据库）等方面进行合理性分析和观点阐述，相信大家都对这些问题都有了解。

而我今天的文章，将从另外一个角度来看待微服务（也包括 Serverless）实践中存在的误区——辛辛苦苦从单体走到微服务，却最后沦为分布式单体。

2、分布式单体

“Distributed Monolith”，分布式单体，这真是一个悲伤的技术术语。而这偏偏是企业采用微服务后通常最容易掉进去的一个“陷阱”。事实上，我看到的很多微服务落地最终都是以”分布式单体”收场，无法获得微服务的完整收益。

问题源于微服务实施的方式 —— 按照业务逻辑拆解单体，划分为多个微服务，定义 API 接口，然后通过 REST 或者 RPC 进行远程调用，最终把这些微服务组合起来提供各种业务功能。简单说，就是在业务拆分的基础上，用进程间的远程调用简单替代原来进程内的方法调用。期间，对于原来使用的各种分布式能力，继续采用之前的方式。简单说：方式不变，只是粒度变小。

从方法论说，这样做无可厚非，这也是微服务采用过程中非常标准的做法。但问题在于，止步于此是不够的 —— 至少存在两个有待继续努力改进的地方。

分布式单体起因之一：通过共享库和网络客户端访问分布式能力

分布式能力的共享库和网络客户端是造成分布式单体问题的原因之一，关于这一点，来自 verizon 的 Mohamad Byan 在他的名为 Avoid the Distributed Monolith!! 的演讲中有详细阐述，我这里援引他的图片和观点：

https://www.slideshare.net/DevOpsDaysDFW/avoid-the-distributed-monolith

上图是微服务体系的逻辑架构，由两部分组成：

内层架构（图上浅蓝色部分），是每个微服务的实现架构；
外层架构 (图上黄色部分)，是构建强大微服务架构所需要的各种能力。这里通常有大家熟悉的各种分布式能力。

特别提示：这里说的“网络客户端”是各种分布式能力的客户端，如服务注册发现 /MQ 中间件 /Redis 等 key-value 存储 / 数据库 / 监控日志追踪系统 / 安全体系等，不是服务间通讯如 RPC 的客户端。

而内层的微服务是通过共享类库和网络客户端来访问外层架构提供的分布式能力：

分布式能力的共享类库和网络客户端会迫使内层微服务和外层架构的各种分布式能力之间产生强耦合，增加运维的复杂性（如升级困难造成版本碎片化），多语言受限于类库和网络客户端支持的语言，各种组件（如消息中间件）往往使用自定义数据格式和通讯协议 —— 所有这些迫使内层微服务不得不实质性受限于外层架构的技术选型。

对于 Function，这个问题就更加明显：Function 的粒度更小，更专注业务逻辑。某些简短的 Function 可能只有几百行代码，但是，为了让这几百行代码运转起来而需要引入的共享类库和网络客户端可能相比之下就规模惊人了。援引一张网上图片作为示意：

分布式单体起因之二：简单用远程调用替代进程内方法调用

在微服务架构改造过程中，熟悉单体系统和架构的开发人员，习惯性的会将这些单体时代的知识和经验重用到新的微服务架构之中。其中最典型的做法就是：在遵循领域模型将现有单体应用按照业务边界拆分为多个微服务时，往往选择用 REST 或者 RPC 等远程调用方式简单替代原有的进程内方法调用。

当两个逻辑上的业务模块存在协作需求时：

从单体到微服务，直接方法调用被替换为远程调用（REST 或者 RPC），即使采用 Service Mesh 也只是在链路中多增加了 sidecar 节点，并未改变远程调用的性质：

这导致了前面所说的 “分布式单体”：

在微服务之前：应用程序由多个耦合在一起的模块组成，这些模块通过内存空间进行方法调用…..
在微服务之后：应用程序由多个耦合在一起的微服务组成，这些微服务通过网络进行远程调用…..

抛开调用方式的差异来看采用微服务前后的系统架构，会发现：两者几乎是完全一样的！！

而微服务版本在某些情况下可能表现的更糟糕：因为调用方式更脆弱，因为网络远比内存不可靠。而我们将网络当成 “胶水” 来使用，试图把分散的业务逻辑模块（已经拆分为微服务）按照单体时代的同样方式简单粘在一起，这当然比单体在同一个进程内直接方法调用更加的不可靠。

关于这一点，在”The Eight Fallacies of Distributed Computing/ 分布式计算的 8 个谬论”一文中有详细阐述。

https://www.red-gate.com/simple-talk/blogs/the-eight-fallacies-of-distributed-computing/

类似的，在采用 Function 时，如果依然沿用上面的方式，以单体或微服务架构的思维方式和设计模式来创建 FaaS/Serverless 架构：

其本质不会发生变化 —— 不过是将微服务变成粒度更小的函数，导致系统中的远程调用数量大为增加：

系统内的耦合并没有发生变化，Serverless 并不能改变微服务中存在的这个内部耦合问题：调用在哪里，则耦合就在哪里！只是把将组件的粒度从 “微服务“换成了 “Function/ 函数”。

耦合的存在是源于系统不同组件之间的通讯模式，而不是实现通讯的技术。

如果让两个组件通过“调用”（后面再展开讲何为调用）进行远程通信，那么不管调用是如何实现的，这两个组件都是紧密耦合。因此，当系统从单体到微服务到 Serverless，如果止步于简单的用远程调用替代进程内方法调用，那么系统依然是高度耦合的，从这个角度来说：

单体应用 ≈ 分布式单体 ≈ Serverless 单体

分布式单体起因小结

上面，我们列出了微服务和 Serverless 实践中容易形成“分布式单体”的两个主要原因：

通过共享库和网络客户端访问分布式能力；
简单用远程调用替代进程内方法调用。

下面我们针对这两个问题探讨解决的思路和对策。

3、引入非侵入式方案：物理隔离 + 逻辑抽象

前面谈到分布式单体产生的一个原因是“通过共享库和网络客户端访问分布式能力”，造成微服务和 Lambda 函数和分布式能力强耦合。以 Service Mesh 为典型代表的非侵入式方案是解决这一问题的有效手段，其他类似方案有 RSocket / Multiple Runtime Architecture，以及数据库和消息的 Mesh 化产品，其基本思路有两点：

委托：通过 Sidecar 或者 Runtime 来进行对分布式能力的访问，避免应用和提供分布式能力的组件直接通讯造成强绑定 —— 通过物理隔离进行解耦。
抽象：对内层微服务隐藏实现细节，只暴露网络协议和数据契约，将外围架构的各种分布式能力以 API 的方式暴露出来，而屏蔽提供这些能力的具体实现 —— 通过逻辑抽象进行解耦。

以 Service Mesh 的 Sidecar 为例，在植入 Sidecar 之后，业务应用需要直接对接的分布式能力就大为减少（物理隔离）：

最近出现的 Multiple Runtime / Mecha 架构，以及遵循这一架构思想的微软开源产品 Dapr ，则将这个做法推进到服务间通讯之外更多的分布式能力。

此外在委托之外，还提供对分布式能力的抽象。比如在 Dapr 中，业务应用只需要使用 Dapr 提供的标准 API，就可以使用这些分布式能力而无法关注提供这些能力的具体产品（逻辑抽象）：

以 pub-sub 模型中的发消息为例，这是 Dapr 提供的 Java 客户端 SDK API：

public interface DaprClient {
    Mono<Void> publishEvent(String topic, Object event);
   Mono<Void> publishEvent(String topic, Object event, Map<String, String> metadata);
}

可见在发送事件时，Dapr 完全屏蔽了底层消息机制的具体实现，通过客户端 SDK 为应用提供发送消息的高层抽象，在 Dapr Runtime 中对接底层 MQ 实现——完全解耦应用和 MQ：

关于 Multiple Runtime / Mecha 架构的介绍不在这里深入展开，有兴趣的同学可以浏览我之前的博客文章——“Mecha：将 Mesh 进行到底”。

https://skyao.io/talk/202004-mecha-mesh-through-to-the-end/

稍后我会有一篇深度文章针对上面这个话题，详细介绍在消息通讯领域和 EDA 架构下如何实现消息通讯和事件驱动的抽象和标准化，以避免业务应用和底层消息产品绑定和强耦合，敬请关注。

4 引入 Event：解除不必要的强耦合

在解决了微服务 /Serverless 系统和外部分布式能力之间紧耦合的问题后，我们继续看微服务 /Serverless 系统内部紧耦合的问题。前面讨论到，从单体到微服务到 Function/Serverless，如果只是简单的将直接方法调用替换为远程调用（REST 或者 RPC），那么两个通讯的模块之间会因为这个紧密耦合的调用而形成依赖，而且依赖关系会伴随调用链继续传递，导致形成一个树形的依赖关系网络，表现为系统间的高度耦合：

要解决这个问题，基本思路在于审视两个组件之间通讯行为的业务语义，然后据此决定两者之间究竟是应该采用 Command/ 命令模式还是 Event/ 事件模式。

温故而知新：Event 和 Command

首先，我们来温习一下 Event 和 Command 的概念和差别，借用一张图片，总结的非常到位：

什么是 Event？

Event: “A significant change in state” — K. Mani Chandy

Event 代表领域中已经发生的事情：通常意味着有行为（Action）已经发生，有状态（Status）已经改变。

因为是已经发生的事情，因此：

Event 可以被理解为是对已经发生的事实的客观陈述；
这意味着 Event 通常是不可变的：Event 的信息（代表着客观事实）不能被篡改，Event 的产生不能逆转；
命名：Event 通常以动词的完成时态命名，如 UserRegistredEvent。

产生 Event 的目标是为了接下来的 Event 传播：

将已经发生的 Event 通知给对此感兴趣的观察者；
收到 Event 的观察者将根据 Event 的内容进行判断和决策：可能会有接下来的动作（Action），有些动作可能需要和其他模块通讯而触发命令（Command），这些动作执行完毕可能会造成领域状态的改变从而继续触发新的事件（Event）。

Event 传播的方式：

Event 有明确的“源 /source”，即 Event 产生（或者说状态改变）的发生地；
但由于生产者并不知道（不愿意 / 不关心）会有哪些观察者对 Event 感兴趣，因此 Event 中并不包含“目的地 /Destination”信息；
Event 通常是通过 MessageQueue 机制，以 pub-sub 的方式传播；
Event 通常不需要回复（ reply）或者应答（response）；
Event 通常用发布（publish）。

什么是 Command？

Command 用于传递一个要求执行某个动作（Action）的请求。

Command 代表将要发生的事情：

通常意味着行为（Action）还未发生但即将发生（如果请求被接受和执行）；
存在被拒绝的可能：不愿意执行（参数校验失败，权限不足），不能执行（接收者故障或者资源无法访问）；
命名：Command 通常以动词的普通形态命名，如 UserRegisterCommand。

产生 Command 的目标是为了接下来的 Command 执行：

将 Command 发送给期望的执行者；
收到 Command 的执行者将根据 Command 的要求进行执行：在执行的过程中内部可能有多个动作（Action），有些动作可能需要和其他模块通讯而触发命令（Command），这些动作执行完毕可能会造成领域状态的改变从而继续触发新的事件（Event）。

Command 的传播方式：

Command 有明确的源（Source），即 Command 的发起者；
Command 也有非常明确的执行者（而且通常是一个），因此命名通常包含“目的地 /Destination”信息；
Command 通常是通过 HTTP / RPC 这样的点对点远程通讯机制，通常是同步；
Command 通常需要应答（Response）：回应 Command 是否被执行（因为可能被拒绝），执行结果（因为可能执行失败）；
Command 通常用发送（Send）。

Command 和 Event 总结

总结 —— Command 和 Event 的本质区别在于他们的意图：

Command 的意图是告知希望发生的事情；
Event 的意图是告知已经发生的事情。

意图上的差异最终会在服务间依赖关系上有特别的体现：

Command 的发起者必须明确知晓 Command 的接收者并明确指示需要做什么（所谓的命令、指示、操纵、编排），尤其当发起者连续发出多个 Command 时，通常这些 Command 会有非常明确的顺序和逻辑关系，以组合为特定的业务逻辑。

Command 的依赖关系简单明确: 发起者 “显式依赖” 接收者

Event 的发起者只需负责发布 Event，而无需关注 Event 的接收者，包括接收者是谁（一个还是多个）以及会做什么（所谓的通知、驱动、协调）。即使 Event 实际有多个接收者，这些接受者之间往往没有明确的顺序关系，其处理过程中的业务逻辑也往往是彼此独立的。

Event 的依赖关系稍微复杂一些：发起者明确不依赖接收者，接收者则存在对发起者 “隐式的反向依赖” ——反向是指和 Command 的依赖关系相比方向调转，是接受者反过来依赖发起者；隐式则是指这种依赖只体现于 “接受者依赖 Event，而 Event 是由发起者发布” 的间接关系中，接受者和发起者之间并不存在直接依赖关系。

从业务视角出发：关系模型决定通讯行为

在温习完 Command 和 Event 之后，我们再来看我们前面的问题：为什么简单的将直接方法调用替换为远程调用（REST 或者 RPC）会出问题？主要原因是在这个替换过程中，所谓简单是指不假思索直接选择远程调用，也就是选择全程 Command 方式：

真实业务场景下各个组件（微服务或者 Function）的业务逻辑关系，通常不会像上图这么夸张，不应该全是 Command （后面会谈到也不应该全是 Event），而应该是类似下图描述的两者结合，以微服务为例（Function 类推）：

业务输入：图上微服务 A 接收到业务请求的输入（可能是 Command 方式，也可能是 Event 方式）

业务逻辑 “实现” 的执行过程：

微服务 A 在执行 Command（或者被 Event 触发）的过程中，会有很多动作（Action）；
有些是微服务 A 内部的动作，比如操作数据库，操作 key-value 存储，内存中的业务逻辑处理等；
有些是和外部微服务进行通讯，如执行查询或要求对方进行某些操作，这些通讯方式是以 Command 的形式，如图上和微服务 B 的通讯；
在这些内部和外部动作完成之后，执行过程完成；
如果是 Command，则需要以应答的形式给回 Command 操作的结果。

业务状态变更触发的后续行为：

在上面的执行过程完成后，如果涉及到业务状态的变更，则需要为此发布事件；
事件通过 event bus 分发给对该事件感兴趣的其他微服务：注意这个过程是解耦的，微服务 A 不清楚也不关心哪些微服务对此事件感兴趣，事件也不需要应答。

上面微服务 A 的业务逻辑执行处理过程中，需要以 Command 或者 Event 方式和其他微服务通讯，如图中的微服务 B/C/D/E。而对于这些微服务 B/C/D/E（视为微服务 A 的下游服务），他们在接受到业务请求后的处理流程和微服务 A 的处理流程是类似的。

因此我们可以简单推导一下，当业务处理逻辑从微服务 A 延展到微服务 A 的下游服务（图中的微服务 B/C/D/E）时的场景：

将图中涉及的微服务 A/B/C/D/E 在处理业务逻辑的行为总结下来，通讯行为大体是一样的：

抽象起来，一个典型的微服务在业务处理流程中的通讯行为可以概括为以下四点：

输入：以一个 Command 请求或者一个 Event 通知为输入，这是业务处理流程的起点。
内部 Action：微服务的内部逻辑，典型如数据库操作，访问 Redis 等 key-value 存储（对应于 Multiple Runtime/Mecha 架构中的各种分布式能力）。可选，通常为 0-N 个。
外部访问：以 Command 形式访问外部的其他微服务。可选，通常为 0-N 个。
通告变更：以 Event 形式对外发布事件，通告上述操作产生的业务状态的变更。可选，通常为 0-1 个。

在这个行为模式中，2 和 3 是没有顺序的，而且可能交错执行，而 4 通常都是在流程的最后：只有当各种内部 Action 和外部 Command 都完成，业务逻辑实现结束，状态变更完成，“木已成舟”，才能以 Event 的方式对外发布：“操作已完成，状态已变更，望周知”。

这里我们回顾一下前面的总结 —— Event 和 Command 的本质区别在于他们的意图：

Event 的意图是告知已经发生的事情；
Command 的意图是告知希望发生的事情。

从业务逻辑处理的角度来看，外部访问的 Command 和内部操作的 Action 是业务逻辑的 “实现” 部分：这些操作组成了完整的业务逻辑——如果这些操作失败，则业务处理将会直接影响（失败或者部分失败）。而发布事件则是业务逻辑完成之后的后续 “通知” 部分：当业务逻辑处理完毕，状态变更完成后，以事件的方式驱动后续的进一步处理。注意是驱动，而不是直接操纵。

从时间线的角度来看整个业务处理流程如下图所示：

全程 Command 带来的问题：不必要的强耦合

全程 Command 的微服务系统，存在的问题就是在上述最后阶段的“状态变更通知”环节，没有采用 Event 和 pub-sub 模型，而是继续使用 Command 逐个调用下游相关的其他微服务：

Event 可以解耦生产者和消费者，因此图中的微服务 A 和微服务 C/D/E 之间没有强烈的依赖关系，彼此无需锁定对方的存在。但是 Command 不同，在采用 Command 方式后微服务 A 和下游相关微服务 C/D/E 会形成强依赖，而且这种依赖关系会蔓延，最终导致形成一颗巨大而深层次的依赖树，而 Function 由于粒度更细，问题往往更严重：

而如果在“状态变更通知”环节引入 Event，则可以解耦微服务和下游被通知的微服务，从而将依赖关系解除，避免无限制的蔓延。如下图所示，左边图形是使用 Event 代替 Command 来进行状态变更通知之后的依赖关系，考虑到 Event 对生产者和消费者的解耦作用，我们“斩断”绿色的 Event 箭头，这样就得到了右边这样一个被分解为多个小范围依赖树的系统依赖关系图：

对 Event 和 Command 使用的建议：

在单体应用拆分为微服务时，不应该简单的将原有的方法调用替换为 Command；
应该审视每个调用在业务逻辑上的语义：是业务逻辑执行的组成部分？还是执行完成之后的状态通知？
然后据此决定采用 Command 还是 Event。

编排和协调

在 Command 和 Event 的使用上，还有两个概念：编排和协调。

这里强烈推荐一篇博客文章， Microservices Choreography vs Orchestration: The Benefits of Choreography，作者 Jonathan Schabowsky ，Solace 的 CTO。他在这边博客中总结了让微服务协同工作的两种模式，并做了一个生动的比喻：

https://solace.com/blog/microservices-choreography-vs-orchestration/

编排（Orchestration）：需要主动控制所有的元素和交互，就像指挥家指挥乐团的乐手一样——对应 Command。

协调（Choreography）：需要建立一个模式，微服务会跟随音乐起舞，不需要监督和指令——对应 Event。

也曾看到很多持类似观点的文章，其中有一张图片印象深刻，我摘录过来：

左边是期望通过编排（Orchestration）方式得到的整齐划一的理想目标，右边是实际得到的大型翻车现场。

全程 Event 带来的问题：开发困难和业务边界不清晰

在 Command 和 Event 的使用上，除了全程使用 Command 之外，还有一个极端是全程使用 Event，这一点在 Lambda（FaaS）中更常见一些：

这个方式首当其冲的问题就是在适用 Command 语义的地方采用了 Event 来替代，而由于 Command 和 Event 在使用语义上的差异，这个替代会显得别扭：

Command 是一对一的，替代它的 Event 也不得不从 “1:N” 退化为 “1:1”，pub-sub 模型不再存在。
Command 是需要返回结果的，尤其是 Query 类的 Command 必须要有查询结果，使用 Event 替代之后，就不得不实现 “支持 Response 的 Event”，典型如在消息机制中实现 Request-Reply 模型的。
或者引入另外一个 Event 来反向通知结果，即用两个异步 Event 来替代一个同步的 Command —— 这需要让发起者进行额外的订阅和处理，开发复杂性远远超过使用简单的 Command。
而且还引入了一个非常麻烦的状态问题：即服务间通讯的上下文中通常是有状态的，Reply Event 必须准确的发送给 Request Event 的发起者的实例，而不能任意选择一个。这使得 Reply Event 不仅仅要 1:1 的绑定订阅者服务，还必须绑定这个服务的特定实例 —— 这样的 Reply Event 已经没法称为 Event 了。
绕开这个状态问题的常见方案是选择无状态的场景，如果处理 Reply Event 时无需考虑状态，那么 Event Reply 才能简单的发送给任意的实例。

对于粒度较大的微服务系统，通常很难实现无状态，所以在微服务中全程采用 Event 通常会比较别扭的，事实上也很少有人这样做。而在粒度非常小的 Function/FaaS 系统中，全程采用 Event 方式比较常见。

关于全程使用 Event，我个人持保留态度，我倾向于即使是在 FaaS 中，也适当保留 Command 的用法：如果某个操作是“业务逻辑”执行中不可或缺的一部分，那么 Command 方式的紧耦合反而更能体现出这个“业务逻辑”的存在：

如果完全采用 Event 方式，“彻底”解耦，则产生新的问题（且不论在编码方面额外带来的复杂度） —— 在海量细粒度的 Event 调用下，业务逻辑已经很难体现，领域模型（Domain Modeling）和有界上下文（Bounded Context）则淹没在这些 Event 调用下，难于识别：

备注：这个问题被称为“Lambda Pinball”，这里不深入展开，后续计划会有一篇文章单独详细探讨“Lambda Pinball”的由来和解决的思路。

Command 和 Event 的选择：实事求是不偏不倚

总结一下 Command 和 Event 的选择，我个人的建议是不要一刀切：全程 Command 方式的缺点容易理解，但简单替换为全程 Event 也未必合适。

我的个人观点是倾向于从实际“业务逻辑”处理的语义出发，判断：

如果是业务逻辑的 “实现” 部分：倾向于选择使用 Command；
如果是业务逻辑完成之后的后续 “通知” 部分：强烈建议选择使用 Event。

5、总结与反思

警惕：不要沦为分布式单体

上面我们列出了微服务和 Serverless 实践中容易形成 “分布式单体” 的两个主要原因和对策：

通过共享库和网络客户端访问分布式能力：引入非侵入方案解耦应用和各种分布式能力；
简单用远程调用替代进程内方法调用：区分 Command 和 Event，引入 Event 来解除微服务间不必要的强耦合。

前者在技术上目前还不太成熟，典型如 Istio/Dapr 项目都还有待加强，暂时在落地上阻力比较大。但后者已经是业界多年的成熟实践，甚至在微服务和 Serverless 兴起之前就广泛使用，因此建议可以立即着手改进。

关于如何更方便的将 Event 和 Event Driven Architecture 引入到微服务和 Serverless 中，同时又不与提供 Message Queue 分布式能力的具体实现耦合，我将在稍后文章中详细展开，敬请期待。

反思：喧闹和谩骂之外的冷静思考

如果我们在微服务和 Serverless 实践中，始终停留在“用远程调用简单替代进程内方法调用”的程度，并固守单体时代的习惯引入各种 SDK，那么分布式单体问题就必然不可避免。我们的微服务转型、Serverless 实践最后得到的往往是：

把单体变成…更糟糕的分布式单体

当然，微服务可能成为分布式单体，但这并不意味着微服务架构是个谎言，也不意味着比单体架构更差。Serverless 可能同样遭遇分布式单体（还有后续要深入探讨的 Lambda Pinball），但这也不意味着 Serverless 不可取 —— 微服务和 Serverless 都是解决特定问题的工具，和所有的工具一样，在使用工具之前，我们需要先研究和了解它们，学习如何正确的使用它们：

需要为微服务创建正确的架构，和单体架构必然会有很大的不同：一定不是“原封不动”的将方法调替换为远程调用，最好不要用共享类库和网络客户端的方式直接使用各种分布式能力；
Serverless 更是需要我们对架构进行彻底的反思，需要改变思维方式，才能保证收益大于弊端。

参考资料和推荐阅读：

《Avoid the Distributed Monolith!!》：Verizon 的 Mohamad Byan 在 2018 年 9 月的一个演讲，描述微服务实践中的分布式单体陷阱和解决的方式。——https://www.slideshare.net/DevOpsDaysDFW/avoid-the-distributed-monolith
《Mecha：将 Mesh 进行到底》：详细介绍 Multiple Runtime / Macha 架构，将更多的分布式能力进行 Mesh 化——https://skyao.io/talk/202004-mecha-mesh-through-to-the-end/
《The Eight Fallacies of Distributed Computing》：分布式计算领域的经典文章，中文翻译如下——http://www.xumenger.com/the-eight-fallacies-of-distributed-computing-20180817/
《Opportunities and Pitfalls of Event-driven Utopia》：Bernd Rücker 在 QCon 上的一个演讲，讲述“事件驱动乌托邦的机遇与陷阱”，本文部分图片来自这份 PPT——https://www.youtube.com/watch?v=jjYAZ0DPLNM
《Practical DDD: Bounded Contexts + Events => Microservices》：Indu Alagarsamy 的一个演讲，介绍领域驱动开发（DDD）和 Messaging 的交集。推荐使用消息技术在干净、定义良好的有界上下文之间进行通信，以去除时空耦合。——https://www.infoq.com/presentations/microservices-ddd-bounded-contexts/
《Building Event-Driven Cloud Applications and Services》：讨论构建事件驱动的应用和服务的通用实践和技术，是一个序列教程——https://medium.com/@ratrosy/building-event-driven-cloud-applications-and-services-ad0b5b970036。
《The Architect’s Guide to Event-Driven Microservices》：Solace 公司网站上的一份 PDF 格式的小册子——https://go.solace.com/wp-download-eventdrivenmicroservices.html
《致传统企业朋友：不够痛就别微服务，有坑》：网易云刘超刘老师的超级好文章，极其实在而全面的讲述微服务落地需要考虑的方方面面以及各种问题，强烈推荐阅读。——https://www.infoq.cn/article/Nd0RofAUp0WtlvlQArbu

深入分析Flink SQL的工作机制

发表于 2020-06-11 | 分类于 flink

摘要：本文整理自 Flink Forward 2020 全球在线会议中文精华版，由 Apache Flink PMC 伍翀（云邪）分享，社区志愿者陈婧敏（清樾）整理。旨在帮助大家更好地理解 Flink SQL 引擎的工作原理。文章主要分为以下四部分：

Flink SQL Architecture

How Flink SQL Works?

Flink SQL Optimizations

Summary and Futures

Tips：点击下方链接可查看作者分享的原版视频～
https://ververica.cn/developers/flink-forward-virtual-conference/

Apache Flink 社区在最近的两个版本（1.9 & 1.10 ）中为面向未来的统一流批处理在架构层面做了很多优化，其中一个重大改造是引入了 Blink Planner，开始支持 SQL & Table API 使用不同的 SQL Planner 进行编译（Planner 的插件化）。

本文首先会介绍推动这些优化背后的思考，展示统一的架构如何更好地处理流式和批式查询，其次将深入剖析 Flink SQL 的编译及优化过程，包括：

Flink SQL 利用 Apache Calcite 将 SQL 翻译为关系代数表达式，使用表达式折叠（Expression Reduce），下推优化（Predicate / Projection Pushdown ）等优化技术生成物理执行计划（Physical Plan），利用 Codegen 技术生成高效执行代码。
Flink SQL 使用高效的二进制数据存储结构 BinaryRow 加速计算性能；使用 Mini-batch 攒批提高吞吐，降低两层聚合时由 Retraction 引起的数据抖动；聚合场景下数据倾斜处理和 Top-N 排序的优化原理。

Flink SQL 架构 & Blink Planner（1.9+ ）

1.1 Old Planner 的限制

要想了解 Flink SQL 在1.9 版本引入新架构的动机，我们首先看下 1.9 版本之前的架构设计

从图中可以看出，虽然面向用户的 Table API & SQL 是统一的，但是流式和批式任务在翻译层分别对应了 DataStreamAPI 和 DataSetAPI，在 Runtime 层面也要根据不同的 API 获取执行计划，两层的设计使得整个架构能够复用的模块有限，不易扩展。

1.2 统一的 Blink Planner

Flink 在设计之初就遵循“批是流的特例”的理念，在架构上做到流批统一是大势所趋。在社区和阿里巴巴的共同努力下，1.9 版本引入了新的 Blink Planner，将批 SQL 处理作为流 SQL 处理的特例，尽量对通用的处理和优化逻辑进行抽象和复用，通过 Flink 内部的 Stream Transformation API 实现流 & 批的统一处理，替代原 Flink Planner 将流 & 批区分处理的方式。

此外，新架构通过灵活的插件化方式兼容老版本 Planner，用户可自行选择。不过在 1.11 版本 Blink Planner 会代替 Old Planner 成为默认的 Planner 来支持流 & 批进一步融合统一（ Old Planner 将在之后逐步退出历史舞台）。

Flink SQL 工作流

Flink SQL 引擎的工作流总结如图所示。

从图中可以看出，一段查询 SQL / 使用TableAPI 编写的程序（以下简称 TableAPI 代码）从输入到编译为可执行的 JobGraph 主要经历如下几个阶段

将 SQL文本 / TableAPI 代码转化为逻辑执行计划（Logical Plan）
Logical Plan 通过优化器优化为物理执行计划（Physical Plan）
通过代码生成技术生成 Transformations 后进一步编译为可执行的 JobGraph 提交运行

本节将重点对 Flink SQL 优化器的常用优化方法和 CodeGen 生成 Transformations 进行介绍。

2.1 Logical Planning

Flink SQL 引擎使用 Apache Calcite SQL Parser 将 SQL 文本解析为词法树，SQL Validator 获取 Catalog 中元数据的信息进行语法分析和验证，转化为关系代数表达式（RelNode），再由 Optimizer 将关系代数表达式转换为初始状态的逻辑执行计划。

备注：TableAPI 代码使用 TableAPI Validator 对接 Catalog 后生成逻辑执行计划。

E.g.1 考虑如下表达 JOIN 操作的一段 SQL。

SELECT 
  t1.id, 1 + 2 + t1.value AS v 
FROM t1, t2 
WHERE 
  t1.id = t2.id AND 
  t2.id < 1000

经过上述操作后得到了一个树状结构的逻辑执行计划，根节点对应最上层的 Select 语句，叶子节点对应输入表 t1 和 t2 的 TableScan 操作，Join 和 Where 条件过滤分别对应了 Join 和 Filter 节点。

LogicalProject(id=[$0], v=[+(+(1, 2), $1)])
+- LogicalFilter(condition=[AND(=($0, $3), <($3, 1000))])
   +- LogicalJoin(condition=[true], joinType=[inner])
      :- LogicalTableScan(table=[[default_catalog, default, t1]])
      +- LogicalTableScan(table=[[default_catalog, default, t2]])

可视化后如图所示，这是优化器开始工作的初始状态。

下面开始介绍 Flink SQL 优化器常见的几种优化方式。

■ 2.1.1 Expression Reduce

表达式（Expression）是 SQL 中最常见的语法。比如 t1.id 是一个表达式， 1 + 2 + t1.value 也是一个表达式。优化器在优化过程中会递归遍历树上节点，尽可能预计算出每个表达式的值，这个过程就称为表达式折叠。这种转换在逻辑上等价，通过优化后，真正执行时不再需要为每一条记录都计算一遍 1 + 2。

■ 2.1.2 PushDown Optimization

下推优化是指在保持关系代数语义不变的前提下将 SQL 语句中的变换操作尽可能下推到靠近数据源的位置以获得更优的性能，常见的下推优化有谓词下推（Predicate Pushdown），投影下推（Projection Pushdown，有时也译作列裁剪）等。

Predicate Pushdown

回顾 E.g.1，我们发现 WHERE 条件表达式中 t2.id < 1000 这个过滤条件描述的是对表 t2 的约束，跟表 t1 无关，完全可以下推到 JOIN 操作之前完成。假设表 t2 中有一百万行数据，但是满足 id < 1000 的数据只有 1,000 条，则通过谓词下推优化后到达 JOIN 节点的数据量降低了1,000 倍，极大地节省了 I / O 开销，提升了 JOIN 性能。

谓词下推（Predicate Pushdown）是优化 SQL 查询的一项基本技术，谓词一词来源于数学，指能推导出一个布尔返回值（TRUE / FALSE）的函数或表达式，通过判断布尔值可以进行数据过滤。谓词下推是指保持关系代数语义不变的前提下将 Filter 尽可能移至靠近数据源的位置（比如读取数据的 SCAN 阶段）来降低查询和传递的数据量（记录数）。

Projection Pushdown

列裁剪是 Projection Pushdown 更直观的描述方式，指在优化过程中去掉没有使用的列来降低 I / O 开销，提升性能。但与谓词下推只移动节点位置不同，投影下推可能会增加节点个数。比如最后计算出的投影组合应该放在 TableScan 操作之上，而 TableScan 节点之上没有 Projection 节点，优化器就会显式地新增 Projection 节点来完成优化。另外如果输入表是基于列式存储的（如 Parquet 或 ORC 等），优化还会继续下推到 Scan 操作中进行。

回顾 E.g.1，我们发现整个查询中只用到了表 t1 的 id 和 value 字段，表 t2 的 id 字段，在 TableScan 节点之上分别增加 Projection 节点去掉多余字段，极大地节省了 I / O 开销

简要总结一下，谓词下推和投影下推分别通过避免处理不必要的记录数和字段数来降低 I / O 开销提升性能。

2.2 Physical Planning on Batch

通过上述一系列操作后，我们得到了优化后的逻辑执行计划。逻辑执行计划描述了执行步骤和每一步需要完成的操作，但没有描述操作的具体实现方式。而物理执行计划会考虑物理实现的特性，生成每一个操作的具体实现方式。比如 Join 是使用 SortMergeJoin、HashJoin 或 BroadcastHashJoin 等。优化器在生成逻辑执行计划时会计算整棵树上每一个节点的 Cost，对于有多种实现方式的节点（比如 Join 节点），优化器会展开所有可能的 Join 方式分别计算。最终整条路径上 Cost 最小的实现方式就被选中成为 Final Physical Plan。

回顾 E.g.1，当它以批模式执行，同时我们可以拿到输入表的 Statistics 信息。在经过前述优化后，表 t2 到达 Join 节点时只有 1,000 条数据，使用 BroadcastJoin 的开销相对最低，则最终的 Physical Plan 如下图所示。

2.3 Translation & Code Generation

代码生成（Code Generation）在计算机领域是一种广泛使用的技术。在 Physical Plan 到生成 Transformation Tree 过程中就使用了 Code Generation。

回顾 E.g.1，以表 t2 之上的 Calc 节点 t2.id < 1000 表达式为例，通过 Code Generation 后生成了描述 Transformation Operator 的一段 Java 代码，将接收到的 Row 中 id < 1000 的 Row 发送到下一个 Operator。

Flink SQL 引擎会将 Physical Plan 通过 Code Generation 翻译为 Transformations，再进一步编译为可执行的 JobGraph。

2.4 Physical Planning on Stream

以上介绍了 Flink SQL 引擎的整体工作流，上述例子是假定以批模式编译的，下面我们来介绍一下以流模式编译时，在生成 Physical Plan 过程中的一个重要机制：Retraction Mechanism （aka. Changelog Mechanism）。

■ 2.4.1 Retraction Mechanism

Retraction 是流式数据处理中撤回过早下发（Early Firing）数据的一种机制，类似于传统数据库的 Update 操作。级联的聚合等复杂 SQL 中如果没有 Retraction 机制，就会导致最终的计算结果与批处理不同，这也是目前业界很多流计算引擎的缺陷。

E.g.2 考虑如下统计词频分布的 SQL。

SELECT cnt, COUNT(cnt) as freq
FROM (
  SELECT word, COUNT(*) as cnt
  FROM words
  GROUP BY word)
GROUP BY cnt

假设输入数据是：

word
Hello
World
Hello

则经过上面的计算后，预期的输出结果应该是：

cnt	freq
1	1
2	1

但与批处理不同，流处理的数据是一条条到达的，理论上每一条数据都会触发一次计算，所以在处理了第一个 Hello 和第一个 World 之后，词频为 1 的单词数已经变成了 2，此时再处理第二个 Hello 时，如果不能修正之前的结果，Hello 就会在词频等于 1 和词频等于 2 这两个窗口下被同时统计，显然这个结果是错误的，这就是没有 Retraction 机制带来的问题。

Flink SQL 在流计算领域中的一个重大贡献就是首次提出了这个机制的具体实现方案。Retraction 机制又名 Changelog 机制，因为某种程度上 Flink 将输入的流数据看作是数据库的 Changelog，每条输入数据都可以看作是对数据库的一次变更操作，比如 Insert，Delete 或者 Update。以 MySQL 数据库为例，其Binlog 信息以二进制形式存储，其中 Update_rows_log_event 会对应 2 条标记 Before Image （BI）和 After Image （AI），分别表示某一行在更新前后的信息。

在 Flink SQL 优化器生成流作业的 Physical Plan 时会判断当前节点是否是更新操作，如果是则会同时发出 2 条消息 update_before 和 update_after 到下游节点，update_before 表示之前“错误”下发的数据，需要被撤回，update_after 表示当前下发的“正确”数据。下游收到后，会在结果上先减去 update_before，再加上 update_after。

回顾 E.g.2，下面的动图演示了加入 Retraction 机制后正确结果的计算过程。

update_before 是一条非常关键的信息，相当于标记出了导致当前结果不正确的那个“元凶”。不过额外操作会带来额外的开销，有些情况下不需要发送 update_before 也可以获得正确的结果，比如下游节点接的是 UpsertSink（MySQL 或者 HBase的情况下，数据库可以按主键用 update_after 消息覆盖结果）。是否发送 update_before 由优化器决定，用户不需要关心。

■ 2.4.2 Update_before Decision

前面介绍了 Retraction 机制和 update_before，那优化器是怎样决定是否需要发送update_before 呢？本节将介绍这一部分的工作。

Step1：确定每个节点对应的 Changelog 变更类型

数据库中最常见的三种操作类型分别是 Insert （记为 [I]），Delete（记为 [D]），Update（记为 [U]）。优化器首先会自底向上检查每个节点，判断它属于哪（几）种类型，分别打上对应标记。

回顾 E.g.2，第一个 Source 节点由于只产生新数据，所以属于 Insert，记为 [I]；第二个节点计算内层的聚合，所以会发出更新的消息，记为 [I，U]；第三个节点裁掉 word 字段，属于简单计算，传递了上游的变更类型，记为 [I，U]；第四个节点是外层的聚合计算，由于它收到了来自上游的 Update 消息，所以额外需要 Delete 操作来保证更新成功，记为 [I，U，D]。

Step2：确定每个节点发送的消息类型

在介绍 Step2 之前，我们先介绍下 Flink 中 Update 消息类型的表示形式。在 Flink 中 Update 由两条 update_before（简称 UB）和 update_after （简称 UA）来表示，其中 UB 消息在某些情况下可以不发送，从而提高性能。

在 Step1 中优化器自底向上推导出了每个节点对应的 Changelog 变更操作，这一步里会先自顶向下推断当前节点需要父节点提供的消息类型，直到遇到第一个不需要父节点提供任何消息类型的节点，再往上回推每个节点最终的实现方式和需要的消息类型。

回顾 E.g.2，由于最上层节点是 UpsertSink 节点，只需要它的父节点提供 [UA] 即可。到了外层聚合的 Aggregate 节点，由于 Aggregate 节点的输入有 Update 操作，所以需要父节点需要提供 [UB，UA]，这样才能正确更新自己的计算状态。

再往下到 Calc 节点，它需要传递 [UB，UA] 的需求给它的父节点，也就是内层的 Aggregate 节点。而到了内层 Aggregation 节点，它的父节点是 Source 节点，不会产生 Update 操作，所以它不需要 Source 节点额外发送任何 [UB / UA ]。当优化器遍历到 Source 节点，便开始进行回溯，如果当前节点能满足子节点的 requirement，则将对应的标签更新到节点上，否则便无法生成 plan。首先内层的 Aggregate 能产生 UB，所以能满足子节点的 requirement，所以优化器会给内层的 Aggregate 节点打上 [UB，UA] 的标签，然后向上传递到 Calc 节点，同样打上 [UB，UA] ，再到外层的 Aggregate 节点，由于它的下游只需要接受更新后的消息，所以打上 [UA] 标签，表示它只需要向下游发送 update_after 即可。

这些标签最终会影响算子的物理实现，比如外层的 Aggregate 节点，由于它会接收到来自上游的 [UB]，所以物理实现会使用带 Retract 的 Count，同时它只会向 Sink 发送 update_after。而内层的 Aggregate 节点，由于上游发送过来的数据没有 [UB]，所以可以采用不带 Retract 的 Count 实现，同时由于带有 [UB] 标签，所以需要往下游发送 update_before。

Flink SQL Internal Optimization

前面介绍了 Flink SQL 引擎的工作原理，接下来会简要概括一下 Flink SQL 内部的一些优化，更多资料可以在 Flink Forward Asia 2019 查看。

3.1 BinaryRow

在 Flink 1.9+ 前， Flink Runtime 层各算子间传递的数据结构是 Row，其内部实现是 Object[]。这种数据结构的问题在于不但需要额外开销存 Object Metadata，计算过程中还涉及到大量序列化 / 反序列（特别是只需要处理某几个字段时需要反序列化整个 Row），primitive 类型的拆 / 装箱等，都会带来大量额外的性能开销。

Flink 1.9 开始引入了 Blink Planner，使用二进制数据结构的 BinaryRow 来表示 Record。BinaryRow 作用于默认大小为 32K 的 Memory Segment，直接映射到内存。BinaryRow 内部分为 Header，定长区和变长区。Header 用于存储 Retraction 消息的标识，定长区使用 8 个 bytes 来记录字段的 Nullable 信息及所有 primitive 和可以在 8 个 bytes 内表示的类型。其它类型会按照基于起始位置的 offset 存放在变长区。

BinaryRow 作为 Blink Planner 的基础数据结构，带来的好处是显而易见的：首先存储上更为紧凑，去掉了额外开销；其次在序列化和反序列化上带来的显著性能提升，可根据 offset 只反序列化需要的字段，在开启 Object Reuse 后，序列化可以直接通过内存拷贝完成。

3.2 Mini-batch Processing

Flink 是纯流式处理框架，在理论上每一条新到的数据都会触发一次计算。然而在实现层面，这样做会导致聚合场景下每处理一条数据都需要读写 State 及序列化 / 反序列化。如果能够在内存中 buffer 一定量的数据，预先做一次聚合后再更新 State，则不但会降低操作 State 的开销，还会有效减少发送到下游的数据量，提升 throughput，降低两层聚合时由 Retraction 引起的数据抖动, 这就是 Mini-batch 攒批优化的核心思想。

3.3 Skew Processing

对于数据倾斜的优化，主要分为是否带 DISTINCT 去重语义的两种方式。对于普通聚合的数据倾斜，Flink 引入了 Local-Global 两阶段优化，类似于 MapReduce 增加 Local Combiner 的处理模式。而对于带有去重的聚合，Flink 则会将用户的 SQL 按原有聚合的 key 组合再加上 DISTINCT key 做 Hash 取模后改写为两层聚合来进行打散。

3.4 Top-N Rewrite

全局排序在流式的场景是很难实现的，但如果只需要计算到目前的 Top-N 极值，问题就变得可解。不过传统数据库求排序的 SQL 语法是通过 ORDER BY 加 LIMIT 限制条数，背后实现的机制也是通过扫描全表排序后再返回 LIMIT 条数的记录。另外如果按照某些字段开窗排序，ORDER BY 也无法满足要求。Flink SQL 借鉴了批场景下开窗求 Top-N 的语法，使用 ROW_NUMBER 语法来做流场景下的 Top-N 排序。

E.g.3 下面这段 SQL 计算了每个类目下销量 Top3 的店铺

SELECT*
FROM(
  SELECT *, -- you can get like shopId or other information from this
    ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS rowNum
  FROM shop_sales ) 
WHERE rowNum <= 3

在生成 Plan 方面，ROW_NUMBER 语义对应 OverAggregate 窗口节点和一个过滤行数的 Calc 节点，而这个窗口节点在实现层面需要为每一个到达的数据重新将 State 中的历史数据拿出来排序，这显然不是最优解。

我们知道流式场景求解极大 / 小值的最优操作是通过维护一个 size 为 N 的 minHeap / maxHeap。由实现反推出我们需要在优化器上新增一条规则，在遇到 ROW_NUMBER 生成的逻辑节点后，将其优化为一个特殊的 Rank 节点，对应上述的最优实现方式（当然这只是特殊 Rank 对应的其中一种实现）。这便是 Top-N Rewrite 的核心思想。

Summary & Futures

本文内容回顾

简要介绍 Flink 1.9 + 在 SQL & TableAPI 上引入新架构，统一技术栈，朝着流 & 批一体的方向迈进了一大步。
深入介绍 Flink SQL 引擎的内部运行机制，以及在对用户透明的同时，Flink SQL 在优化方面做的许多工作。

未来工作计划

在 Flink 1.11+ 后的版本，Blink Planner 将作为默认的 Planner 提供生产级别的支持。
FLIP-95：重构 TableSource & TableSink 的接口设计，面向流批一体化，在 Source 端支持 changelog 消息流，从而支持 FLIP-105 的 CDC 数据源。
FLIP-105：Flink TableAPI & SQL 对 CDC 的支持。
FLIP-115：扩展目前只支持 CSV 的 FileSystem Connector，使其成为流批统一的 Generalized FileSystem Connector。
FLIP-123：对 Hive DDL 和 DML 的兼容，支持用户在 Flink 中运行 Hive DDL。

Istio 介绍

发表于 2020-06-05

官方文档：https://istio.io/zh/docs/

概念

Istio 是什么？

云平台令使用它们的公司受益匪浅。但不可否认的是，上云会给 DevOps 团队带来压力。为了可移植性，开发人员必须使用微服务来构建应用，同时运维人员也正在管理着极端庞大的混合云和多云的部署环境。 Istio 允许您连接、保护、控制和观察服务。

从较高的层面来说，Istio 有助于降低这些部署的复杂性，并减轻开发团队的压力。它是一个完全开源的服务网格，作为透明的一层接入到现有的分布式应用程序里。它也是一个平台，拥有可以集成任何日志、遥测和策略系统的 API 接口。Istio 多样化的特性使您能够成功且高效地运行分布式微服务架构，并提供保护、连接和监控微服务的统一方法。

服务网格是什么？

Istio 解决了开发人员和运维人员所面临的从单体应用向分布式微服务架构转变的挑战。了解它是如何做到这一点的可以让我们更详细地理解 Istio 的服务网格。

术语服务网格用来描述组成这些应用程序的微服务网络以及它们之间的交互。随着服务网格的规模和复杂性不断的增长，它将会变得越来越难以理解和管理。它的需求包括服务发现、负载均衡、故障恢复、度量和监控等。服务网格通常还有更复杂的运维需求，比如 A/B 测试、金丝雀发布、速率限制、访问控制和端到端认证。

Istio 提供了对整个服务网格的行为洞察和操作控制的能力，以及一个完整的满足微服务应用各种需求的解决方案。

为什么使用 Istio？

通过负载均衡、服务间的身份验证、监控等方法，Istio 可以轻松地创建一个已经部署了服务的网络，而服务的代码只需很少更改甚至无需更改。通过在整个环境中部署一个特殊的 sidecar 代理为服务添加 Istio 的支持，而代理会拦截微服务之间的所有网络通信，然后使用其控制平面的功能来配置和管理 Istio，这包括：

为 HTTP、gRPC、WebSocket 和 TCP 流量自动负载均衡。
通过丰富的路由规则、重试、故障转移和故障注入对流量行为进行细粒度控制。
可插拔的策略层和配置 API，支持访问控制、速率限制和配额。
集群内（包括集群的入口和出口）所有流量的自动化度量、日志记录和追踪。
在具有强大的基于身份验证和授权的集群中实现安全的服务间通信。

Istio 为可扩展性而设计，可以满足不同的部署需求。

核心特性

Istio 以统一的方式提供了许多跨服务网络的关键功能：

流量管理

Istio 简单的规则配置和流量路由允许您控制服务之间的流量和 API 调用过程。Istio 简化了服务级属性（如熔断器、超时和重试）的配置，并且让它轻而易举的执行重要的任务（如 A/B 测试、金丝雀发布和按流量百分比划分的分阶段发布）。

有了更好的对流量的可视性和开箱即用的故障恢复特性，您就可以在问题产生之前捕获它们，无论面对什么情况都可以使调用更可靠，网络更健壮。

请参考流量管理文档获取更多细节。

安全

Istio 的安全特性解放了开发人员，使其只需要专注于应用程序级别的安全。Istio 提供了底层的安全通信通道，并为大规模的服务通信管理认证、授权和加密。有了 Istio，服务通信在默认情况下就是受保护的，可以让您在跨不同协议和运行时的情况下实施一致的策略——而所有这些都只需要很少甚至不需要修改应用程序。

Istio 是独立于平台的，可以与 Kubernetes（或基础设施）的网络策略一起使用。但它更强大，能够在网络和应用层面保护pod到 pod 或者服务到服务之间的通信。

请参考安全文档获取更多细节。

可观察性

Istio 健壮的追踪、监控和日志特性让您能够深入的了解服务网格部署。通过 Istio 的监控能力，可以真正的了解到服务的性能是如何影响上游和下游的；而它的定制 Dashboard 提供了对所有服务性能的可视化能力，并让您看到它如何影响其他进程。

Istio 的 Mixer 组件负责策略控制和遥测数据收集。它提供了后端抽象和中介，将一部分 Istio 与后端的基础设施实现细节隔离开来，并为运维人员提供了对网格与后端基础实施之间交互的细粒度控制。

所有这些特性都使您能够更有效地设置、监控和加强服务的 SLO。当然，底线是您可以快速有效地检测到并修复出现的问题。

请参考可观察性文档获取更多细节。

平台支持

Istio 独立于平台，被设计为可以在各种环境中运行，包括跨云、内部环境、Kubernetes、Mesos 等等。您可以在 Kubernetes 或是装有 Consul 的 Nomad 环境上部署 Istio。Istio 目前支持：

Kubernetes 上的服务部署
基于 Consul 的服务注册
服务运行在独立的虚拟机上

整合和定制

Istio 的策略实施组件可以扩展和定制，与现有的 ACL、日志、监控、配额、审查等解决方案集成。

安装

要开始使用 Istio，只需遵循以下三个步骤：

搭建平台

在安装 Istio 之前，需要一个运行着 Kubernetes 的兼容版本的 cluster。

Istio 1.6 已经在 Kubernetes 版本 1.15, 1.16, 1.17, 1.18 中测试过。

通过选择合适的 platform-specific setup instructions 来创建一个集群。

有些平台提供了 managed control plane，您可以使用它来代替手动安装 Istio。如果您选择的平台支持这种方式，并且您选择使用它，那么，在创建完集群后，您将完成 Istio 的安装。因此，可以跳过以下说明。

下载 Istio

下载 Istio，下载内容将包含：安装文件、示例和 istioctl 命令行工具。

访问 Istio release 页面下载与您操作系统对应的安装文件。在 macOS 或 Linux 系统中，也可以通过以下命令下载最新版本的 Istio：
1
$ curl -L https://istio.io/downloadIstio | sh -

切换到 Istio 包所在目录下。例如：Istio 包名为 istio-1.6.0，则：
1
$ cd istio-1.6.0

安装目录包含如下内容：

install/kubernetes 目录下，有 Kubernetes 相关的 YAML 安装文件
samples/ 目录下，有示例应用程序
bin/ 目录下，包含 istioctl 的客户端文件。istioctl 工具用于手动注入 Envoy sidecar 代理。

将 istioctl 客户端路径增加到 path 环境变量中，macOS 或 Linux 系统的增加方式如下：
1
$ export PATH=$PWD/bin:$PATH

在使用 bash 或 ZSH 控制台时，可以选择启动 auto-completion option。

安装 Istio

请按照以下步骤在您所选的平台上使用 demo 配置文件安装 Istio。

安装 demo 配置

1	$ istioctl manifest apply --set profile=demo

为了验证是否安装成功，需要先确保以下 Kubernetes 服务正确部署，然后验证除 jaeger-agent 服务外的其他服务，是否均有正确的 CLUSTER-IP：

$ kubectl get svc -n istio-system
NAME                     TYPE           CLUSTER-IP       EXTERNAL-IP     PORT(S)                                                                                                                                      AGE
grafana                  ClusterIP      172.21.211.123   <none>          3000/TCP                                                                                                                                     2m
istio-citadel            ClusterIP      172.21.177.222   <none>          8060/TCP,15014/TCP                                                                                                                           2m
istio-egressgateway      ClusterIP      172.21.113.24    <none>          80/TCP,443/TCP,15443/TCP                                                                                                                     2m
istio-galley             ClusterIP      172.21.132.247   <none>          443/TCP,15014/TCP,9901/TCP                                                                                                                   2m
istio-ingressgateway     LoadBalancer   172.21.144.254   52.116.22.242   15020:31831/TCP,80:31380/TCP,443:31390/TCP,31400:31400/TCP,15029:30318/TCP,15030:32645/TCP,15031:31933/TCP,15032:31188/TCP,15443:30838/TCP   2m
istio-pilot              ClusterIP      172.21.105.205   <none>          15010/TCP,15011/TCP,8080/TCP,15014/TCP                                                                                                       2m
istio-policy             ClusterIP      172.21.14.236    <none>          9091/TCP,15004/TCP,15014/TCP                                                                                                                 2m
istio-sidecar-injector   ClusterIP      172.21.155.47    <none>          443/TCP,15014/TCP                                                                                                                            2m
istio-telemetry          ClusterIP      172.21.196.79    <none>          9091/TCP,15004/TCP,15014/TCP,42422/TCP                                                                                                       2m
jaeger-agent             ClusterIP      None             <none>          5775/UDP,6831/UDP,6832/UDP                                                                                                                   2m
jaeger-collector         ClusterIP      172.21.135.51    <none>          14267/TCP,14268/TCP                                                                                                                          2m
jaeger-query             ClusterIP      172.21.26.187    <none>          16686/TCP                                                                                                                                    2m
kiali                    ClusterIP      172.21.155.201   <none>          20001/TCP                                                                                                                                    2m
prometheus               ClusterIP      172.21.63.159    <none>          9090/TCP                                                                                                                                     2m
tracing                  ClusterIP      172.21.2.245     <none>          80/TCP                                                                                                                                       2m
zipkin                   ClusterIP      172.21.182.245   <none>          9411/TCP                                                                                                                                     2m

如果集群运行在一个不支持外部负载均衡器的环境中（例如：minikube），istio-ingressgateway 的 EXTERNAL-IP 将显示为 `` 状态。请使用服务的 NodePort 或端口转发来访问网关。

请确保关联的 Kubernetes pod 已经部署，并且 STATUS 为 Running：

$ kubectl get pods -n istio-system
NAME                                                           READY   STATUS      RESTARTS   AGE
grafana-f8467cc6-rbjlg                                         1/1     Running     0          1m
istio-citadel-78df5b548f-g5cpw                                 1/1     Running     0          1m
istio-egressgateway-78569df5c4-zwtb5                           1/1     Running     0          1m
istio-galley-74d5f764fc-q7nrk                                  1/1     Running     0          1m
istio-ingressgateway-7ddcfd665c-dmtqz                          1/1     Running     0          1m
istio-pilot-f479bbf5c-qwr28                                    1/1     Running     0          1m
istio-policy-6fccc5c868-xhblv                                  1/1     Running     2          1m
istio-sidecar-injector-78499d85b8-x44m6                        1/1     Running     0          1m
istio-telemetry-78b96c6cb6-ldm9q                               1/1     Running     2          1m
istio-tracing-69b5f778b7-s2zvw                                 1/1     Running     0          1m
kiali-99f7467dc-6rvwp                                          1/1     Running     0          1m
prometheus-67cdb66cbb-9w2hm                                    1/1     Running     0          1m

后续步骤

安装 Istio 后，就可以部署您自己的服务，或部署安装程序中系统的任意一个示例应用。

应用程序必须使用 HTTP/1.1 或 HTTP/2.0 协议用于 HTTP 通信；HTTP/1.0 不支持。

当使用 kubectl apply 来部署应用时，如果 pod 启动在标有 istio-injection=enabled 的命名空间中，那么，Istio sidecar 注入器将自动注入 Envoy 容器到应用的 pod 中：

1 2	$ kubectl label namespace <namespace> istio-injection=enabled $ kubectl create -n <namespace> -f <your-app-spec>.yaml

在没有 istio-injection 标记的命名空间中，在部署前可以使用 istioctl kube-inject 命令将 Envoy 容器手动注入到应用的 pod 中：

1	$ istioctl kube-inject -f <your-app-spec>.yaml \| kubectl apply -f -

如果您不确定要从哪开始，可以先部署 Bookinfo 示例，它会让您体验到 Istio 的流量路由、故障注入、速率限制等功能。然后您可以根据您的兴趣浏览各种各样的 Istio 任务。

下列任务都是初学者开始学习的好入口：

下一步，可以定制 Istio 并部署您自己的应用。在您开始自定义 Istio 来适配您的平台或者其他用途之前，请查看以下资源：

任务

示例

运维

未命名

发表于 2020-06-02

永远不要使用双花括号初始化实例，否则可能会OOM！

发表于 2020-05-23 | 分类于 java

生活中的尴尬无处不在，有时候你只是想简单的装一把，但某些“老同志”总是在不经意之间，给你无情的一脚，踹得你简直无法呼吸。

但谁让咱年轻呢？吃亏要趁早，前路会更好。

喝了这口温热的鸡汤之后，咱们来聊聊是怎么回事。

事情是这样的，在一个不大不小的项目中，小王写下了这段代码：

Map<String, String> map = new HashMap() {{
    put("map1", "value1");
    put("map2", "value2");
    put("map3", "value3");
}};
map.forEach((k, v) -> {
    System.out.println("key:" + k + " value:" + v);
});

本来是用它来替代下面这段代码的：

Map<String, String> map = new HashMap();
map.put("map1", "value1");
map.put("map2", "value2");
map.put("map3", "value3");
map.forEach((k, v) -> {
    System.out.println("key:" + k + " value:" + v);
});

两块代码的执行结果也是完全一样的：

key:map3 value:value3

key:map2 value:value2

key:map1 value:value1

所以小王正在得意的把这段代码介绍给部门新来的妹子小甜甜看，却不巧被正在经过的老张也看到了。

老张本来只是想给昨天的枸杞再续上一杯 85° 的热水，但说来也巧，刚好撞到了一次能在小甜甜面前秀技术的一波机会，于是习惯性的整理了一下自己稀疏的秀发，便开启了 diss 模式。

“小王啊，你这个代码问题很大啊！”

“怎么能用双花括号初始化实例呢？”

此时的小王被问的一脸懵逼，内心有无数个草泥马奔腾而过，心想你这头老牛竟然也和我争这颗嫩草，但内心却有一种不祥的预感，感觉自己要输，瞬间羞涩的不知该说啥，只能红着小脸，轻轻的“嗯？”了一声。

老张：“使用双花括号初始化实例是会导致内存溢出的啦！侬不晓得嘛？”

小王沉默了片刻，只是凭借着以往的经验来看，这“老家伙”还是有点东西的，于是敷衍的“哦~”了一声，仿佛自己明白了怎么回事一样，，其实内心仍然迷茫的一匹，为了不让其他同事发现，只得这般作态。

于是片刻的敷衍，待老张离去之后，才悄悄的打开了 Google，默默的搜索了一下。

小王：哦，原来如此……

双花括号初始化分析

首先，我们来看使用双花括号初始化的本质是什么？

以我们这段代码为例：

Map<String, String> map = new HashMap() {{
    put("map1", "value1");
    put("map2", "value2");
    put("map3", "value3");
}};

这段代码其实是创建了匿名内部类，然后再进行初始化代码块。

这一点我们可以使用命令javac将代码编译成字节码之后发现，我们发现之前的一个类被编译成两个字节码（.class）文件，如下图所示：

我们使用 Idea 打开DoubleBracket$1.class文件发现：

import java.util.HashMap;

class DoubleBracket$1 extends HashMap {
    DoubleBracket$1(DoubleBracket var1) {
        this.this$0 = var1;
        this.put("map1", "value1");
        this.put("map2", "value2");
    }
}

此时我们可以确认，它就是一个匿名内部类。那么问题来了，匿名内部类为什么会导致内存溢出呢？

匿名内部类的“锅”

在 Java 语言中非静态内部类会持有外部类的引用，从而导致 GC 无法回收这部分代码的引用，以至于造成内存溢出。

思考 1：为什么要持有外部类？

这个就要从匿名内部类的设计说起了，在 Java 语言中，非静态匿名内部类的主要作用有两个。

1、当匿名内部类只在外部类（主类）中使用时，匿名内部类可以让外部不知道它的存在，从而减少了代码的维护工作。

2、当匿名内部类持有外部类时，它就可以直接使用外部类中的变量了，这样可以很方便的完成调用，如下代码所示：

public class DoubleBracket {
    private static String userName = "磊哥";
    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
        Map<String, String> map = new HashMap() {{
            put("map1", "value1");
            put("map2", "value2");
            put("map3", "value3");
            put(userName, userName);
        }};
    }
}

从上述代码可以看出在HashMap的方法内部，可以直接使用外部类的变量userName。

思考 2：它是怎么持有外部类的？

关于匿名内部类是如何持久外部对象的，我们可以通过查看匿名内部类的字节码得知，我们使用javap -c DoubleBracket\$1.class命令进行查看，其中$1为以匿名类的字节码，字节码的内容如下；

javap -c DoubleBracket\$1.class
Compiled from "DoubleBracket.java"
class com.example.DoubleBracket$1 extends java.util.HashMap {
  final com.example.DoubleBracket this$0;

  com.example.DoubleBracket$1(com.example.DoubleBracket);
    Code:
       0: aload_0
       1: aload_1
       2: putfield      #1                  // Field this$0:Lcom/example/DoubleBracket;
       5: aload_0
       6: invokespecial #7                  // Method java/util/HashMap."<init>":()V
       9: aload_0
      10: ldc           #13                 // String map1
      12: ldc           #15                 // String value1
      14: invokevirtual #17                 // Method put:(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
      17: pop
      18: aload_0
      19: ldc           #21                 // String map2
      21: ldc           #23                 // String value2
      23: invokevirtual #17                 // Method put:(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
      26: pop
      27: return
}

其中，关键代码的在putfield这一行，此行表示有一个对DoubleBracket的引用被存入到this$0中，也就是说这个匿名内部类持有了外部类的引用。

如果您觉得以上字节码不够直观，没关系，我们用下面的实际的代码来证明一下：

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;

public class DoubleBracket {
    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {
        Map map = new DoubleBracket().createMap();
        // 获取一个类的所有字段
        Field field = map.getClass().getDeclaredField("this$0");
        // 设置允许方法私有的 private 修饰的变量
        field.setAccessible(true);
        System.out.println(field.get(map).getClass());
    }
    public Map createMap() {
        // 双花括号初始化
        Map map = new HashMap() {{
            put("map1", "value1");
            put("map2", "value2");
            put("map3", "value3");
        }};
        return map;
    }
}

当我们开启调试模式时，可以看出map中持有了外部对象DoubleBracket，如下图所示：

以上代码的执行结果为：

class com.example.DoubleBracket

从以上程序输出结果可以看出：匿名内部类持有了外部类的引用，因此我们才可以使用$0正常获取到外部类，并输出相关的类信息。

什么情况会导致内存泄漏？

当我们把以下正常的代码：

public void createMap() {
    Map map = new HashMap() {{
        put("map1", "value1");
        put("map2", "value2");
        put("map3", "value3");
    }};
    // 业务处理....
}

改为下面这个样子时，可能会造成内存泄漏：

public Map createMap() {
    Map map = new HashMap() {{
        put("map1", "value1");
        put("map2", "value2");
        put("map3", "value3");
    }};
    return map;
}

为什么用了「可能」而不是「一定」会造成内存泄漏？

这是因为当此map被赋值为其他类属性时，可能会导致 GC 收集时不清理此对象，这时候才会导致内存泄漏。可以关注我「Java中文社群」后面会专门写一篇关于此问题的文章。

如何保证内存不泄露？

要想保证双花扣号不泄漏，办法也很简单，只需要将map对象声明为static静态类型的就可以了，代码如下：

public static Map createMap() {
    Map map = new HashMap() {{
        put("map1", "value1");
        put("map2", "value2");
        put("map3", "value3");
    }};
    return map;
}

什么？你不相信！

没关系，我们用事实说话，使用以上代码，我们重新编译一份字节码，查看匿名类的内容如下：

javap -c  DoubleBracket\$1.class
Compiled from "DoubleBracket.java"
class com.example.DoubleBracket$1 extends java.util.HashMap {
  com.example.DoubleBracket$1();
    Code:
       0: aload_0
       1: invokespecial #1                  // Method java/util/HashMap."<init>":()V
       4: aload_0
       5: ldc           #7                  // String map1
       7: ldc           #9                  // String value1
       9: invokevirtual #11                 // Method put:(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
      12: pop
      13: aload_0
      14: ldc           #17                 // String map2
      16: ldc           #19                 // String value2
      18: invokevirtual #11                 // Method put:(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
      21: pop
      22: aload_0
      23: ldc           #21                 // String map3
      25: ldc           #23                 // String value3
      27: invokevirtual #11                 // Method put:(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
      30: pop
      31: return
}

从这次的代码我们可以看出，已经没有putfield关键字这一行了，也就是说静态匿名类不会持有外部对象的引用了。

为什么静态内部类不会持有外部类的引用？

原因其实很简单，因为匿名内部类是静态的之后，它所引用的对象或属性也必须是静态的了，因此就可以直接从 JVM 的 Method Area（方法区）获取到引用而无需持久外部对象了。

双花括号的替代方案

即使声明为静态的变量可以避免内存泄漏，但依旧不建议这样使用，为什么呢？

原因很简单，项目一般都是需要团队协作的，假如那位老兄在不知情的情况下把你的static给删掉呢？这就相当于设置了一个隐形的“坑”，其他不知道的人，一不小心就跳进去了，所以我们可以尝试一些其他的方案，比如 Java8 中的 Stream API 和 Java9 中的集合工厂等。

替代方案 1：Stream

使用 Java8 中的 Stream API 替代，示例如下。原代码：

List<String> list = new ArrayList() {{
    add("Java");
    add("Redis");
}};

替代代码：

1	List<String> list = Stream.of("Java", "Redis").collect(Collectors.toList());

替代方案 2：集合工厂

使用集合工厂的of方法替代，示例如下。原代码：

Map map = new HashMap() {{
    put("map1", "value1");
    put("map2", "value2");
}};

替代代码：

1	Map map = Map.of("map1", "Java", "map2", "Redis");

显然使用 Java9 中的方案非常适合我们，简单又酷炫，只可惜我们还在用 Java 6...6...6...心碎了一地。

总结

本文我们讲了双花括号初始化因为会持有外部类的引用，从而可以会导致内存泄漏的问题，还从字节码以及反射的层面演示了这个问题。

要想保证双花括号初始化不会出现内存泄漏的办法也很简单，只需要被static修饰即可，但这样做还是存在潜在的风险，可能会被某人不小心删除掉，于是我们另寻它道，发现了可以使用 Java8 中的 Stream 或 Java9 中的集合工厂of方法替代双大括号。

被我误解的max_connect_errors

发表于 2020-05-22 | 分类于 mysql

原文：https://developer.aliyun.com/article/755441

第一节什么是max_connect_errors

一开始接触这个参数的时候，感觉他和max_connections的含义差不多，字面意思简单明了，这个参数的含义是最大连接错误数，翻翻mysql的文档中的解释是If more than this many successive connection requests from a host are interrupted without a successful connection, the server blocks that host from further connections,大意是：如果mysql服务器连续接收到了来自于同一个主机的请求，且这些连续的请求全部都没有成功的建立连接就被断开了，当这些连续的请求的累计值大于 max_connect_errors的设定值时，mysql服务器就会阻止这台主机后续的所有请求。”without a successful connection”那太好办了，故意输错密码不就行了，并且网上搜索了下该参数的说明，大量的文章充斥着” 防止暴力破解密码”的内容，于是兴高采烈的去做了测试。以下测试基于自建的mysql（非rds for mysql），由于rds for mysql无法直接设置set global，设置时需要在”rds控制台-参数这里”里进行设置：https://help.aliyun.com/document_detail/26179.html?spm=5176.11065259.1996646101.searchclickresult.44156de7pLffcV

第二节测试max_connect_errors

1,创建账号：

2,设置max_connect_errors为3：

3，故意输错密码3次，第四次使用正确密码登录进行验证：

4,结论是第四次依然可以登录，即密码不对（认证失败）不属于” ”without a successful connection””的范畴，网上的” 防止暴力破解密码”也不成立了。

第三节继续分析max_connect_errors

再继续看文档，发现还有以下说明：

You can unblock blocked hosts by flushing the host cache. To do so, issue a FLUSH HOSTS statement or execute a mysqladmin flush-hosts command.

大意是：

当你遇到主机被阻止的时候，你可以清空host cache来解决，具体的清空方法是执行flush hosts或者在mysql服务器的shell里执行 mysqladmin flush-hosts操作

既然清空host cache可以解决主机被阻止访问的问题，那应该与host cache有些关系，看看host cache的介绍可能会有些眉目，关于host cache，文档解释如下：

The MySQL server maintains a host cache in memory that contains information about clients: IP address, host name, and error information. The server uses this cache for nonlocal TCP connections. It does not use the cache for TCP connections established using a loopback interface address (127.0.0.1 or ::1), or for connections established using a Unix socket file, named pipe, or shared memory.

大意是：

Mysql服务器会在内存里管理一个host cache，host cache里保存了一些客户端的ip地址，主机名，以及这个客户端在与server建立连接时遇到的一些错误信息，host cache对不是本地的TCP连接才有效，所以host cache对127.0.0.1 或者::1是无效的，并且对于Unix socket file、named pipe以及 shared memory方式建立的连接也是无效的。并且通过了解，host cache的内容可以通过performance_schema.host_cache来查看，通过performance_schema.host_cache表里的几个列的描述信息，对之前的测试不成立的原因有些了解了，部分相关列如下：

IP
The IP address of the client that connected to the server, expressed as a string.

连接到mysql server的主机的连接地址

HOST
The resolved DNS host name for that client IP, or NULL if the name is unknown.

通过dns解析IP地址获取到的该IP地址对应的mysql client的主机名

SUM_CONNECT_ERRORS
The number of connection errors that are deemed “blocking” (assessed against the max_connect_errors system variable). Only protocol handshake errors are counted, and only for hosts that passed validation (HOST_VALIDATED = YES).
COUNT_HANDSHAKE_ERRORS
The number of errors detected at the wire protocol level.

通过SUM_CONNECT_ERRORS(连接错误计数)描述，重点是红色部分：只计算协议握手过程的错误(Only protocol handshake errors are counted),也就是说max_connect_errors 可能记录的是协议(不确定是tcp协议还是应用协议，通过抓包以及COUNT_HANDSHAKE_ERRORS的” the wire protocol level”说明可能是指应用协议)的握手过程中出现的错误，也就是可以说网络不好(无法顺利握手)会导致该问题。

第四节继续测试max_connect_errors

通过之前的说明，需要模拟应用协议握手失败的情况，最后考虑使用telnet一些来做测试

1,创建账号

2,设置max_connect_errors为3：

3,先使用telnet 10.26.254.217 3306连接3次，第四次使用正确的账号密码尝试登陆：

telnet前查看performance_schema.host_cache的记录为空

第一次telnet 10.26.254.217 3306

第二次 telnet 10.26.254.217 3306

第三次telnet 10.26.254.217 3306

第四次执行mysql -h10.26.254.217 -utestcon -p123 -P3306

问题复现了，出现了错误提示ERROR 1129 (HY000): Host ‘10.24.236.231’ is blocked because of many connection errors; unblock with ‘mysqladmin flush-hosts’

第五节 ERROR 1129 (HY000)问题延伸

解决ERROR 1129 (HY000)的方法是执行flush host或者 mysqladmin flush-hosts，其目的是为了清空host cache里的信息，那是不是说不使用host cache就可以了？使host cache不生效的方式有如下两种：

1,设置 host_cache_size 为0/ 打开skip-host-cache

2,打开skip-name-resolve

需要通过测试看下推测是否生效

5.1 设置 host_cache_size 为0/ 打开skip-host-cache
1,设置host_cache_size为0

2,再次查询performance_schema.host_cache

3,继续之前的测试：先使用telnet 10.26.254.217 3306连接3次，第四次使用正确的账号密码尝试登陆

更改已经生效，max_connect_errors的作用无效了

5.2 打开skip-name-resolve
1,在cnf配置文件里设置skip-name-resolve 以此打开skip-name-resolve

2,继续之前的测试：先使用telnet 10.26.254.217 3306连接3次，第四次使用正确的账号密码尝试登陆

3,查询performance_schema.host_cache

更改已经生效，max_connect_errors的作用无效了，RDS for mysql 的skip_name_resolve是on的状态，

所以很少会出现ERROR 1129 (HY000)的错误

动态规划的理解

发表于 2020-05-15

动态规划，一直以来听着就是一种很高深莫测的算法思想。尤其是上学时候算法的第一堂课，老师巴拉巴拉列了一大堆的算法核心思想，贪心、回溯、动态规划… …，开始感觉要在算法世界里游刃有余的进行解决各种各样牛B问题了，没想到的还是稀里糊涂学过了之后还就真的是学过了（大学的课程还真是一个样子）。再后来才明白，大学的课程一般来说就是入门级讲解，用来开拓眼界的，真正想要有一番自己的见解，必须要在背后下一番辛苦，形成自己的思考逻辑。

再后来返回头来看，动态规划理解起来还是比较困难，什么重叠子问题、动态转移方程，优化点等等等等，稀里糊涂，最后痛定思痛，好好看着其他人的分享理解了一部分，疯狂刷题几十道。算是基本可以佛挡杀佛了.

在我的这些学习积累过程中，总结出来希望可以给到大家一点小小的帮助，相信在读完这篇文章的时候，你会感觉到动态规划给你带来的奇妙之处。也一定对动态规划形成自己的思考方式. 很🐂的DP！！！

首先，先大致列下这篇文章会讲到什么

1.相较于暴力解法，动态规划带给我们的是什么？为什么会有重叠子问题以及怎么去避免的?

2.用不同难度的动态规划问题举例说明，最后会使用《打家劫舍》系列三个题再重温一次.

看完本篇文章后，相信大家会对DP问题会有一个初步的思考，一定会入门。后面大家可以继续练习相关问题，熟能生巧，思考的多了就会形成自己的思维逻辑.

好了，话不多说，开搞…

一、动态规划带给我们的优势

很有趣，一定要看完，必定有收获，加油！💪💪💪

平时在我们算法设计的过程中，一般讲求的是算法的执行效率和空间效率的利用情况

也就是我们熟知的时间复杂度（执行时耗费时间的长度）和空间复杂度（执行时占用存储单元的长度）

那下面用时间复杂度和空间复杂度来评估下传统算法设计和用动态规划思想解决下的效率情况

传统递归 vs. DP

先用一个被大佬们举例举到烂的🌰，这个栗子很烂，但是真的很香：必须着重强调.

《斐波那契（Fibonacci）数列的第n项》

举荐理由：在我自己看来Fibonacci是动态规划设计中的入门级案例，就好比说编程中的“hello world”，大数据中的“word count”.

Fibonacci几乎完美的诠释了动态规划带来的思想和技巧然而没有任何其他的要考虑的细枝末节，这种很清晰的方法看起来很适合整个的动态规划的思维方式，很适合入门来进行的思考方式.

接下来咱们先来看题目：

写一个函数，输入n，求斐波那契（Fibonacci）数列的第 n 项。斐波那契数列的定义如下：

F(0) = 0, F(1) = 1
F(N) = F(N - 1) + F(N - 2), 其中 N > 1.
斐波那契数列由 0 和 1 开始，之后的斐波那契数就是由之前的两数相加而得出。

比较一下传统递归解法和动态规划思想下的解决对比

1. 先递归解决

传统对于这种题目的思考方式会利用递归求解，做起来比较简单，就是不断的去递归调用，看下面代码：

class Solution(object):
    i = 0
    def fib_recur(self, N):
        print "F(",self.i,") = ", N   # 此处仅仅来看递归输出的N
        self.i += 1
        
        if N <= 1:
            return N
        return self.fib_recur(N-1) + self.fib_recur(N-2)  # 递归输出

输出的结果：

重复计算

明显可以看到，总计 8 次的计算过程中，相同的计算结果有三对进行了重复计算（下图中同色项，不包含灰色），也就是说在递归的过程中，把曾经计算过的项进行了又一次的重复计算，这样对于时间效率是比较低的，唯一的好处可能就是代码看起来比较好懂，但是终归不是一个好的算法设计方法。

代码中，在计算N的时候就去递归计算 fib(N-1) + fib(N-2)，那么，这种情况下的计算过程中。会是下面图中的一个计算过程。

可以发现，会有相当一部分的重复计算，这样对于时间都是重复的消耗。

参考图中相同颜色的项，比如说粉色的重复计算、黄色的重复计算等

注意：递归中没有对空间进行了增加，始终都是同样的长度，仅仅是不断的弹出和压入

为了更好的说明这种重复计算带来时间效率的低下。再比如说，相比上述图中的计算节点，再增加一个节点的计算，增加计算F(5)，那么由于递归的计算方式，会有更多的项（下图中线框中部分）进行了重复的计算。在计算F(5)的时候，会递归调用F(4)和F(3)，而在下图中，计算F(4)的时候，又会完整的去计算F(3)。这样，如果N很大的话，会有更大的时间消耗.

这样，这棵树的规模进行进行成倍增加，时间复杂度很明显的进行了成倍的扩张。对于时间上来说是很恐怖的.

时间复杂度带来的低效率严重超过了代码的可读性，所以我们可以想办法将过去计算过的节点进行保存。这样，我们就会用到下面要说的动态规划思想带来的时间上的高效.

时间复杂度： —> 指数级

空间复杂度：

2. 后动态规划解决

大概解释一下字面意思：

动态规划：\我们不直接去解决问题，而是在每一步解决问题的时候，达到每一步的最优情况。换句话说，就是在每一步解决问题过程中，利用*过去的状态*以及当前状态的情况而达到一个当前的最优状态.

规划：\在一般解决该类问题的时候，会有一个“填表格”的过程，无论是简单情况下的*一维表格*还是复杂一点的二维表格，都是以开辟空间换时间的思想，以争取最佳的时间效率. （保存过程中间值，方便后续直接使用）.

动态：用上面的案例来说，递归解决过程中的每一步都会从基本问题不断的“自顶向下”去求解，在每一步骤中，会有相同的计算逻辑进行了重复的计算。相比于递归思想，动态规划思想增加了对历史上计算结果的保存，逐步记录下中间的计算结果，在每一步求得最优值.

因此，动态规划可以避免重复计算，达到了时间上的最优，从指数级变为常数级别，相较于开辟的一段内存空间存放中间过程值的开销，是非常值得的.

那么，接下来咱们依照动态规划的思路进行对Fibonacci进行下解决

依据题中的规则：

F(0) = 0, F(1) = 1

F(N) = F(N - 1) + F(N - 2), when N > 1

那么，👇👇F(N) 的值只与他的前两个状态有关系👇👇

a. 初始化值 : F(0) = 0, F(1) = 1 b. 想要计算得到F(2)，那么F(2) = F(0) + F(1) –> 保存 F(2) c. 想要计算得到F(3)，那么F(3) = F(2) + F(1) –> 保存 F(3) d. 想要计算得到F(3)，那么F(4) = F(3) + F(2) –> 保存 F(4)

利用动态规划思想，以一维数组辅助实现的Fibonacci，看下图

是不是很简单的思路，仅仅靠保存过程中的一些值就能很简单的利用循环就可以实现了，没必要用递归反复计算进行实现。

想要计算得到第 n 个值的多少？那么，以下几点是我们必须要做到的

a. 定义一个一维数组 —> 一般用dp来命名

b. 动态方程的设定 —> 题中的F(N) = F(N - 1) + F(N - 2)

c. 初始化数值 —> F(0) = 0和F(1) = 1

上述的 a、b 和 c 点就是动态规划思想的几个核心要素

下面来看下要实现的代码（代码中，用dp来代替上面的F()）

class Solution(object):
    def fib(self, N):
        if N == 0:
            return 0
     
        dp = [0 for _ in range(N+1)]  # 1定义dp[i]保存第i个计算得到的数值
        dp[0] = 0    # 2初始化
        dp[1] = 1   # 2初始化
        for i in range(2, N+1): # 3动态方程实现，由于0和1都实现了赋值，现在需要从第2个位置开始赋值
            dp[i] = dp[i - 1] + dp[i - 2]
       
        print dp   # 记录计算过程中的次数，与上述递归形成对比
        return dp[N]

输出:

1 2	[0, 1, 1, 2, 3] 3

以上，最重要的就是1 2 3 点，而执行过程参照输出对比递归算法，计算少了很多，同样的计算只计算了一次。

时间复杂度：

空间复杂度：

介绍了上面的内容了，此处来条分割线吧，针对上述的 递归 vs. DP

既然动态规划的方案也介绍了，下面咱们再仔细看看，是否有优化的空间，毕竟对于一个算法方案的设计，都有找到其优化点，无论是时间还是空间的效率都想要达到一个理想的值。

3. 动态规划 + 优化

咱们看下这张图解，发现每个计算节点都只与前两个项有关系。换句话说，咱们只要保存两个值就好了，计算新的节点值的时候，把新的值赋值给前两个值的第一个就好了

话说只要两个值，现在定义两个变量 dp1 和 dp2。那么，现在咱们一步一步模拟一下：

a. 初始化值 : F(0) = 0, F(1) = 1

b. 想要计算得到F(2)，那么F(2) = F(0) + F(1) –> 保存 F(2)

顺带将F(1)赋值给dp1， f(2)赋值给dp2

c. 想要计算得到F(3)，那么F(3) = F(2) + F(1) –> 保存 F(3)

顺带将F(2)赋值给dp1， F(3)赋值给dp2

d. 想要计算得到F(3)，那么F(4) = F(3) + F(2) –> 保存 F(4)

顺带将F(3)赋值给dp1， F(4)赋值给dp2

至此为止，整个过程仅仅用到了两个变量来存储过程中产生的值，也就之前没有优化的空间效率得到了优化

咱们把代码也贴一下吧，供参考

class Solution(object):
    def fib_dp1(self, N):
        if N == 0: return 0

        dp1, dp2 = 0, 1

        for i in range(2, N+1):
            dp1 = dp1 + dp2
            dp1, dp2 = dp2, dp1

        return dp2

看起来是不是更加简洁了。

洋洋洒洒不知不觉写了这么多了。

如果有读者说这太简单了，我这篇文章内容面对的是小白级别的，如果读者是中等往上的水平，可直接跳到后面的案例三开始参考。

另外，如果有任何的意见可随时对我的文章进行评论，欢迎&感谢大家一起讨论

也可关注公众号「计算广告生态」，回复“DP”，获取该文件的pdf文件

大家感觉这个例子怎么样，三点说明：1.定义dp数组 2.动态方程 3.初始化数值

这也说明了为什么用斐波那契数列来引入动态规划的，因为斐波那契数列本身就明确的告诉你动态方程是什么，初始化的值是什么，所以好好的体会这种思想，尤其是从传统递归 -> 动态规划的思想解决，再到优化的方面，很值得深思。

那接下来，咱们就找几个有代表性的栗子来尝尝鲜

二、动态规划四大解题步骤处理问题

上面用斐波那契数列问题，引出了下面的几点，在这里再详细赘述一下

在后面的案例中将会尽量严格按照这几个步骤进行解决问题

步骤一：定义dp数组的含义

步骤二：定义状态转移方程

步骤三：初始化过程转移的初始值

步骤四：可优化点(可选)

步骤一：定义dp数组的含义

绝大部分情况下，我们需要定义一维数组或者二维数组进行存储在计算过程中产生的最优值，这里为什么是最优值呢？是因为在解决问题过程中，一般情况dp数组用来保存从开始到当前情况的最优值，故而保存的是截止到目前的最优值，避免重复计算（这里看起来思维有混乱的同学们，想想上面Fibonacci 递归解法和动态规划的对比）

所以，dp无论是一维的还是二维的，要想清楚代表什么，一般来说代表的是截止到目前情况下的最优值

步骤二：定义状态转移方程

什么是动态转移方程？如果有一个问题摆在我们面前，然后这个问题在解决的过程中，会发现有很多的重叠的子问题，重叠子结构，而通过这些子问题的解决，最终将会把该问题进行解决

通俗来说，在解决问题过程中，能够发现一个不断解决子问题的动态规律，比如说Fibonacci中的F(N) = F(N - 1) + F(N - 2)，而在其他的可以用动态规划解决的问题中，需要我们自己去发现这样的内在规律。这个是最难的也是最终于要的，只要这一步解决了，接下来我们解决这个问题基本就没问题了.

步骤三：初始化过程转移的初始值

顺着步骤二的思路来，既然动态方程定义好了，是不是需要一个支点来撬动它进行不断的计算下去。

那么，这个支点就需要我们来初始定义，将动态方程激活，进行计算。举例来说Fibonacci中的F(0) = 0和F(1) = 1，有了这两个值，它的动态方程F(N) = F(N - 1) + F(N - 2)就可以进行下去了

这个就是我们要想好的初始值，实际问题可能还需要我们想想清楚.

步骤四：可优化点(可选)

可优化的这里，最重要的会是dp数组这块，也会有不同问题不同的优化点

在例子中，我们会进行不同的优化.

总之一点，建议大家动笔多画画图，很多细节慢慢就会出现了.

案例一：打家劫舍I 「来自leetcode198」

你是一个专业的小偷，计划偷窃沿街的房屋。每间房内都藏有一定的现金，影响你偷窃的唯一制约因素就是相邻的房屋装有相互连通的防盗系统，如果两间相邻的房屋在同一晚上被小偷闯入，系统会自动报警。

给定一个代表每个房屋存放金额的非负整数数组，计算你在不触动警报装置的情况下，能够偷窃到的最高金额。

示例 1:

输入: [1,2,3,1]
输出: 4
解释: 偷窃 1 号房屋 (金额 = 1) ，然后偷窃 3 号房屋 (金额 = 3)。
     偷窃到的最高金额 = 1 + 3 = 4 。

示例2：

输入: [2,7,9,3,1]
输出: 12
解释: 偷窃 1 号房屋 (金额 = 2), 偷窃 3 号房屋 (金额 = 9)，接着偷窃 5 号房屋 (金额 = 1)。
     偷窃到的最高金额 = 2 + 9 + 1 = 12 。

把经典案例系列拆分开讨论下吧，咱们首先将「打家劫舍I」来看看

该题可以用动态规划的思想来解决的原因是，在小偷不断偷取的过程中，始终想要偷得的物品价值最大，最优，每一步骤都与之前的偷取情况有关系，而且每一步都要考虑是否能偷，是否会带来最大利益，这就使得我们可以用动态规划的思想来解决问题。然后严格按照四步骤进行解题.

步骤一：定义dp数组的含义

之前提到的，dp数组存储的值一般代表截止目前的最优值，在该题目中，我们定义：

dp[i] 代表到达第 i 个房屋偷得的最高金额，也就是当前最大子序和

无论房屋有几间，最后我们取到dp数组的最后一个值就求得小偷偷得的最高金额

步骤二：找出关系元素间的动态方程

动态规划解决的问题，一般来说就是解决最优子问题，“自顶向下” 的去不断的计算每一步骤的最优值。

也就是想要得到dp[i]的值，我们必须要知道dp[i-1]，dp[i-2]，dp[i-3] … 的每一步的最优值，在这个状态转移的过程中，我们必须要想清楚怎么去定义关系式。然而在每一步的计算中，都与前几项有关系，这个固定的关系就是我们要寻找的重叠子问题，也同样是接下来要详细定义的动态方程

该题目中，当小偷到达第 i 个屋子的时候，他的选择有两种：一种是偷，另外一种是不偷，然后选择价值较大者

a. 偷的情况计算：必然是dp[3] = nums[2] + dp[1]，如果是偷取该屋子的话，相邻屋子是不能偷取的，因此，通项式子是：dp[i] = nums[i-1] + dp[i-2]

b. 不偷的情况计算：必然是dp[3] = dp[2]，如果是不偷取该屋子的话，相邻屋子就是其最优值，因此，通项式子是：dp[i] = dp[i-1]

最后，要想偷得最高金额，那么，必须选取在偷与不偷之间的最大值作为我们是否选取的关键点。即：

动态方程: dp[i] = max(dp[i-1], nums[i-1]+dp[i-2])

步骤三：初始化数值设定

初始化: 给没有房子时，dp一个位置，即：dp[0] 1 当size=0时，没有房子，dp[0]=0；2 当size=1时，有一间房子，偷即可：dp[1]=nums[0]

那么，按照这个思路来整理一下代码：

class Solution(object):

    def rob(self, nums):
      # 1.dp[i] 代表当前最大子序和
      # 2.动态方程: dp[i] = max(dp[i-1], nums[i-1]+dp[i-2])
      # 3.初始化: 给没有房子时，dp一个位置，即：dp[0]
      #   3.1 当size=0时，没有房子，dp[0]=0；
      #   3.2 当size=1时，有一间房子，偷即可：dp[1]=nums[0]
      size = len(nums)
      if size == 0:
        return 0

      dp = [0 for _ in range(size+1)]

      dp[0] = 0
      dp[1] = nums[0]
      for i in range(2, size+1):
        dp[i] = max(dp[i-1], nums[i-1]+dp[i-2])
        return dp[size]

时间复杂度：O(N)

空间复杂度：O(N)

那下面想想看有没有可优化的地方，尽量的释放一部分计算机资源

步骤四：优化

从 dp[i] = max(dp[i-1], nums[i-1]+dp[i-2]) 关系来看，每一次动态变化，都与前两次状态有关系（dp[i-1], dp[i-2])，而前面的一些值是没有必要留存的.

所以，dp只需要定义两个变量就好，将空间复杂度降为O(1)

class Solution(object):

    def rob_o(self, nums):
        # 依照上面的思路，其实我们用到的数据永远都是dp的dp[i-1]和dp[i-2]两个变量
        # 因此，我们可以使用两个变量来存放前两个状态值
        # 空间使用由O(N) -> O(1)

        size = len(nums)
        if size == 0:
            return 0

        dp1 = 0
        dp2 = nums[0]
        for i in range(2, size+1):
            dp1 = max(dp2, nums[i-1]+dp1)
            dp1, dp2 = dp2, dp1
        return dp2

时间复杂度：O(N)

空间复杂度：O(1)

说完《打家劫舍I 》，中间穿插另外一道题目，利用二维dp来解决的一个问题。

最后再说说《打家劫舍II 》和《打家劫舍III》，把这一系列的打家劫舍问题搞明白了，相信你对动态规划有一个较为深刻的入门体验

如果有读者说这太简单了，我这篇文章内容面对的是小白级别的，如果读者是中等往上的水平，可直接跳到后面的案例三开始参考。

另外，如果有任何的意见可随时对我的文章进行评论，欢迎&感谢大家一起讨论

案例二：不同路径「来自leetcode62」

一个机器人位于一个 m x n 网格的左上角（起始点在下图中标记为“Start” ）。

机器人每次只能向下或者向右移动一步。机器人试图达到网格的右下角（在下图中标记为“Finish”）。

问总共有多少条不同的路径？

示例 1:

输入: m = 3, n = 2
输出: 3
解释:
从左上角开始，总共有 3 条路径可以到达右下角。

1. 向右 -> 向右 -> 向下
2. 向右 -> 向下 -> 向右
3. 向下 -> 向右 -> 向右

示例 2:

1 2	输入: m = 7, n = 3 输出: 28

提示：

1 <= m, n <= 100 题目数据保证答案小于等于 2 * 10 ^ 9

下面依然按照四个步骤来进行讨论：

步骤一：定义dp数组的含义

当前这道题是从左上角到右下角的，题目中规定只能向右或者向下走，所以我们必须要定义一个二维数组来保存计算过程中的值。所以，这块定义：

dp[i][j]: 代表到达位置 (i, j) 的所有路径的总数

即：机器人从左上角到右下角所有路径的总和，dp中每个位置的值代表行走到达 (i, j) 每个位置的总共的路径数

步骤二：找出关系元素间的动态方程

由于题目中规定只能向右或者向下走，所以在机器人行进的时候，只能是向右或向下.

那么，分别讨论下两种情况，想要到达位置(i, j)，可以从位置(i-1, j)或者(i, j-1)出发到达。因此，到达位置(i, j) 的总的路径数一定是 到达位置(i-1, j)路径数 + 到达位置(i, j-1)路径数。那么，现在可以定义动态方程：

动态方程：dp[i][j] = dp[i-1][j] + dp[i][j-1]

步骤三：初始化数值设定

很明显，在机器人走第 0 行，第 0 列的时候，无论怎么走，都只有 1 种走法。

因此，初始化值的设定，一定是 dp[0..m][1] 或者 dp[1][0..n] 都等于1

因此初始值如下：

dp[0] [0….n-1] = 1; // 机器人一直向右走，第 0 列统统为 1

dp[0…m-1] [0] = 1; // 机器人一直向下走，第 0 列统统为 1

现在，按照这个思路来整理一下代码

class Solution(object):

    def uniquePaths1(self, m, n):

        # 初始化表格，由于初始化0行 0列都为1。那么，先全部置为1
        dp = [[1 for _ in range(m)] for _ in range(n)]

        for i in range(1, n):
            for j in range(1, m):
                dp[i][j] = dp[i-1][j] + dp[i][j-1]

        return dp[n-1][m-1]

上述代码中由于dp[0..m][1] 或者 dp[1][0..n] 都等于1，所以在定义二维数组dp时候，统统赋初始值为 1

然后从位置(1, 1)开始计算每个位置的总路径数

时间复杂度：O(M*N)

空间复杂度：O(M*N)

既然到这里了，下面再想想看有没有可优化的地方

步骤四：优化

可以依照前面的解决的思路，应该也可以从空间上进行一定的优化

参照前面的案例，之前定义的是一维数组dp，优化点是每一步骤都只与前面的两个计算好的数值有关系，然后优化点就是将dp[N] -> dp1和dp2，空间复杂度由 O(N) -> O(1)，如果是很大规模的数据计算的话，空间效率提升了不少.

现在这个例子中的动态方程是dp[i][j] = dp[i-1][j] + dp[i][j-1]，很明显，每一步骤中的状态值只与左边相邻的值和上面的值相关。举例（为了方便，用 3*4 来举例）：

这个完整的图片描述中，机器人从左上角的位置(1, 1)开始移动，逐渐每一步都根据动态方程进行前进，明显的可以看出机器人每移动一格，所得到的路径总和只与它的上方和左方数值有关系。也就是我们会发现，机器人移动到第2行的时候，第0行数据完全是没有用的状态。

因此，这个优化点就出来了，在算法设计的时候，dp仅仅定义2行N列的数组就ok了，省去了m-2行的空间开销。这个代码如果大家想明白了请自行设计出来，自己写出来一定会有更加深刻的理解，再强调：多思考，形成潜移默化的思维方式.

看完这个步骤之后，是不是很明显的优化点，为什么上面没有给出大家代码呢？是因为我看到貌似可以继续优化的点（粘住空间优化项了哈哈哈），那就继续在空间开销上做文章。

引导：根据上述咱们的优化方案，说道 “机器人移动到第2行的时候，第0行数据完全是没有用的状态”，其实当前聪明的读者你想想，再看看，下面的图中（从上图截取过来）。其实，不仅仅是第 0 行完全没用了，而且在第2 行做移动的时候，移动到位置(i, j)的时候，计算好位置(i, j)，那么接下来，位置(i-1, j)的数据也就没用了。换句话说，边走，第 1 行开始的某些数据也就没用了，还在占着空间

这块大家一定多想想，多理解，多画图

下面按照这种思路，看下图的步骤，也画好了用一维数组进行解决问题，也画出来每一步骤与上图的类比过程：

在这里，有犯困的同学可以自己动手画一画，理解一下，个人感觉是一个很好的思维扩展

接下来，就按照这样的思路进行代码实现，会发现码起来很简单

class Solution(object):

    def uniquePaths2(self, m, n):
        if m > n:
            m, n = n, m

        dp = [1 for _ in range(m)]

        for i in range(1, n):
            for j in range(1, m):
                dp[j] = dp[j] + dp[j-1]

        return dp[m-1]

时间复杂度：O(m*n)

空间复杂度：O(min(m ,n))

是不是从思维方面简单干净了许多

搞清楚上面的栗子之后呢，我们将上面的例题进行一个简单的难度增加，说白了，就是在路上打几个阻碍点

来看:

案例三：不同路径II 「来自leetcode63」

一个机器人位于一个 m x n 网格的左上角（起始点在下图中标记为“Start” ）。

机器人每次只能向下或者向右移动一步。机器人试图达到网格的右下角（在下图中标记为“Finish”）。

现在考虑网格中有障碍物。那么从左上角到右下角将会有多少条不同的路径？

说明：m 和 n 的值均不超过 100。

示例 1:

输入:
[
  [0,0,0],
  [0,1,0],
  [0,0,0]
]
输出: 2
解释:
3x3 网格的正中间有一个障碍物。
从左上角到右下角一共有 2 条不同的路径：

1. 向右 -> 向右 -> 向下 -> 向下
2. 向下 -> 向下 -> 向右 -> 向右

咱们先看一下题中的两个关键点： 关键点1：只能向右或者向下关键点2：有障碍物为1，无障碍物为0

根据关键点1 和关键点2 依然按照四个步骤来进行讨论：

步骤一：定义dp数组的含义

这个题目中定义的dp数组是和上一个例题中定义的dp数组的含义是相同的，但由于该题中已经定义有数组obstacleGrid，可以直接利用，无需额外开辟空间那么，就利用obstacleGrid作为动态规划中存储计算过程中的最优值

步骤二：找出关系元素间的动态方程

参照上一题目，规定动态方程: obstacleGrid[i][j] = obstacleGrid[i-1][j] + obstacleGrid[i][j-1] 由于机器人在移动过程中有障碍物，那么，对上面动态方程加一些限制条件 a.若当前 obstacleGrid[i][j] 为0。那么，直接计算动态方程下的计算过程 b.若当前 obstacleGrid[i][j] 不为0。那么，直接置该位置的值为0

所以，在进行动态方程遍历的时候，先进行 obstacleGrid[i][j]的判断，再进行动态方程的计算执行

步骤三：初始化数值设定

相比于上一题目，相似的是，在机器人走第 0 行，第 0 列的时候，无论怎么走，都只有 1 种走法但由于有障碍物，那走到障碍物的时候，后面都是走不下去的（下图用第一行来举例）。

所以，初始化第 0 行，第 0 列的时候，障碍物 1 后面的都是不可达的。所以，初始化行和列的逻辑表达：

该位置是否可达=前一个位置的状态and该位置能否可达得到能否到达这个位置

只有前一个位置为1（可达，只有1种方式），当前位置为0（无障碍物）这种情况才能到达该位置，然后为该位置设 1 （可达，只有1种方式）

# 0 行初始化表达式: 
obstacleGrid[0][row] = int(obstacleGrid[0][row] == 0 and obstacleGrid[0][row-1] == 1)
# 0 列初始化表达式: 
obstacleGrid[clo][0] = int(obstacleGrid[clo][0] == 0 and obstacleGrid[clo-1][0] == 1)

这些都准备就绪之后，按照相关思路进行编码

class Solution(object):

    def uniquePathsWithObstacles1(self, obstacleGrid):
       # 行列长度
        m = len(obstacleGrid)
        n = len(obstacleGrid[0])

        # 如果在位置(0, 0)，哪里都去不了，直接返回0
        if obstacleGrid[0][0] == 1:
            return 0

        # 否则，位置(0, 0)可以到达
        obstacleGrid[0][0] = 1

        # 初始化 0 列
        for clo in range(1, m):
            obstacleGrid[clo][0] = int(obstacleGrid[clo][0] == 0 and obstacleGrid[clo-1][0] == 1)

        # 初始化 0 行
        for row in range(1, n):
            obstacleGrid[0][row] = int(obstacleGrid[0][row] == 0 and obstacleGrid[0][row-1] == 1)

        # 从位置(1, 1)根据动态方程开始计算
        for i in range(1, m):
            for j in range(1, n):
                if obstacleGrid[i][j] == 0:
                    obstacleGrid[i][j] = obstacleGrid[i-1][j] + obstacleGrid[i][j-1]
                else:
                    obstacleGrid[i][j] = 0

        return obstacleGrid[m-1][n-1]

时间复杂度: O(mxn)

空间复杂度: O(1)

步骤四：优化

这块的优化先不谈了，这里基本没有什么优化点，之前都是由于自己要开辟内存空间，通过空间的优化来进行，而本题是在给定的数组中进行操作的

有了这几个案例的基础之后，咱们后面把经典的《打家劫舍》系列剩下的两个题目讨论完，就先告一段落，后面也希望以不同的方式与大家多多交流，互相学习

如果有读者看着累了，可以先保存，收藏下来，待消化了前面的内容，方便再回来看看。

案例四：打家劫舍II 「来自leetcode213」

你是一个专业的小偷，计划偷窃沿街的房屋，每间房内都藏有一定的现金。这个地方所有的房屋都围成一圈，这意味着第一个房屋和最后一个房屋是紧挨着的。同时，相邻的房屋装有相互连通的防盗系统，如果两间相邻的房屋在同一晚上被小偷闯入，系统会自动报警。

给定一个代表每个房屋存放金额的非负整数数组，计算你在不触动警报装置的情况下，能够偷窃到的最高金额。

示例 1:

1
2
3

输入: [2,3,2]
输出: 3
解释: 你不能先偷窃 1 号房屋（金额 = 2），然后偷窃 3 号房屋（金额 = 2）, 因为他们是相邻的。

示例 2:

输入: [1,2,3,1]
输出: 4
解释: 你可以先偷窃 1 号房屋（金额 = 1），然后偷窃 3 号房屋（金额 = 3）。
     偷窃到的最高金额 = 1 + 3 = 4 。

与《打家劫舍I》不同的是，《打家劫舍I》的屋子是线性的，而《打家劫舍II》是环状的，所以要考虑的点会增加一些，因为首位相连接的情况，咱们分为下面三种情况进行设定：

a. 不偷首偷尾

b. 偷首不偷尾

c. 首位都不偷显然，c 种方式损失太大，不会获得最高的金额，故选取 a 和 b。那么，下面分为两种情况，分别计算不包含首和不包含尾这两种情况来判断小偷哪种方式偷取的金额最高。

下面依然按照之前的四个步骤来进行分析

步骤一：定义dp数组的含义

dp[i] 代表的含义和之前一致，dp数组存储的值一般代表截止目前的最优值

所以，dp[i] 代表到达第 i 个房屋偷得的最高金额，也就是当前最大子序和

但是最后会讨论不包含首和不包含尾这两种情况下得到的dp数组的最后一位，然后获取其中较大者，就是我们要取得的最高金额

#####步骤二：找出关系元素间的动态方程

动态方程可参照《打家劫舍I》，有很详细的图解过程，该例子动态方程的变化和之前是完全一致的：

dp[i] = max(dp[i-1], nums[i-1]+dp[i-2])

步骤三：初始化设定

初始化: 给没有房子时，dp一个位置，即：dp[0]

a. 当size=0时，没有房子，小偷没办法偷：dp[0]=0；

b. 当size=1时，有一间房子，只要偷即可：dp[1]=nums[0]

由于屋子首位相连接，所以在计算时候，直接分为两种情况。第一种略过第一个屋子，第二种略过第二个屋子，这样得到的两个数组结果。最后只要比较最后一位数值的大小就ok了。解决！

该例子步骤三之后，感兴趣的同学可以自己写一下代码，和《打家劫舍I》的代码很类似，后面我写了优化后的代码，可能会更加的明白怎么写。咱们直接到步骤四，有了上面的案例，直接来看看优化后的方案:

步骤四：优化

同样从 dp[i] = max(dp[i-1], nums[i-1]+dp[i-2]) 关系来看，每一次动态变化，都与前两次状态有关系（dp[i-1], dp[i-2])，而前面的一些值是没有必要留存的，只要保存两个变量来保存过程最优值就好.

代码中有详细的注释：

class Solution(object):

    def rob(self, nums):
        # 点睛：与打家劫舍I的区别是屋子围成了一个环
        #   那么，很明显可以分为三种情况：
        #   1. 首位都不偷
        #   2. 偷首不偷尾
        #   3. 不偷首偷尾
        # 显然，第1种方式损失太大，选取2、3。
        # 那么，分为两种情况，分别计算不包含首和不包含尾这两种情况来判断哪个大哪个小

        # 1.dp[i] 代表当前最大子序和
        # 2.动态方程: dp[i] = max(dp[i-1] and , nums[i-1]+dp[i-2])
        # 3.初始化: 给没有房子时，dp一个位置，即：dp[0]
        #   3.1 当size=0时，没有房子，dp[0]=0；
        #   3.2 当size=1时，有一间房子，偷即可：dp[1]=nums[0]

        # 依照《打家劫舍I》的优化方案进行计算

        # nums处理，分别切割出去首和去尾的子串
        nums1 = nums[1:]
        nums2 = nums[:-1]

        size = len(nums)
        if size == 0:
            return 0
        if size == 1:
            return nums[0]

        def handle(size, nums):
            dp1 = 0
            dp2 = nums[0]
            for i in range(2, size+1):
                dp1 = max(dp2, nums[i-1]+dp1)
                dp1, dp2 = dp2, dp1
            return dp2

        res1 = handle(size-1, nums1)
        res2 = handle(size-1, nums2)

        return max(res1, res2)

时间复杂度：O(N)

空间复杂度：O(1)

再看看下面小偷遇到的情况，感叹：即使当小偷，也要步步规划才能拿到最高的金额啊…

案例五：打家劫舍III 「来自leetcode337」

在上次打劫完一条街道之后和一圈房屋后，小偷又发现了一个新的可行窃的地区。这个地区只有一个入口，我们称之为“根”。除了“根”之外，每栋房子有且只有一个“父“房子与之相连。一番侦察之后，聪明的小偷意识到“这个地方的所有房屋的排列类似于一棵二叉树”。如果两个直接相连的房子在同一天晚上被打劫，房屋将自动报警。

计算在不触动警报的情况下，小偷一晚能够盗取的最高金额。

示例 1:

输入: [3,2,3,null,3,null,1]

    3
  / \
   2   3
    \   \ 
     3   1

输出: 7 
解释: 小偷一晚能够盗取的最高金额 = 3 + 3 + 1 = 7.

示例 2:

输入: [3,4,5,1,3,null,1]

    3
  / \
   4   5
  / \   \ 
 1   3   1
输出: 9
解释: 小偷一晚能够盗取的最高金额 = 4 + 5 = 9.

题目出的很好，但是立马会给人一种小偷也不是好当的的赶脚…

言归正传，咱们先来说说题目本身

《打家劫舍》的小偷从一维线性到环形，再到二维矩形的屋子？是我想简单了，直接就干到树形了，是不是看着很香，而且很想，看下去，研究研究…

来整理几点思路，再来按照四步走：

1.由于房屋是树状的，因此，我们可以使用遍历树的传统方法进行遍历(前序、中序、后续) 2.简单的思路是，从树低进行往上遍历，拿到最优的打劫值。可以选用后续遍历 3.得到每一节点的最优值，最后选取最优的结果

依然按照三个步骤来进行分析（无优化点）

步骤一：定义dp数组的含义

dp[i]代表该节点及以下打最多的劫（拿到最多的钱）

步骤二：找出关系元素间的动态方程

根据我们每走到一个节点，都会有两种情况，那就是 偷(1) 与 不偷(0)。我们分开来讨论：

a. 用 dp[0] 代表不偷取该节点到目前为止拿到最多的钱，那么儿子节点偷不偷都ok。

所以： dp[0] = max(left[0], left[1]) + max(right[0], right[1])

b. 用 dp[1] 代表偷了该节点到目前为止拿到最多的钱，则儿子节点都不能被偷。

所以：dp[1] = value + left[0] + right[0] (value代表该节点的价值)

有看不懂的地方吗？再紧接着解释一下：

left[0]代表不偷取左孩子拿到最高的金额

left[1]代表偷取左孩子拿到最高的金额

right[0]代表不偷取右孩子拿到最高的金额

right[1]代表偷取右孩子拿到最高的金额

如果还有不太懂的话，留言或者添加我的公众号「计算广告生态」联系我，随时骚扰我哈

步骤三：初始化设定

该例子的初始化比较简单，就是当前树的形状为空的时候，直接返回dp[0, 0]

下面贴出完整代码，其中包含树的初始化代码 && 一大堆的注释：

# Definition for a binary tree node.
class TreeNode(object):
    def __init__(self, x):
        self.val = x
        self.left = None
        self.right = None


class Solution():

    def rob(self, root):

        # 说明：
        # 1.由于房屋是树状的，因此，我们可以使用遍历树的传统方法进行遍历(前序、中序、后续)
        # 2.简单的思路是，从树低进行往上遍历，拿到最优的打劫值。可以选用后续遍历
        # 3.得到每一节点的最优值，最后选取最优的结果

        # 1.dp[i]代表该节点及以下拿到的最多的钱
        # 2.动态方程：
        #   2.1 dp[0]代表不偷该节点拿到最多的钱，则儿子节点偷不偷都ok。dp[0] = max(left[0], left[1]) + max(right[0], right[1])
        #   2.2 dp[1]代表偷了该节点拿到最多的钱，则儿子节点都不能被偷。dp[1] = var + left[0] + right[0]
        # 3.初始化：当前树的形状为空的时候，直接返回dp[0, 0]
        def postTrasval(root):
            dp = [0, 0]
            if not root:
                return dp
            left = postTrasval(root.left)
            right = postTrasval(root.right)

            dp[0] = max(left[0], left[1]) + max(right[0], right[1])
            dp[1] = root.val + left[0] + right[0]

            return dp

        dp = postTrasval(root)
        return max(dp[0], dp[1])


if __name__ == '__main__':
    # initial tree structure
    T = TreeNode(3)
    T.left = TreeNode(2)
    T.right = TreeNode(3)
    T.left.right = TreeNode(3)
    T.right.right = TreeNode(1)

    # The solution to the Question
    s = Solution()
    print(s.rob(T))

至此为止，想要讲解的全部完毕了

洋洋洒洒过万字，自己都没想到写了这么多

在强调一点吧，这些题目全部理解外加自己另外练习，一定能够cover关于动态规划80%以上的题目，基本上都是dp为一维数组，二维数组的题目，很少有很奇怪的题型出现。所以，本文将《打家劫舍》经典案例详细讲解了一次，还有不同路径的问题，也是很经典的题目，而经典题目一定很具有代表性。优化方向很多，本文也只介绍了关于空间方面的优化，因为这个是最最常见的。

最后，大家一定多画图，多思考，题解百边其义自见（百边有点多哈哈哈）

还有，多理解四步骤，加油！

centos8 通过 rancher2 部署 k8s

发表于 2020-05-01 | 分类于 devops ， rancher

环境：

192.168.145.150 rancher 2核4G

192.168.145.151 k8s-dn1 4核8G

192.168.145.152 k8s-dn2 4核8G

192.168.145.153 k8s-dn3 4核8G

域名：

192.168.145.150 rancher.dev

1、虚拟机安装centos

https://www.cnblogs.com/wcwen1990/p/7630545.html

2、先设置为NAT保证可以访问外网：

http://baijiahao.baidu.com/s?id=1597809303775176940&wfr=spider&for=pc

su root
vi /etc/sysconfig/network-scripts/ifcfg-ensXXXX

ONBOOT=yes
BOOTPROTO=static
IPADDR=192.168.145.150
NETMASK=255.255.255.0
GATEWAY=192.168.145.2
DNS1=192.168.145.2

# centos 7
service network restart
# centos 8
nmcli networking off && nmcli networking on 
ifdown eth0 && ifup eth0 
nmcli con down eth0 && nmcli con up eth0
systemctl restart NetworkManager

reboot
ping www.baidu.com

3、安装ifconfig

https://www.cnblogs.com/jtnote/p/6103754.html

1	yum install -y net-tools.x86_64

4、安装ssh

https://www.cnblogs.com/liuhouhou/p/8975812.html

yum install -y openssl openssh-serverifcon

vi /etc/ssh/sshd_config

RSAAuthentication yes
PubkeyAuthentication yes

systemctl restart sshd.service
systemctl enable sshd.service

5、用 xshell 连接 contos7

6、配置 hosts 系统文件


192.168.145.150 rancher
192.168.145.151 k8s-dn1
192.168.145.152 k8s-dn2
192.168.145.153 k8s-dn3

192.168.145.150 rancher.dev

7、修改时区

\cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime

vi /etc/sysconfig/clock
# 添加如下内容
ZONE="Asia/Shanghai"
UTC=false
ARC=false

8、安装 ntp，设置节点间 ntp 同步

# 查看是否安装：
rpm -q ntp
# centos 7 安装
yum install ntpdate ntp -y
# centos 8 安装
rpm -ivh http://mirrors.wlnmp.com/centos/wlnmp-release-centos.noarch.rpm
yum install wntp -y
ntpdate ntp1.aliyun.com

9、安装 ssh，配置节点间免密登录

设置主机名

hostnamectl set-hostname rancher
hostnamectl set-hostname k8s-dn1
hostnamectl set-hostname k8s-dn2
hostnamectl set-hostname k8s-dn3

依次在每台服务器上执行

su root
cd ~
ssh-keygen -t rsa
ssh-copy-id -i .ssh/id_rsa.pub root@rancher
ssh-copy-id -i .ssh/id_rsa.pub root@k8s-dn1
ssh-copy-id -i .ssh/id_rsa.pub root@k8s-dn2
ssh-copy-id -i .ssh/id_rsa.pub root@k8s-dn3

测试

ssh 'root@k8s-dn1'
ssh 'root@k8s-dn2'
ssh 'root@k8s-dn3'
ssh 'root@k8s-rancher'

10、关闭防火墙

1 2	systemctl stop firewalld systemctl disable firewalld

11、kernel调优

vi /etc/security/limits.conf 在文件末尾添加以下内容：

root soft nofile 655350
root hard nofile 655350
* soft nofile 655350
* hard nofile 655350

vi /etc/sysctl.conf 添加如下内容：

net.ipv4.ip_forward=1
net.bridge.bridge-nf-call-iptables=1
net.bridge.bridge-nf-call-ip6tables=1
vm.swappiness=0
vm.max_map_count=655360

12、关闭swap

vi /etc/fstab 注释或删除swap交换分区：

1
2
3

/dev/mapper/cl-root     /                       xfs     defaults        0 0
UUID=5fe3b563-4639-47d5-ab24-8161c324f532 /boot                   ext4    defaults        1 2
# /dev/mapper/cl-swap     swap                    swap    defaults        0 0

13、关闭 selinux

将SELINUX值设置为disabled：

1 2	vim /etc/selinux/config SELINUX=disabled

14、yum 源更新

阿里云 yum 源 https://developer.aliyun.com/mirror/centos?spm=a2c6h.13651102.0.0.3e221b11HiFxxt

各种源：https://developer.aliyun.com/mirror/

cd /etc/yum.repos.d/
rm -f CentOS-Base.repo CentOS-AppStream.repo CentOS-PowerTools.repo CentOS-centosplus.repo CentOS-Extras.repo

curl -o /etc/yum.repos.d/CentOS-Base.repo https://mirrors.aliyun.com/repo/Centos-8.repo

yum clean all && yum makecache

15、docker 安装配置

每个节点执行

yum remove docker \
                  docker-client \
                  docker-client-latest \
                  docker-common \
                  docker-latest \
                  docker-latest-logrotate \
                  docker-logrotate \
                  docker-engine

yum install -y yum-utils device-mapper-persistent-data lvm2
yum-config-manager --add-repo https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
yum install -y https://download.docker.com/linux/fedora/30/x86_64/stable/Packages/containerd.io-1.2.6-3.3.fc30.x86_64.rpm
yum install -y docker-ce docker-ce-cli containerd.io

systemctl start docker
docker run hello-world

systemctl enable docker

配置docker源

mkdir -p /etc/docker
vi /etc/docker/daemon.json
{
  "registry-mirrors": ["https://k8spv7nq.mirror.aliyuncs.com"]
}

systemctl daemon-reload
systemctl restart docker

16、docker安装rancher2

在rancher节点执行

1
2
3

docker run -d --restart=unless-stopped -p 80:80 -p 443:443 rancher/rancher

docker logs -f ec91d23fb68b

http://192.168.145.150

访问 rancher

在主机hosts添加

1
2
3

> C:\Windows\System32\drivers\etc\hosts
> 192.168.145.150  rancher.dev
>

https://rancher.dev

admin/123456

17、安装k8s

在 worker节点机器中配置hosts地址

1 2	vi /etc/hosts 192.168.145.150 rancher.dev

1
2
3

docker run -d --privileged --restart=unless-stopped --net=host -v /etc/kubernetes:/etc/kubernetes -v /var/run:/var/run rancher/rancher-agent:v2.4.5 --server https://rancher.dev --token vpcq7clf5r5v7vcck8wd759xjdb87bj966hzl5tdfjm7mgmdzc5zwq --ca-checksum 9e76ce060a22f87140fca47d350ccbf99d8d7790f76915db6dba7c50d2b95be6 --etcd --controlplane --worker

docker run -d --privileged --restart=unless-stopped --net=host -v /etc/kubernetes:/etc/kubernetes -v /var/run:/var/run rancher/rancher-agent:v2.4.5 --server https://rancher.dev --token vpcq7clf5r5v7vcck8wd759xjdb87bj966hzl5tdfjm7mgmdzc5zwq --ca-checksum 9e76ce060a22f87140fca47d350ccbf99d8d7790f76915db6dba7c50d2b95be6 --worker

耗时10分钟左右，集群可用

18、安装 kubectl 并配置 kubeconfig

在4台机器上安装

curl -LO https://storage.googleapis.com/kubernetes-release/release/v1.18.0/bin/linux/amd64/kubectl
chmod +x ./kubectl
mv ./kubectl /usr/local/bin/kubectl
kubectl version --client

配置kubeconfig

mkdir -p ~/.kube
vi ~/.kube/config
# 将上图中的文件内容，复制到文件中

kubectl get all

1.1、流批的本质

1.2、流计算的核心问题

延时问题

更新撤回

容错续跑

透明升级

乱序问题

正确性问题

1.3、Flink应用场景

事件驱动型应用

数据分析型应用

数据管道型应用

1.4、怎样理解流批一体融合？

2.1、并行处理和编程范式

2.2、DataStream API 概览

2.3、状态和时间

第三章 运行时架构

3.1 、Runtime 总览

3.2、 JobMaster — 作业的控制中心

3.3、TaskExecutor — 任务的运行容器

3.4、ResourceManager — 资源管理中心

第四章 流计算的状态 和 容错机制

4.1、流计算中的状态

4.2、全局一致性快照

全局一致性快照的实现方法

异步全局一致性快照算法

Chandy-Lamport 算法流程

4.3、Flink的容错机制

如何保证 Exactly once

复杂场景 —— 多流输入

降低一致性要求

通过异步快照减少停顿

4.4、Flink的状态管理

本地状态后端 JVM

RocksDB 作为状态后端

5.1、Flink SQL

5.3 实战Demo

6.1、PyFlink简介

6.2、PyFlink 功能介绍

Python Table API

Python UDF

Python UDF 架构

向量化 Python UDF

Python UDF Metrics

Python UDF 执行优化

6.4、PyFlink 下一步规划

第七章 Flink Ecosystems 连接外部生态系统

7.1、连接外部系统

7.2、常用Connector

Kafka Connector

Elasticsearch Connector

FileSystem Connector

Hive Connector

DataGen Connector

Print Connector

BlackHole Connector

7.3、示例 & Demo

kafka交互

写入es

hive交互

k8s简介和安装

1、k8s简介

3、基础知识

4、安装配置

内核升级

k8s核心概念：Pod、控制器、Services、网络通信

kubernetes 对象

基本对象

Pod

网络

存储

高级对象

DaemonSet

k8s 资源清单及编写、Pod生命周期

Job

CronJob

Pod的生命周期

探针：存活检测

postStart

preStop

第三章运行时架构

第四章流计算的状态和容错机制