如何画架构设计图

转载于知乎

定义

什么是架构

架构就是对系统中的实体以及实体之间的关系所进行的抽象描述，是一系列的决策。

架构是结构和愿景。

系统架构是概念的体现，是对物/信息的功能与形式元素之间的对应情况所做的分配，是对元素之间的关系以及元素同周边环境之间的关系所做的定义。

做好架构是个复杂的任务有了架构之后，就需要让干系人理解、遵循相关决策。

什么是架构图

系统架构图是为了抽象的表示软件系统的整体轮廓和各个组件之间的相互关系和约束边界，以及软件系统的物理部署和软件系统的演进方向的整体视图。

架构图的作用

一图胜千言。要让干系人理解、遵循架构决策，就需要把架构信息传递出去。架构图就是一个很好的载体。那么，画架构图是为了：

• 解决沟通障碍
• 达成共识
• 减少歧义

架构图分类

搜集了很多资料，分类有很多，有一种比较流行的是4+1视图，分别为场景视图、逻辑视图、物理视图、处理流程视图和开发视图。

场景视图

场景视图用于描述系统的参与者与功能用例间的关系,反映系统的最终需求和交互设计,通常由用例图表示。

逻辑视图

逻辑视图用于描述系统软件功能拆解后的组件关系,组件约束和边界,反映系统整体组成与系统如何构建的过程,通常由UML的组件图和类图来表示。

物理视图

物理视图用于描述系统软件到物理硬件的映射关系,反映出系统的组件是如何部署到一组可计算机器节点上,用于指导软件系统的部署实施过程

处理流程视图

处理流程视图用于描述系统软件组件之间的通信时序,数据的输入输出,反映系统的功能流程与数据流程,通常由时序图和流程图表示。

开发视图

开发视图用于描述系统的模块划分和组成,以及细化到内部包的组成设计,服务于开发人员,反映系统开发实施过程。

5 种架构视图从不同角度表示一个软件系统的不同特征，组合到一起作为架构蓝图描述系统架构。

好的架构图

上面的分类是前人的经验总结，图也是从网上摘来的，那么这些图画的好不好呢？是不是我们要依葫芦画瓢去画这样一些图？

先不去管这些图好不好，我们通过对这些图的分类以及作用，思考了一下，总结下来，我们认为，在画出一个好的架构图之前， 首先应该要明确其受众，再想清楚要给他们传递什么信息 ，所以，不要为了画一个物理视图去画物理视图，为了画一个逻辑视图去画逻辑视图，而应该根据受众的不同，传递的信息的不同，用图准确地表达出来，最后的图可能就是在这样一些分类里。那么，画出的图好不好的一个直接标准就是：受众有没有准确接收到想传递的信息。

明确这两点之后，从受众角度来说，一个好的架构图是不需要解释的，它应该是自描述的，并且要具备一致性和足够的准确性，能够与代码相呼应。

常见问题

• 方框代表什么？

为什么适用方框而不是圆形，它有什么特殊的含义吗？随意使用方框或者其它形状可能会引起混淆。

• 虚线、实线什么意思？箭头什么意思？颜色什么意思？

随意使用线条或者箭头可能会引起误会。

• 运行时与编译时冲突？层级冲突？

架构是一项复杂的工作，只使用单个图表来表示架构很容易造成莫名其妙的语义混乱。

如何更好的表达软件架构

C4

C4 模型使用容器（应用程序、数据存储、微服务等）、组件和代码来描述一个软件系统的静态结构。这几种图比较容易画，也给出了画图要点，但最关键的是，我们认为，它明确指出了每种图可能的受众以及意义。

下面的案例来自C4官网，然后加上了一些我们的理解。

语境图(System Context Diagram)

这是一个想象的待建设的互联网银行系统，它使用外部的大型机银行系统存取客户账户、交易信息，通过外部电邮系统给客户发邮件。可以看到，非常简单、清晰，相信不需要解释，都看的明白，里面包含了需要建设的系统本身，系统的客户，和这个系统有交互的周边系统。

用途

这样一个简单的图，可以告诉我们，要构建的系统是什么；它的用户是谁，谁会用它，它要如何融入已有的IT环境。这个图的受众可以是开发团队的内部人员、外部的技术或非技术人员。即：

• 构建的系统是什么
• 谁会用它
• 如何融入已有的IT环境

怎么画

中间是自己的系统，周围是用户和其它与之相互作用的系统。这个图的关键就是梳理清楚待建设系统的用户和高层次的依赖，梳理清楚了画下来只需要几分钟时间。

容器图(Container Diagram)

容器图是把语境图里待建设的系统做了一个展开。

上图中，除了用户和外围系统，要建设的系统包括一个基于java\spring mvc的web应用提供系统的功能入口，基于xamarin架构的手机app提供手机端的功能入口，一个基于java的api应用提供服务，一个mysql数据库用于存储，各个应用之间的交互都在箭头线上写明了。

看这张图的时候，不会去关注到图中是直角方框还是圆角方框，不会关注是实线箭头还是虚线箭头，甚至箭头的指向也没有引起太多注意。

我们有许多的画图方式，都对框、线的含义做了定义，这就需要画图的人和看图的人都清晰的理解这些定义，才能读全图里的信息，而现实是，这往往是非常高的一个要求，所以，很多图只能看个大概的含义。

用途

这个图的受众可以是团队内部或外部的开发人员，也可以是运维人员。用途可以罗列为：

• 展现了软件系统的整体形态
• 体现了高层次的技术决策
• 系统中的职责是如何分布的，容器间的是如何交互的
• 告诉开发者在哪里写代码

怎么画

用一个框图来表示，内部可能包括名称、技术选择、职责，以及这些框图之间的交互，如果涉及外部系统，最好明确边界

组件图(Component Diagram)

组件图是把某个容器进行展开，描述其内部的模块。

用途

这个图主要是给内部开发人员看的，怎么去做代码的组织和构建。其用途有

• 描述了系统由哪些组件/服务组成
• 厘清了组件之间的关系和依赖
• 为软件开发如何分解交付提供了框架

类图(Code/Class Diagram)

工具使用

本地协作

Mac:OmniGraffle

OmniGraffle 是由 The Omni Group 制作的一款绘图软件，它曾获得苹果设计奖。OmniGraffle 可以支持流程图、逻辑图或者网页产品模型设计等，功能非常强大。与 Graffle 对应的是在Windows平台广泛应用的 MS Visio（ Graffle 这个词据说就是为了和Visio区分而硬造出来的）

Windows:Visio

Office Visio 是office软件系列中的负责绘制流程图和示意图的软件，是一款便于IT和商务人员就复杂信息、系统和流程进行可视化处理、分析和交流的软件。使用具有专业外观的 Office Visio 图表，可以促进对系统和流程的了解，深入了解复杂信息并利用这些知识做出更好的业务决策。
Microsoft Office Visio帮助您创建具有专业外观的图表，以便理解、记录和分析信息、数据、系统和过程。

在线协作

Process On

是一款用HTML5、Canvas以及JavaScript技术开发而成的在线网页版作图工具。只需在工具栏拖放对应的图形到画布中,就可进行编辑，还支持流程图、思维导图、原型、拓扑图等，最让人心动的就是有实时协助的功能，从此再也不需要和同学、老师、领导之间来回传送文件，直接邀请，一起协作完成。

案例分享

下面是 城市运营态势 工具的一个架构图。作为一个应该自描述的架构图，这里不多做解释了。如果有看不明白的，那肯定是还画的不够好。

简介

官方文档:https://www.elastic.co/guide/en/elastic-stack/current/overview.html

搭建

本教程基于Elasticsearch版本7.7.0

要在docker中搭建，要起3个套件

其中elasticearch会遇到docker内存问题，可以看这个解决
https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html

或者
macOS with Docker for Mac

The vm.max_map_count setting must be set within the xhyve virtual machine:

From the command line, run:

1

screen ~/Library/Containers/com.docker.docker/Data/vms/0/tty

Press enter and usesysctl to configure vm.max_map_count:

1

sysctl -w vm.max_map_count=262144

To exit the screen session, type Ctrl a d.

1
2
3
4
5
6
7

docker network create elasticsearch
docker pull docker.elastic.co/elasticsearch/elasticsearch:7.7.0
docker run -d -p 9200:9200 -p 9300:9300 --network elasticsearch -e "discovery.type=single-node" --name elasticsearch docker.elastic.co/elasticsearch/elasticsearch:7.7.0
docker pull docker.elastic.co/kibana/kibana:7.7.0
docker run -d --link elasticsearch:elasticsearch --name kibana -p 5601:5601 --network elasticsearch docker.elastic.co/kibana/kibana:7.7.0
docker pull docker.elastic.co/logstash/logstash:7.7.0
docker run -d -p 5044:5044  --network elasticsearch --link elasticsearch:elasticsearch --name logstash docker.elastic.co/logstash/logstash:7.7.0

遇到错误

1

Security must be explicitly enabled when using a [basic] license. Enable security by setting [xpack.security.enabled] to [true] in the elasticsearch.yml file and restart the node.

编辑elasticsearch.yml，加入xpack.security.enabled:true,然后重启节点
重启时遇到错误:

1
2

ERROR: [1] bootstrap checks failed
 [1]: Transport SSL must be enabled if security is enabled on a [basic] license. Please set [xpack.security.transport.ssl.enabled] to [true] or disable security by setting [xpack.security.enabled] to [false]

编辑elasticsearch.yml，加入xpack.security.transport.ssl.enabled:true,然后重启节点

执行设置用户名和密码的命令,这里需要为4个用户分别设置密码，elastic, kibana, logstash_system,beats_system

1

bin/elasticsearch-setup-passwords interactive

配置文件修改

进入logstash容器里面修改配置文件

1
2

docker exec -it 54b504186a47 /bin/bash # 这里 54b504186a47是容器id
vi /usr/share/logstash/config/logstash.yml

logstash.yml配置文件如下

1
2
3
4

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]
path.config: /usr/share/logstash/config/*.conf
path.logs: /var/log/logstash

修改logstash-sample.conf

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

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    codec => json_lines
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "springboot-logstash-%{+YYYY.MM.dd}"
    #user => "elastic"
    #password => "changeme"
  }
}

这样logstash的读取就是通过一个tcp服务读取

springboot结合

引入包

1
2
3
4
5

<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>6.4</version>
</dependency>

添加配置文件logback-spring.xml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/base.xml" />

    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>127.0.0.1:5044</destination>
        <!-- 日志输出编码 -->
        <encoder charset="UTF-8"
                 class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>UTC</timeZone>
                </timestamp>
                <pattern>
                    <pattern>
                        {
                        "logLevel": "%level",
                        "serviceName": "${springAppName:-}",
                        "pid": "${PID:-}",
                        "thread": "%thread",
                        "class": "%logger{40}",
                        "rest": "%message"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="LOGSTASH" />
        <appender-ref ref="CONSOLE" />
    </root>

</configuration>

启动应用，配置kibana的索引，如图所示

如果添加了索引，页面没有显示索引，还要继续添加的话，这个是Kibana的问题，重启一下Kibana容器就好了。

概念

可以先看一本书《精通Web Analytics 2.0》

埋点是什么

所谓埋点就是在应用中特定的流程收集一些信息，用来跟踪应用使用的状况，后续用来进一步优化产品或是提供运营的数据支撑，包括访问数（Visits），访客数（Visitor），停留时长（Time On Site），页面浏览数（Page Views）和跳出率（Bounce Rate）。这样的信息收集可以大致分为两种：页面统计（track this virtual page view），统计操作行为（track this button by an event）。

埋点是谁的工作

现在公司通常都会有数据产品经理或业务线数据分析师，结合版本迭代过程进行埋点规划。如果是代码埋点，还需要开发完成相应的埋点代码。

埋点的意义

埋点就是为了对产品进行持续追踪，通过深度数据分析不断优化产品。好比去医院体检，医生测了你身体的各个健康指标，以此来判断你的健康状况。埋点的目的，其实就是随时或者定期监测你的产品的“健康”状况。

任何一个系统在设计初始阶段只关心核心业务的功能，等到系统上线以后，数据分析师对用户行为分析时会发现缺少很多数据，此时需要采用埋点的方法进行采集需要的数据。

业务人员主要通过自有或第三方的数据统计平台了解产品的概览性数据指标，包括新增用户数、活跃用户数等。这些指标能帮助企业宏观的了解用户访问的整体情况和趋势，从整体上把握产品的运营状况，但很难基于这些指标直接得到切实的产品改进策略。

而埋点将产品数据分析的深度下钻到流量分布和流动层面，通过对产品中的用户交互行为的统计分析，对宏观指标进行深入剖析，发现指标背后的问题，寻找人群的行为特点和关系，洞察用户行为与提升业务价值之间的潜在关联，了解组成特定数据现象的原因，并据此构建产品优化 迭代 和运营策略。

对于产品来说，用户在你的产品里做了什么、停留了多久、有什么异样，都是需要关注的。比如用户点击率怎么样？用户在核心使用路径上是否顺畅？有没有得到用户的认可？有没有因为设计按钮过多导致用户行为无效？用户希望有什么样的功能更新等等问题都可以通过埋点的方法实现。

埋点做好才能用来进行数据驱动产品和精细化运营。而埋点质量的好坏也直接影响到了产品运营的质量。因此它贯穿了产品的整个生命周期，为产品优化指明了方向。所以说，好的数据埋点，就成功了一半。

when && where

埋点是目的导向。

在产品规划时就要思考数据埋点问题，如果在产品外发后再考虑怎么埋点，就会导致前期版本用户的数据无法收集，想要看某个数据时就会非常无奈，只有等到新版本完善来弥补。

思考要埋哪些点、埋点的形式，需要紧密结合产品迭代的方向、运营需求，并和数据开发等进行充分沟通以确认：

• 埋点能够得到想要的数据解决/支持；
• 能够得到当前版本的复盘情况；
• 后续版本的数据支撑。

通常的沟通过程以 埋点文档为载体；数据埋点评审为终结。

当前版本的复盘情况：

• 新版本功能使用情况，是否符合预期；
• 新功能上线后对其他功能点的影响？是否为整体均有积极作用；
• 版本运营活动目标群体的特征获取;
• 新增商业化目标的监测……

后续版本的数据支撑：

• 规划方向的用户行为分析
• 画像特征分析

埋点指标：

基础指标

基础指标是一些常用的参数指标，比如用户行为相关的，有用户数、新增用户数、活跃用户数等；用户设备相关的，比如电脑系统、地区、语言、国家、产品版本等；用户属性相关的，性别、年龄等。这一部分一般很难通过埋点拿到，但是通过一般第三方工具可以看个大概。

用户行为数据指标

用户行为的行为数据，就是埋点的核心了。这一部分根据不同的产品，收集的数据也不同。再拿PC端的腾讯视频举例，每个用户每天使用腾讯视频的时长是多少？每次看的是哪些视频？用户最喜欢看哪个频道的视频？每个频道的使用情况是怎么样的？用户一般是在一天中的哪个时间段打开腾讯视频的。想要看的数据非常多，而这些，都是要提前规划好，有目的性得去决定要埋什么点、怎么埋。如果你的目的是为了研究一下用户对弹幕的接受程度，那围绕弹幕，设置一些指标数据，比如发布弹幕的次数和频率？看视频时是否开启了弹幕？这样的话，就能够通过数据，看出用户对弹幕的真实反映，对于提升产品功能和运营都有很大帮助。

核心质量指标

Crash等一些因为产品质量引起的问题，都是用户所不能忍受的。而且用户量越大，越容易出现产品质量问题。尤其是一些产品，他的产品属性决定了，产品质量是吸引用户购买的一个很重要的因素。比如迅雷、photoshop等软件，核心质量的指标能帮助我们监控产品的“健康”情况。

这里说的质量指标，不单单指产品的 异常退出率、Crash率， 还包括和产品自身业务相关的指标。比如，对于视频编辑产品， 编辑成功率 是很重要的指标；对于迅雷， 下载成功率 是核心指标；对于在线类产品， 资源解析 成功率 是核心指标……

访问与访客

访问次数（Visits）与访问人数（Vistors）是几乎所有应用都需要统计的指标，这也是最基础的指标。对于应用的统计来说，经常看到的DAU（日活跃用户数量），MAU（用户数量统计），UV（独立（IP）访客）等指标都是指统计访客（Vistors）。访问（Visits）是指会话层，用户打开应用花一段时间浏览又离开，从指标定义（访问次数）来说这被称之为统计会话（Session）数。一次会话（Session 或 Visit）是打开应用的第一个请求（打开应用）和最后一个请求决定的。如果用户打开应用然后放下手机或是离开电脑，并在接下来30分钟内没有任何动作，此次会话自动结束，通常也算作一次访问或会话期（30分钟是早起网页版应用约定俗成的会话数定义，目前用户停留在应用的时长变长，30分钟的限定也可能随之不同，总之是能代表一次用户访问的时长）。在计算访问人数（Vistors）时，埋点上报的数据是尽可能接近真实访客的人数。对于有需要统计独立访客这个指标的场景，这里还是需要强调一下，访问人数（Vistors）并不是真实独立的人，因此收集数据时必须知道访问人数虽然能够很好的反映使用应用的真实访问者的数量，但不等于使用应用的真实人数。（原因是，重复安装的应用，或是手机参数被修改都会使得独立访客的指标收到影响。计算访问人数的埋点都是依赖Cookie，用户打开应用，应用都会在此人的终端创建一个独立Cookie, Cookie会被保留，但还是难免会被用户手动清理或是Cookie被禁用导致同一用户使用应用Cookie不一致，所以独立访客只能高度接近于使用应用的真实人数。）

停留时长

停留时长用来衡量用户在应用的某一个页面或是一次访问（会话）所停留的时间。页面停留时长，表示在每个页面所花费的时间；例如：首页就是进入首页（10：00）到离开首页进入下一个页面(10:01)的时长，首页停留时长计算为1分钟。页面A是2分钟。停留时长的数据并不都是一定采集得到的，比如页面B进入时间（10：03），离开出现异常或是退出时间没有记录，这时候计算就是0 （所以指标计算时需要了解埋点的状况，剔除这样的无效数据）。应用的停留时长，表示一次访问（会话）所停留的时间，计算起来就是所有页面的访问时长，同样是上一个流程，应用的停留时长就是4分钟。

跳出率

跳出率的计算方法现在在各个公司还是很多种，最经常被使用的是：用户只访问了一个页面所占的会话比例（原因是：假设这种场景，用户来了访问了一个页面就离开了，想想用户使用的心里画面应该是：打开应用，心想什么鬼，然后关闭应用甚至卸载了。这个场景多可怕，这也是为什么跳出率指标被如此关注）跳出率可以分解到两个层次：一是整个应用的跳出率，二是重点的着陆页的跳出率，甚至是搜索关键词的跳出率。跳出率的指标可操作性非常强，通过统计跳出率可以直接发现页面的问题发现关键词的问题。

退出率

退出率是针对页面的，这个指标的目标很简单，就是在针对某个页面有多少用户离开了应用，主要用户反映用户从应用离开的情况。哪些页面需要被改进最快的方式被发掘。（注意：退出率高不一定是坏事。例如：预测流程的最终节点的退出率就应该是高的）

转化率

我们在产品上投入这么多，不就是为了衡量产出么？所以对于电商类应用，还有比转化率更值得关注的指标吗？转化率的计算方法是某种产出除以独立访客或是访问量，对于电商产品来说，就是提交订单用户数除以独立访客。转化率的计算看起来想到那简单，但却是埋点中最贴近业务的数据收集。这也是最体现埋点技巧的指标，需要结合业务特点制定计算方法。提交订单量/访客数是最基本的转化率，转化率还可以分层次，指定用户路径的，如：完成某条路径的提交订单数/访客数。

参与度

参与度并不是一个指标，而是一系列的指标的统称，例如访问深度，访问频次，针对电商的下单次数，针对内容服务商的播放次数，及用户行为序列这些都可以是衡量参与度的指标。之所以把参与度列为一个指标，是希望大家明白把指标结合业务，产生化学反应，活学活用去发现事物的本质。

数据采集普遍遇到的几个问题

• 实时性，对于工具性产品在无网条件下的数据，无法实时上报；
• 完整性，由于用户隐私协议&欧盟通用数据保护条例的，部分数据无法采集；
• 异常，android_id、idfa、idfv 随版本升级变化或无法获取。

怎么埋点

代码埋点

以为需要监测网站上/app上用户的行为，是需要在网页/app中加上一些代码的，当用户触发相应行为时，进行数据上报，也就是代码埋点。这样的代码，在网站上叫监测代码，在app中叫SDK（Software Development Kit）。市场上的第三方数据采集均支持代码埋点，GA, GrowingIO，神策等。

• 优点
  • 采集的数据比较具有针对性，更加适合精细化数据分析。
  • 同时也能提高数据的准确性。
• 缺点
  • 每一个控件的埋点都需要添加相应的代码，不仅工作量大，而且限定了必须是技术开发人员才能完成。
  • 每一次产品迭代，都需要更新埋点方案。
• 适用场景
  • 有具体的业务分析需求，且按照各个事件埋点的方式不能满足。
  • 需要对埋点事件进行传参等自定义属性设置。代码埋点虽然较复杂，但功能最完善，覆盖了埋点中的不同业务需求。

可视化埋点

利用可视化交互手段，数据产品/数据分析师可以通过可视化界面（管理后台连接设备） 配置事件，如下是腾讯移动分析的可视化埋点界面。可视化埋点仍需要先配置相关事件，再采集。

例如腾讯的https://mta.qq.com/

蚂蚁金服的移动开发平台
https://tech.antfin.com/docs/2/49549

• 优点
  • 业务人员可用，无需技术人员进行SDK嵌入，不懂代码的产品运营人员也可通过后台可视化界面配置和统计埋点并实时下发到客户端生效。
  • 无需版本更新，由于不需要嵌入新SDK，不需要发布新版本，可谓即时生效。
  • 对所有版本生效：新增埋点在所有版本生效，不存在迭代问题。
• 缺点
  • 可覆盖的功能有限，目前并不是所有控件操作都可以通过这种方案进行定制。
  • 不能自定义交互事件属性，由于获取的是交互事件元素的DOMpath，无法对具体事件设置参数。
  • 不支持可以不断加载的内容瀑布流交互。
• 适用场景
  • 分析或统计需求简单，不需要对埋点事件进行传参等自定义属性设置。
  • 频繁上线或更新的H5类型的运营活动

无埋点

无埋点是指开发人员集成采集 SDK 后，SDK 便直接开始捕捉和监测用户在应用里的所有行为，并全部上报，不需要开发人员添加额外代码。

数据分析师/数据产品 通过管理后台的圈选功能来选出自己关注的用户行为，并给出事件命名。之后就可以结合时间属性、用户属性、事件进行分析了。所以无埋点并不是真的不用埋点了。目前市场第三方工具GrowingIO支持无埋点全量行为数据抓取https://growingio.jinshuju.com/

• 优点
  • 因为无埋点对页面所有元素进行埋点，那么这个页面每个元素被点击的概率你也就知道，对点击概率比较大的元素可以进行深入分析。
  • 可以在系统上线后使用，支持基于全量的数据回溯，因为无埋点在你部署SDK的时候数据就一直在收集，可帮助进行启发式、探索式的数据分析。
  • 它技术门槛低，部署简单。
• 缺点
  • 无埋点无法采集自定义属性，只使用通用大部门，通用的场景。
  • 数据形式非业务导向，因为是对所有事件数据的自动收集，没有按照业务需求进行事件或区域设置，业务或数据人员在使用时或许不能直接使用，需要二次计算或处理。
  • 兼容性不好，传输时效性较差等问题，因为是对所有的元素数据都收集，会给数据传输和服务器带来较大的压力。
• 适用场景
  • 分析或统计需求简单，不需要对埋点事件进行传参等自定义属性设置的事件。
  • 针对快速、频繁上线和迭代的H5类型的运营活动的评估。

总体来说，无埋点和可视化埋点更侧重结果的展现，对过程追溯少，更适合产品经理分析基础的产品功能流畅度、用户体验、产品路径设置等。代码埋点和后端埋点，不仅能展现结果，也会记录用户行为过程，支持深度的行为分析和偏好洞察，还可将行为数据与业务数据打通，适合产品和运营人员深度使用。

无论采用哪种埋点方式，都应该根据业务场景和产品阶段，梳理和构建数据分析体系。埋点规划混乱、数据采集无序、数据分析断层，最终将会让企业陷入“有数据而无价值”的境地。

架构设计

通过 日志埋点 来实现业务监控和行为分析主要需要以下4个步骤

数据生成(埋点)
数据收集
数据解析(结构化)
数据落盘
数据使用(展示/分析)

数据使用

如何做数据埋点

其实这个问题不应该问别人，应该问你自己，通过上文，我们已经知道了，如果你想绘制基础的人群画像你就需要获取用户机型、网络类型、操作系统，IP地域等数据；如果你想分析每一个注册转化率，你就需要获取每一个步骤的点击次数，然后制作成漏斗，看那一步转化率出现了问题……目的不一样，获取的数据也不一样，使用的埋点技术也不一样。

那么，我们该如何选择埋点方式呢？

我们的目的是实现深度数据分析，不应该采用与其他企业通用的埋点方法，应该采用适合自己的埋点方法。也就是做到“因系统而异、避免千系统一面的情况。

在系统刚上线的初期阶段，我们可以采用无埋点的方式。因为我们通过UV、PV、点击率等基本指标及即可满足数据分析需求。如果产品上线时间很长，我们需要进行深度数据分析则选择代码埋点。它可以帮我们收集需要的属性。另外，如何埋点既可以在前端实现，也可以在后端实现，我们推荐在后端实现。因为后端数据可以保证数据的准确性。最后，如果您为了方便快捷并且免费，可以选择第三方统计工具，但是一定要选择适合自己业务的统计工具。

从业务过程中采集埋点，是数据驱动型公司的必要条件。一般公司的产品功能评审环节，不仅有 PRD (Product requirement document），还加入了对应的 DRD ( Data requirement document）。对于埋点而言，DRD 需要明确业务目标与埋点缺口之间的关系以及需求的优先级。埋点的需求大多来自于 DRD，整个过程会涉及多个角色，主要包括产品经理、业务数据负责人、开发工程师、测试工程师，如图所示。

总之，如果您需要深度分析，选择后端（手动）埋点和无埋点组合的方案；如果您只是想看宏观数据，可以选择无埋点。无论采用哪种埋点方法，一定要慎重，根据需要来设置，最好不要出现错埋或者漏埋的情况。最后，数据分析师一定要和业务工程团队（部署实施埋点的部门）配合好才能实现完美的数据采集方案，有时候沟通比选择埋点方式更重。

优秀的实践案例

有赞埋点实践https://tech.youzan.com/track-1/

美团点评前端无痕埋点实践https://tech.meituan.com/2017/03/02/mt-mobile-analytics-practice.html

产品经理该如何做好数据埋点https://www.uisdc.com/product-manager-makes-data-event-tracking

数据统计埋点工作框架及细节规范https://www.inneed.club/articles/detail/pkx0epxgew

Activiti7.X结合SpringBoot2.1、Mybatis

Activiti简介

Activiti介绍

Activiti 是由 jBPM 的创建者 Tom Baeyens 离开 JBoss 之后建立的项目，构建在开发 jBPM 版本 1 到 4 时积累的多年经验的基础之上，旨在创建下一代的 BPM 解决方案。

Activiti是一个开源的工作流引擎，它实现了BPMN 2.0规范，可以发布设计好的流程定义，并通过api进行流程调度。

Activiti 作为一个遵从 Apache 许可的工作流和业务流程管理开源平台，其核心是基于Java的超快速、超稳定的 BPMN2.0 流程引擎，强调流程服务的可嵌入性和可扩展性，同时更加强调面向业务人员。

Activiti 流程引擎重点关注在系统开发的易用性和轻量性上。每一项 BPM 业务功能 Activiti 流程引擎都以服务的形式提供给开发人员。通过使用这些服务，开发人员能够构建出功能丰富、轻便且高效的 BPM 应用程序。

Activiti是一个针对企业用户、开发人员、系统管理员的轻量级工作流业务管理平台，其核心是使用Java开发的快速、稳定的BPMN e 2.0流程引擎。Activiti是在ApacheV2许可下发布的，可以运行在任何类型的Java程序中，例如服务器、集群、云服务等。Activiti可以完美地与Spring集成。同时，基于简约思想的设计使Activiti非常轻量级。

目前Activiti有2个版本，一个本地的core，一个可以支持分布式的cloud，本文只介绍core，新版 Activiti 7.0.0 发布后，Activiti Cloud 现在是新一代商业自动化平台，提供一组旨在在分布式基础架构上运行的 Cloud原生构建块。Cloud可以参考官网

Activiti 7.x 主要突出了 Spring Boot 2.x 应用程序中的 ProcessRuntime 和 TaskRuntime API 的使用。

BPMN

BPMN（Business Process Model And Notation）-业务流程模型和符号是由BPMI（Business Process Management Initiative）开发的一套标准的业务流程建模符号，使用BPMN提供的符号可以创建业务流程。

Activiti 就是使用BPMN 2.0 进行流程建模、流程执行管理，它包括很多的建模符号，比如：Event 用一个圆圈表示，它是流程中运行过程中发生的事情。

一个bpmn图形的例子：

首先当事人发起一个请假单
其次他所在部门的经理对请假单进行审核
然后人事经理进行复核并进行备案
最后请假流程结束

创建应用

代码库地址：https://github.com/dinghuang/activiti-service.git

创建maven应用，pom引入包

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

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <artifactId>dinghuang-framework-parent</artifactId>
        <groupId>org.dinghuang</groupId>
        <version>0.1.0-RELEASE</version>
    </parent>

    <artifactId>activiti</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>jar</packaging>
    <name>activiti</name>
    <description>activiti project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <mapstruct>1.3.0.Final</mapstruct>
        <activiti.version>7.1.0.M6</activiti.version>
        <spring-boot>2.1.0.RELEASE</spring-boot>
        <fastjson>1.2.47</fastjson>
        <commons-collections>3.2.2</commons-collections>
        <commons-lang3>3.8.1</commons-lang3>
        <swagger-annotations>1.5.16</swagger-annotations>
        <springfox-swagger2>2.7.0</springfox-swagger2>
        <druid>1.1.10</druid>
        <mybatis-plus-boot-starter>3.1.0</mybatis-plus-boot-starter>
        <feign-hystrix>9.5.0</feign-hystrix>
        <lombok>1.16.20</lombok>
        <mysql-connector-java>8.0.15</mysql-connector-java>
        <liquibase-core>3.5.3</liquibase-core>
        <validation-api>2.0.1.Final</validation-api>
        <hibernate-validator>6.0.15.Final</hibernate-validator>
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.activiti.dependencies</groupId>
                <artifactId>activiti-dependencies</artifactId>
                <version>${activiti.version}</version>
                <scope>import</scope>
                <type>pom</type>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>5.1.12.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.activiti</groupId>
            <artifactId>activiti-spring-boot-starter</artifactId>
            <version>${activiti.version}</version>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <version>42.2.10</version>
        </dependency>
        <!-- Activiti生成流程图 -->
        <dependency>
            <groupId>org.activiti</groupId>
            <artifactId>activiti-image-generator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
            <version>${spring-boot}</version>
        </dependency>
        <dependency>
            <groupId>org.dinghuang</groupId>
            <artifactId>dinghuang-framework-core</artifactId>
            <version>0.1.0-RELEASE</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <annotationProcessorPaths>
                        <!--这个配置是因为lombok跟mapstruct一起用的时候maven对注解的解析器选择有点问题-->
                        <path>
                            <groupId>org.mapstruct</groupId>
                            <artifactId>mapstruct-processor</artifactId>
                            <version>${mapstruct}</version>
                        </path>
                        <path>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                            <version>1.16.20</version>
                        </path>
                    </annotationProcessorPaths>
                    <compilerArgs>
                        <compilerArg>
                            -Amapstruct.defaultComponentModel=spring
                        </compilerArg>
                    </compilerArgs>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
        <finalName>demo</finalName>
    </build>

</project>

与mysql结合

在resource目录下准备文件activiti.cfg.xml，内容如下：

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

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
                           http://www.springframework.org/schema/beans/spring-beans.xsd">
    <bean id="processEngineConfiguration" class="org.activiti.engine.impl.cfg.StandaloneProcessEngineConfiguration">
        <property name="databaseType" value="mysql"></property>
        <property name="jdbcUrl" value="jdbc:mysql://localhost:3306/activiti"></property>
        <property name="jdbcDriver" value="com.mysql.jdbc.Driver"></property>
        <property name="jdbcUsername" value="root"></property>
        <property name="jdbcPassword" value="root"></property>
    </bean>
</beans>

执行下面的代码，初始化db数据库或者配置springboot配置文件，会自动生成25个表，代码执行如下：

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

package org.dinghuang.activiti.conf;

import org.activiti.engine.impl.db.DbSchemaCreate;

/**
 * @author dinghuang123@gmail.com
 * @since 2020/2/28
 */
public class Test {

    public static void main(String[] args) {
        DbSchemaCreate.main(args);
    }
}

配置文件如下：

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

spring:
 activiti:
  # 自动建表
  database-schema: ACTIVITI
  #表示启动时检查数据库表，不存在则创建
  database-schema-update: true
  #表示哪种情况下使用历史表，这里配置为full表示全部记录历史，方便绘制流程图
  history-level: full
  #表示使用历史表，如果不配置，则工程启动后可以检查数据库，只建立了17张表，历史表没有建立，则流程图及运行节点无法展示
  db-history-used: true
project:
  manifest:
    file:
      path: classpath:/default-project.json
logging:
  level:
    org.activiti: debug

数据库说明

数据库会生成25张表，ER图如图所示：

表的列表：

1
2
3
4
5
6

ACT_RE_*: RE表示repository，这个前缀的表包含了流程定义和流程静态资源
ACT_RU_*: RU表示runtime，这些运行时的表，包含流程实例，任务，变量，异步任务等运行中的数据。Activiti只在流程实例执行过程中保存这些数据， 在流程结束时就会删除这些记录。
ACT_ID_*: ID表示identity，这些表包含身份信息，比如用户，组等。这些表现在已废弃。
ACT_HI_*: HI表示history，这些表包含历史数据，比如历史流程实例， 变量，任务等。
ACT_GE_*: 通用数据， 用于不同场景下。
ACT_EVT_*: EVT表示EVENT，目前只有一张表ACT_EVT_LOG，存储事件处理日志，方便管理员跟踪处理

具体表详解可以参考文章

表详解

工作流使用

activiti7内置了Spring security框架,官方demo跟spring结合的必须与spring-security结合，这里我不用spring-security，因为现在没有用户表了，所以自定义一些用户角色表去结合，更容易理解。

关于security问题
activiti7最新的类似Runtime API和Task API都集成了security。
如果使用上述的API,那么必须要使用security，不能屏蔽security，否则会报错。
使用引擎服务类的时候，可以排除security，因为这些是最原始的API。但是activiti7官方已经明确说了，随时可能会干掉这些API。不建议开发人员直接使用引擎类以及引擎配置了、服务类等。

相关配置类

自定义id策略

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

package org.dinghuang.activiti.conf;

import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import org.activiti.engine.impl.cfg.IdGenerator;
import org.springframework.stereotype.Component;

/**
 * 自定义id策略
 *
 * @author dinghuang123@gmail.com
 * @since 2020/3/4
 */
@Component
public class ActivitiIdGeneratorConfiguration implements IdGenerator {

    @Override
    public String getNextId() {
        return String.valueOf(IdWorker.getId());
    }
}

自定义用户组

activiti7已经抛弃了identity相关的表，同时与springSecurity结合，所以这边如果想自定义可以这么改，但是实际用途不大，最重要的是要结合自己的用户表设计来。

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

package org.dinghuang.activiti.conf;

import org.activiti.api.runtime.shared.identity.UserGroupManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 重写用户权限
 *
 * @author dinghuang123@gmail.com
 * @since 2020/3/2
 */
@Component
public class ActivitiUserGroupManagerConfiguration implements UserGroupManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(ActivitiUserGroupManagerConfiguration.class);

    public static List<String> roles = new ArrayList<>(3);
    public static List<String> groups = new ArrayList<>(1);
    public static List<String> users = new ArrayList<>(3);
    public static Map<String, String> userRoleMap = new HashMap<>(3);

    static {
        roles.add("workCreate");
        roles.add("workPermit");
        roles.add("workLeader");

        groups.add("workGroupA");

        users.add("admin");
        users.add("laowang");
        users.add("xiaofang");

        userRoleMap.put("admin", "workCreate");
        userRoleMap.put("laowang", "workPermit");
        userRoleMap.put("xiaofang", "workLeader");
    }

    @Override
    public List<String> getUserGroups(String s) {
        LOGGER.info("get user groups");
        return groups;
    }

    @Override
    public List<String> getUserRoles(String s) {
        String role = userRoleMap.get(s);
        List<String> list = new ArrayList<>();
        list.add(role);
        LOGGER.info("get user roles");
        return list;
    }

    @Override
    public List<String> getGroups() {
        LOGGER.info("get groups");
        return groups;
    }

    @Override
    public List<String> getUsers() {
        LOGGER.info("get users");
        return users;
    }
}

其他自定义配置

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

package org.dinghuang.activiti.conf;

import org.activiti.api.runtime.shared.identity.UserGroupManager;
import org.activiti.core.common.spring.project.ProjectModelService;
import org.activiti.engine.impl.history.HistoryLevel;
import org.activiti.spring.SpringAsyncExecutor;
import org.activiti.spring.SpringProcessEngineConfiguration;
import org.dinghuang.activiti.controller.ActivitiController;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.io.IOException;

/**
 * @author dinghuang123@gmail.com
 * @since 2020/3/2
 */
@Configuration
public class ActivitiSpringIdentityAutoConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(ActivitiController.class);

    private static final int CORE_POOL_SIZE = 10;
    private static final int MAX_POOL_SIZE = 30;
    private static final int KEEP_ALIVE_SECONDS = 300;
    private static final int QUEUE_CAPACITY = 300;

    @Autowired
    private DataSource dataSource;

    @Autowired
    private PlatformTransactionManager platformTransactionManager;

    @Autowired
    private UserGroupManager userGroupManager;

    @Autowired
    private ActivitiIdGeneratorConfiguration activitiIdGeneratorConfiguration;

    @Autowired
    private ProjectModelService projectModelService;

    /**
     * 处理引擎配置
     */
    @Bean
    public SpringProcessEngineConfiguration springProcessEngineConfiguration() {
        SpringProcessEngineConfiguration configuration = new SpringProcessEngineConfiguration(projectModelService);
        configuration.setDataSource(this.dataSource);
        configuration.setTransactionManager(this.platformTransactionManager);
        SpringAsyncExecutor asyncExecutor = new SpringAsyncExecutor();
        asyncExecutor.setTaskExecutor(workFlowAsync());
        configuration.setAsyncExecutor(asyncExecutor);
        configuration.setDatabaseSchemaUpdate("true");
        configuration.setUserGroupManager(this.userGroupManager);
        configuration.setHistoryLevel(HistoryLevel.FULL);
        configuration.setDbHistoryUsed(true);
        configuration.setIdGenerator(this.activitiIdGeneratorConfiguration);
        Resource[] resources = null;
        // 启动自动部署流程
        try {
            resources = new PathMatchingResourcePatternResolver().getResources("classpath*:bpmn/*.bpmn");
        } catch (IOException e) {
            LOGGER.error("Start the automated deployment process error", e);
        }
        configuration.setDeploymentResources(resources);
        return configuration;
    }

    /**
     * 线程池
     */
    @Primary
    @Bean("workFlowTaskExecutor")
    public ThreadPoolTaskExecutor workFlowAsync() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setThreadNamePrefix("workFlowTaskExecutor-");
        executor.initialize();
        return executor;
    }

}

因为去掉了springSecurity,所以启动类得排除springSecurity的自动配置。

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

package org.dinghuang.activiti;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.servlet.config.annotation.EnableWebMvc;
import org.springframework.web.servlet.view.InternalResourceViewResolver;

@SpringBootApplication(exclude = {
        org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration.class,
        org.springframework.boot.actuate.autoconfigure.security.servlet.ManagementWebSecurityAutoConfiguration.class,
        org.activiti.core.common.spring.identity.config.ActivitiSpringIdentityAutoConfiguration.class
})
@EnableWebMvc
public class ActivitiApplication {

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

    @Bean
    public InternalResourceViewResolver setupViewResolver() {
        InternalResourceViewResolver resolver = new InternalResourceViewResolver();
        /** 设置视图路径的前缀 */
        resolver.setPrefix("resources/templates");
        /** 设置视图路径的后缀 */
        resolver.setSuffix(".html");
        return resolver;
    }
}

流程绘制

流程绘制这里提供2种，一种用IDEA下载ActiBPM插件去画，这里给一段我自己画的xml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:activiti="http://activiti.org/bpmn" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:tns="http://sourceforge.net/bpmn/definitions/a123123" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:yaoqiang="http://bpmn.sourceforge.net" expressionLanguage="http://www.w3.org/1999/XPath" id="m1583310491794" name="" targetNamespace="http://sourceforge.net/bpmn/definitions/a123123" typeLanguage="http://www.w3.org/2001/XMLSchema">
  <process id="dinghuangTest" isClosed="false" isExecutable="true" name="请假流程" processType="None">
    <startEvent id="_2" name="开始"/>
    <userTask activiti:assignee="bilu" activiti:candidateGroups="manager" activiti:exclusive="true" id="_3" name="部门经理审批"/>
    <endEvent id="_4" name="结束"/>
    <sequenceFlow id="_5" sourceRef="_2" targetRef="_3"/>
    <sequenceFlow id="_6" name="事情不重要" sourceRef="_3" targetRef="_4">
      <conditionExpression xsi:type="tFormalExpression"><![CDATA[${!important}]]></conditionExpression>
    </sequenceFlow>
    <userTask activiti:candidateGroups="admin" activiti:exclusive="true" id="_7" name="总经理审批"/>
    <sequenceFlow id="_8" name="事情重要" sourceRef="_3" targetRef="_7">
      <conditionExpression xsi:type="tFormalExpression"><![CDATA[${important}]]></conditionExpression>
    </sequenceFlow>
    <sequenceFlow id="_9" name="审批通过" sourceRef="_7" targetRef="_4"/>
  </process>
  <bpmndi:BPMNDiagram documentation="background=#3C3F41;count=1;horizontalcount=1;orientation=0;width=842.4;height=1195.2;imageableWidth=832.4;imageableHeight=1185.2;imageableX=5.0;imageableY=5.0" id="Diagram-_1" name="New Diagram">
    <bpmndi:BPMNPlane bpmnElement="dinghuangTest">
      <bpmndi:BPMNShape bpmnElement="_2" id="Shape-_2">
        <dc:Bounds height="32.0" width="32.0" x="100.0" y="135.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="32.0" width="32.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="_3" id="Shape-_3">
        <dc:Bounds height="55.0" width="85.0" x="235.0" y="140.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="55.0" width="85.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="_4" id="Shape-_4">
        <dc:Bounds height="32.0" width="32.0" x="565.0" y="160.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="32.0" width="32.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape bpmnElement="_7" id="Shape-_7">
        <dc:Bounds height="55.0" width="85.0" x="415.0" y="55.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="55.0" width="85.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNEdge bpmnElement="_5" id="BPMNEdge__5" sourceElement="_2" targetElement="_3">
        <di:waypoint x="132.0" y="151.0"/>
        <di:waypoint x="235.0" y="167.5"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="0.0" width="0.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="_6" id="BPMNEdge__6" sourceElement="_3" targetElement="_4">
        <di:waypoint x="320.0" y="167.5"/>
        <di:waypoint x="565.0" y="176.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="0.0" width="0.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="_8" id="BPMNEdge__8" sourceElement="_3" targetElement="_7">
        <di:waypoint x="320.0" y="167.5"/>
        <di:waypoint x="415.0" y="82.5"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="0.0" width="0.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge bpmnElement="_9" id="BPMNEdge__9" sourceElement="_7" targetElement="_4">
        <di:waypoint x="500.0" y="82.5"/>
        <di:waypoint x="565.0" y="176.0"/>
        <bpmndi:BPMNLabel>
          <dc:Bounds height="0.0" width="0.0" x="0.0" y="0.0"/>
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
    </bpmndi:BPMNPlane>
  </bpmndi:BPMNDiagram>
</definitions>

这是一个很简单的流程，如图所示:

前端实现自定义流程、输出流程图

官方提供bpmn.js可以实现前端拖动画图，github地址：https://github.com/bpmn-io/bpmn-js

可以把这个加到项目中，这里就不解释了。

前端实现输出流程图效果如下，流程节点会高亮显示：

这里是我写的一个小demo，代码可以查看我的github代码库。项目启动后，访问http://localhost:8080/v1/activiti/index

接口调试

服务启动后访问地址：http://localhost:8080/swagger-ui.html#/
接口已经写在代码中，如果需要调试的话，顺序是

• 启动实例流程
• 根据用户名称查询任务列表（用户名bilu，拿到任务id后去完成任务）
• 审批approve
• 回退 back
  

实现随意跳转和回退撤回功能

因为activiti7是以图的形式来操作的，所以这边就要考虑连线的情况。

本文写了并行和串行的撤回以及撤回功能的连线的图，最终效果图如图所示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426

/**
    * 输出图像
    *
    * @param response               响应实体
    * @param bpmnModel              图像对象
    * @param flowIds                已执行的线集合
    * @param executedActivityIdList void 已执行的节点ID集合
    */
   public void outputImg(HttpServletResponse response, BpmnModel bpmnModel, List<String> flowIds, List<String> executedActivityIdList) {
       InputStream imageStream = null;
       try {
           imageStream = new DefaultProcessDiagramGenerator().generateDiagram(bpmnModel, executedActivityIdList, flowIds, "宋体", "微软雅黑", "黑体", true, "png");
           // 输出资源内容到相应对象
           byte[] b = new byte[1024];
           int len;
           while ((len = imageStream.read(b, 0, 1024)) != -1) {
               response.getOutputStream().write(b, 0, len);
           }
           response.getOutputStream().flush();
       } catch (Exception e) {
           LOGGER.error("out put process img error！", e);
       } finally { // 流关闭
           try {
               if (imageStream != null) {
                   imageStream.close();
               }
           } catch (IOException e) {
               LOGGER.error("IoException", e);
           }
       }
   }

   public void showImg(String instanceKey, HttpServletResponse response) {
       if (instanceKey == null) {
           LOGGER.error("process instance not exist");
           return;
       }
       //获取流程实例
       HistoricProcessInstance processInstance = historyService.createHistoricProcessInstanceQuery().processInstanceId(instanceKey).singleResult();
       if (processInstance == null) {
           LOGGER.error("process instance {} not exist", instanceKey);
           return;
       }

       // 根据流程对象获取流程对象模型
       BpmnModel bpmnModel = repositoryService.getBpmnModel(processInstance.getProcessDefinitionId());

       //查看已执行的节点集合, 获取流程历史中已执行节点，并按照节点在流程中执行先后顺序排序
       // 构造历史流程查询
       HistoricActivityInstanceQuery historyInstanceQuery = historyService.createHistoricActivityInstanceQuery().processInstanceId(instanceKey);
       // 查询历史节点,根据id排序
       List<HistoricActivityInstance> historicActivityInstanceList = historyInstanceQuery.orderBy(HistoricActivityInstanceQueryProperty.HISTORIC_ACTIVITY_INSTANCE_ID).asc().list();
       if (historicActivityInstanceList == null || historicActivityInstanceList.size() == 0) {
           LOGGER.info("process instance history node info not exist", instanceKey);
           outputImg(response, bpmnModel, null, null);
           return;
       }
       Map<String, FlowNode> flowNodeMap = getFlowNodeMap(historicActivityInstanceList, processInstance.getProcessDefinitionId());
       //处理撤回这种情况 根据id排序
       List<Task> tasks = taskService.createTaskQuery().processInstanceId(instanceKey).orderBy(TaskQueryProperty.TASK_ID).asc().list();
       Set<String> executedActivityIdList = new LinkedHashSet<>();
       List<String> taskIdList = tasks == null ? null : tasks.stream().map(TaskInfo::getId).collect(Collectors.toList());
       List<String> taskKeyList = tasks == null ? null : tasks.stream().map(TaskInfo::getTaskDefinitionKey).collect(Collectors.toList());
       if (tasks != null) {
           //串行
           if (tasks.size() == 1) {
               for (int i = 0; i < historicActivityInstanceList.size(); i++) {
                   if (historicActivityInstanceList.get(i).getTaskId() == null || !historicActivityInstanceList.get(i).getActivityId().equals(tasks.get(0).getTaskDefinitionKey())) {
                       executedActivityIdList.add(historicActivityInstanceList.get(i).getActivityId());
                   } else {
                       executedActivityIdList.add(historicActivityInstanceList.get(i).getActivityId());
                       break;
                   }
               }
           } else {
               List<HistoricVariableInstance> historicVariableInstances = historyService.createHistoricVariableInstanceQuery()
                       .processInstanceId(processInstance.getId()).list();
               Map<String, HistoricVariableInstance> historicVariableInstanceMap = historicVariableInstances.stream()
                       .collect(Collectors.toMap(HistoricVariableInstance::getVariableName,
                               historicVariableInstance -> historicVariableInstance, BinaryOperator.maxBy(Comparator.comparing(HistoricVariableInstance::getId))));
               //并行
               Collection<FlowElement> flowElementCollection = bpmnModel.getMainProcess().getFlowElements();
               Map<String, List<String>> parentMap = new HashMap<>(tasks.size());
               for (FlowElement flowElement : flowElementCollection) {
                   List<String> parentCodeList = new LinkedList<>();
                   if (flowNodeMap.get(flowElement.getId()) != null) {
                       List<SequenceFlow> sequenceFlows = flowNodeMap.get(flowElement.getId()).getIncomingFlows();
                       if (sequenceFlows != null && !sequenceFlows.isEmpty()) {
                           for (SequenceFlow sequenceFlow : sequenceFlows) {
                               parentCodeList.add(sequenceFlow.getSourceRef());
                           }
                           parentMap.put(flowElement.getId(), parentCodeList);
                       }
                   }
               }
               Set<String> sameParentTaskCode = new LinkedHashSet<>();
               for (Task task : tasks) {
                   //找到所有任务拥有相同父级的集合任务
                   for (String taskKey : parentMap.get(task.getTaskDefinitionKey())) {
                       for (String key : parentMap.keySet()) {
                           if (parentMap.get(key).contains(taskKey)) {
                               sameParentTaskCode.add(key);
                               break;
                           }
                       }
                   }
               }
               //说明是并行，但是做完的任务
               for (String sameParentTask : sameParentTaskCode) {
                   if (!taskKeyList.contains(sameParentTask)) {
                       List<SequenceFlow> sequenceFlows = flowNodeMap.get(sameParentTask).getOutgoingFlows();
                       if (sequenceFlows != null && !sequenceFlows.isEmpty()) {
                           for (SequenceFlow sequenceFlow : sequenceFlows) {
                               if (querySequenceFlowCondition(sequenceFlow, historicVariableInstanceMap)) {
                                   executedActivityIdList.add(sequenceFlow.getTargetRef());
                               }
                           }
                       }
                   }
               }

               for (String taskKey : sameParentTaskCode) {
                   for (int i = 0; i < historicActivityInstanceList.size(); i++) {
                       if (historicActivityInstanceList.get(i).getTaskId() == null || !historicActivityInstanceList.get(i).getActivityId().equals(taskKey)) {
                           executedActivityIdList.add(historicActivityInstanceList.get(i).getActivityId());
                       } else {
                           executedActivityIdList.add(historicActivityInstanceList.get(i).getActivityId());
                           break;
                       }
                   }

               }
           }
       }
       //获取流程走过的线
       List<String> executedActivityIdListResult = new ArrayList<>(executedActivityIdList);
       List<String> flowIds = getHighLightedFlows(bpmnModel, historicActivityInstanceList, executedActivityIdListResult, taskIdList);
       //输出图像，并设置高亮
       outputImg(response, bpmnModel, flowIds, executedActivityIdListResult);
   }

   private Map<String, FlowNode> getFlowNodeMap(List<HistoricActivityInstance> historicActivityInstanceList, String processDefinitionId) {
       org.activiti.bpmn.model.Process process = repositoryService
               .getBpmnModel(processDefinitionId)
               .getMainProcess();
       Map<String, FlowNode> flowNodeMap = new HashMap<>(historicActivityInstanceList.size());
       for (HistoricActivityInstance historicActivityInstance : historicActivityInstanceList) {
           if (flowNodeMap.get(historicActivityInstance.getActivityId()) == null) {
               FlowNode sourceNode = (FlowNode) process.getFlowElement(historicActivityInstance.getActivityId());
               flowNodeMap.put(historicActivityInstance.getActivityId(), sourceNode);
           }
       }
       return flowNodeMap;
   }

   /**
    * 撤回任务
    *
    * @param currentTaskId currentTaskId
    * @param targetTaskId  targetTaskId 目标任务，如果为空，默认返回上级，如果找到上级有2个，那目标任务必须得传
    */
   @Transactional(rollbackFor = Exception.class)
   public void backTask(String currentTaskId, String targetTaskId) {
       //准备数据
       TaskService taskService = processEngine.getTaskService();
       // 当前任务
       Task currentTask = taskService.createTaskQuery().taskId(currentTaskId).singleResult();
       String processInstanceId = currentTask.getProcessInstanceId();
       // 获取流程定义
       //任务历史数据
       List<HistoricTaskInstance> historicTaskInstances = historyService
               .createHistoricTaskInstanceQuery()
               .processInstanceId(currentTask.getProcessInstanceId())
               .orderBy(HistoricTaskInstanceQueryProperty.HISTORIC_TASK_INSTANCE_ID)
               .desc()
               .list();
       Map<String, HistoricTaskInstance> historicTaskInstanceMap = historicTaskInstances.stream().collect(Collectors.toMap(HistoricTaskInstance::getId, Function.identity()));
       //所有节点操作数据
       HistoricActivityInstanceQuery historyInstanceQuery = historyService.createHistoricActivityInstanceQuery().processInstanceId(processInstanceId);
       List<HistoricActivityInstance> historicActivityInstanceList = historyInstanceQuery.orderBy(HistoricActivityInstanceQueryProperty.HISTORIC_ACTIVITY_INSTANCE_ID).asc().list();
       Map<String, List<HistoricActivityInstance>> historicActivityInstanceMap = historicActivityInstanceList.stream().collect(Collectors.groupingBy(HistoricActivityInstance::getActivityId));
       Map<String, FlowNode> flowNodeMap = getFlowNodeMap(historicActivityInstanceList, currentTask.getProcessDefinitionId());
       //排除当前任务外的所有正在进行的任务
       List<Task> taskList = taskService.createTaskQuery().processInstanceId(processInstanceId).list().stream().filter(task -> !task.getId().equals(currentTask.getId())).collect(Collectors.toList());
       handleBackTask(currentTask, currentTask.getTaskDefinitionKey(), targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
   }

   @Transactional(rollbackFor = Exception.class)
   public void handleBackTask(Task currentTask, String taskDefinitionKey, String targetTaskId, Map<String, HistoricTaskInstance> historicTaskInstanceMap, Map<String, List<HistoricActivityInstance>> historicActivityInstanceMap, Map<String, FlowNode> flowNodeMap, List<Task> taskList, List<HistoricActivityInstance> historicActivityInstanceList) {
       //判断是否并行
       if (taskList == null || taskList.isEmpty()) {
           //串行
           handleSerial(currentTask, taskDefinitionKey, targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
       } else {
           //并行
           handleParallel(currentTask, taskDefinitionKey, targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
       }
   }

   @Transactional(rollbackFor = Exception.class)
   public void handleParallel(Task currentTask, String taskDefinitionKey, String targetTaskId, Map<String, HistoricTaskInstance> historicTaskInstanceMap, Map<String, List<HistoricActivityInstance>> historicActivityInstanceMap, Map<String, FlowNode> flowNodeMap, List<Task> taskList, List<HistoricActivityInstance> historicActivityInstanceList) {
       List<SequenceFlow> sequenceFlows = flowNodeMap.get(taskDefinitionKey).getIncomingFlows();
       if (sequenceFlows.size() == 1) {
           //当前节点的上级节点只有一条
           SequenceFlow sequenceFlow = sequenceFlows.get(0);
           //查询历史节点
           HistoricActivityInstance historicActivityInstance = historicActivityInstanceList.stream().filter(query -> query.getActivityId().equals(sequenceFlow.getSourceRef())).collect(Collectors.toList()).get(0);
           //判断来源类型
           if (historicActivityInstance.getActivityType().equals(PARALLEL_GATEWAY)) {
               //网关
               //查找网关的父任务
               Set<String> parentFlowNodes = queryParentFlowNode(historicActivityInstance.getActivityId(), flowNodeMap);
               if (!parentFlowNodes.isEmpty()) {
                   if (parentFlowNodes.size() == 1) {
                       //如果只有一个父节点
                       String activityId = new ArrayList<>(parentFlowNodes).get(0);
                       if (historicActivityInstanceMap.get(activityId).get(0).getActivityType().equals(USER_TASK)) {
                           //用户任务
                           deleteTaskMultiple(flowNodeMap, null, null, activityId, currentTask, taskList, historicActivityInstance.getActivityId());
                       } else {
                           //递归去查找父任务的前一个
                           handleBackTask(currentTask, historicActivityInstanceMap.get(activityId).get(0).getActivityId(), targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
                       }
                   } else {
                       //当前节点的上级节点有多条 这里需要指定要回退的taskId
                       deleteTaskMultiple(flowNodeMap, historicTaskInstanceMap, targetTaskId, null, currentTask, taskList, historicActivityInstance.getActivityId());
                   }
               } else {
                   //没有父级任务，图有问题
                   throw new CommonValidateException("bpmn doc error");
               }

           } else if (historicActivityInstance.getActivityType().equals(USER_TASK)) {
               //用户任务
               deleteTaskMultiple(flowNodeMap, null, null, historicActivityInstance.getActivityId(), currentTask, taskList, historicActivityInstance.getActivityId());
           } else {
               //todo 还没想好这种场景
               throw new CommonValidateException(BPMN_NOT_SUPPORT);
           }
       } else {
           //当前节点的上级节点有多条 这里需要指定要回退的taskId
           deleteTaskMultiple(flowNodeMap, historicTaskInstanceMap, targetTaskId, null, currentTask, taskList, null);

       }
   }

   @Transactional(rollbackFor = Exception.class)
   public void handleSerial(Task currentTask, String taskDefinitionKey, String targetTaskId, Map<String, HistoricTaskInstance> historicTaskInstanceMap, Map<String, List<HistoricActivityInstance>> historicActivityInstanceMap, Map<String, FlowNode> flowNodeMap, List<Task> taskList, List<HistoricActivityInstance> historicActivityInstanceList) {
       FlowNode currentNode = flowNodeMap.get(taskDefinitionKey);
       List<SequenceFlow> sequenceFlows = currentNode.getIncomingFlows();
       if (sequenceFlows.size() == 1) {
           SequenceFlow sequenceFlow = sequenceFlows.get(0);
           HistoricActivityInstance historicActivityInstance = historicActivityInstanceMap.get(sequenceFlow.getSourceRef()).get(0);
           //网关
           if (historicActivityInstance.getActivityType().equals(PARALLEL_GATEWAY) || historicActivityInstance.getActivityType().equals(EXCLUSIVE_GATEWAY)) {
               //查找网关的父任务
               Set<String> parentFlowNodes = queryParentFlowNode(historicActivityInstance.getActivityId(), flowNodeMap);
               if (!parentFlowNodes.isEmpty()) {
                   handleBackTaskSingle(parentFlowNodes, currentTask, targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
               } else {
                   //当前节点的上级节点有多条 这里需要指定要回退的taskId
                   deleteTaskMultiple(flowNodeMap, historicTaskInstanceMap, targetTaskId, null, currentTask, taskList, null);
               }
           } else if (historicActivityInstance.getActivityType().equals(USER_TASK)) {
               deleteTaskSingle(flowNodeMap, historicActivityInstance.getActivityId(), currentTask.getId());
           } else {
               //todo 还没想好这种场景
               throw new CommonValidateException(BPMN_NOT_SUPPORT);
           }
       } else {
           Map<String, HistoricVariableInstance> historicVariableInstanceMap = getHistoricVariableInstanceMap(currentTask.getProcessInstanceId());
           //串行的也有多条连线，可能是通过排他网关过来的
           Set<HistoricActivityInstance> historicActivityInstances = new HashSet<>();
           for (SequenceFlow sequenceFlow : sequenceFlows) {
               //这边他的parent可能是没做过的，要找做过的
               if (historicActivityInstanceMap.get(sequenceFlow.getSourceRef()) != null && querySequenceFlowCondition(sequenceFlow, historicVariableInstanceMap)) {
                   historicActivityInstances.addAll(historicActivityInstanceMap.get(sequenceFlow.getSourceRef()));
               }
           }
           //走过的只有一个
           if (historicActivityInstances.size() == 1) {
               List<HistoricActivityInstance> historicActivityInstancesList = new ArrayList<>(historicActivityInstances);
               if (historicActivityInstancesList.get(0).getActivityType().equals(USER_TASK)) {
                   deleteTaskSingle(flowNodeMap, historicActivityInstancesList.get(0).getActivityId(), currentTask.getId());
               } else if (historicActivityInstancesList.get(0).getActivityType().equals(EXCLUSIVE_GATEWAY)) {
                   //排他网关
                   Set<String> parentFlowNodes = queryParentFlowNode(historicActivityInstancesList.get(0).getActivityId(), flowNodeMap);
                   handleBackTaskSingle(parentFlowNodes, currentTask, targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
               } else {
                   //todo 还没想好这种场景
                   throw new CommonValidateException(BPMN_NOT_SUPPORT);
               }
           } else {
               //当前节点的上级节点有多条 这里需要指定要回退的taskId
               deleteTaskMultiple(flowNodeMap, historicTaskInstanceMap, targetTaskId, null, currentTask, taskList, null);
           }
       }
   }

   private Map<String, HistoricVariableInstance> getHistoricVariableInstanceMap(String processInstanceId) {
       List<HistoricVariableInstance> historicVariableInstances = historyService.createHistoricVariableInstanceQuery()
               .processInstanceId(processInstanceId).list();
       Map<String, HistoricVariableInstance> historicVariableInstanceMap = historicVariableInstances.stream()
               .collect(Collectors.toMap(HistoricVariableInstance::getVariableName,
                       historicVariableInstance -> historicVariableInstance, BinaryOperator.maxBy(Comparator.comparing(HistoricVariableInstance::getId))));
       return historicVariableInstanceMap;
   }

   @Transactional(rollbackFor = Exception.class)
   public void handleBackTaskSingle(Set<String> parentFlowNodes, Task currentTask, String targetTaskId, Map<String, HistoricTaskInstance> historicTaskInstanceMap, Map<String, List<HistoricActivityInstance>> historicActivityInstanceMap, Map<String, FlowNode> flowNodeMap, List<Task> taskList, List<HistoricActivityInstance> historicActivityInstanceList) {
       if (parentFlowNodes.size() == 1) {
           List<String> parentFlowNodeList = new ArrayList<>(parentFlowNodes);
           if (historicActivityInstanceMap.get(parentFlowNodeList.get(0)).get(0).getActivityType().equals(USER_TASK)) {
               deleteTaskSingle(flowNodeMap, parentFlowNodeList.get(0), currentTask.getId());
           } else {
               //递归去查找父任务的前一个
               handleBackTask(currentTask, historicActivityInstanceMap.get(parentFlowNodeList.get(0)).get(0).getActivityId(), targetTaskId, historicTaskInstanceMap, historicActivityInstanceMap, flowNodeMap, taskList, historicActivityInstanceList);
           }
       } else {
           //当前节点的上级节点有多条 这里需要指定要回退的taskId
           deleteTaskMultiple(flowNodeMap, historicTaskInstanceMap, targetTaskId, null, currentTask, taskList, null);
       }
   }

   private void validatorTargetTask(Map<String, HistoricTaskInstance> historicTaskInstanceMap, String targetTaskId) {
       if (StringUtils.isEmpty(targetTaskId) || StringUtils.isBlank(targetTaskId)) {
           throw new CommonValidateException("target task id cannot be null");
       }
       if (historicTaskInstanceMap == null || historicTaskInstanceMap.isEmpty()) {
           throw new CommonValidateException("historic task instance cannot be null");
       }
   }

   @Transactional(rollbackFor = Exception.class)
   public void deleteTaskMultiple(Map<String, FlowNode> flowNodeMap, Map<String, HistoricTaskInstance> historicTaskInstanceMap, String targetTaskId, String targetTaskDefinitionKey, Task currentTask, List<Task> taskList, String targetParentTaskDefinitionKey) {
       if (StringUtils.isEmpty(targetTaskDefinitionKey) || StringUtils.isBlank(targetTaskDefinitionKey)) {
           validatorTargetTask(historicTaskInstanceMap, targetTaskId);
           targetTaskDefinitionKey = historicTaskInstanceMap.get(targetTaskId).getTaskDefinitionKey();
       }
       FlowNode targetNode = flowNodeMap.get(targetTaskDefinitionKey);
       ManagementService managementService = processEngine.getManagementService();
       //删除当前任务
       managementService.executeCommand(new DeleteTaskCmd(currentTask.getId()));
       // 删除当前运行的其他相同父任务的子任务
       Set<Task> sameParentTasks = getSameParentTasks(flowNodeMap, taskList, targetParentTaskDefinitionKey);
       for (Task task : sameParentTasks) {
           managementService.executeCommand(new DeleteTaskCmd(task.getId()));
       }
       // 流程执行到来源节点
       managementService.executeCommand(new SetFLowNodeAndGoCmd(targetNode, currentTask.getExecutionId()));
   }

   @Transactional(rollbackFor = Exception.class)
   public void deleteTaskSingle(Map<String, FlowNode> flowNodeMap, String targetTaskActivitiId, String currentTaskId) {
       ManagementService managementService = processEngine.getManagementService();
       FlowNode targetNode = flowNodeMap.get(targetTaskActivitiId);
       // 删除当前运行任务
       String executionEntityId = managementService.executeCommand(new DeleteTaskCmd(currentTaskId));
       // 流程执行到来源节点
       managementService.executeCommand(new SetFLowNodeAndGoCmd(targetNode, executionEntityId));
   }

   private Set<String> queryParentFlowNode(String activityId, Map<String, FlowNode> flowNodeMap) {
       Set<String> flowNodeList = new HashSet<>();
       for (String key : flowNodeMap.keySet()) {
           if (!key.equals(activityId)) {
               FlowNode flowNode = flowNodeMap.get(key);
               List<SequenceFlow> sequenceFlows = flowNode.getOutgoingFlows();
               for (SequenceFlow sequenceFlow : sequenceFlows) {
                   if (sequenceFlow.getTargetRef().equals(activityId)) {
                       flowNodeList.add(key);
                       break;
                   }
               }
           }
       }
       return flowNodeList;
   }

   private Set<Task> getSameParentTasks(Map<String, FlowNode> flowNodeMap, List<Task> taskList, String taskDefinitionKey) {
       if (taskDefinitionKey == null) {
           return new HashSet<>(taskList);
       }
       Set<Task> tasks = new HashSet<>();
       for (Task task : taskList) {
           List<SequenceFlow> sequenceFlows = flowNodeMap.get(task.getTaskDefinitionKey()).getIncomingFlows();
           for (SequenceFlow sequenceFlow : sequenceFlows) {
               if (sequenceFlow.getSourceRef().equals(taskDefinitionKey)) {
                   tasks.add(task);
                   break;
               }
           }
       }
       return tasks;
   }

   @Transactional(rollbackFor = Exception.class)
   public void importBpmnFile(MultipartFile file, String type, String typeName) {
       try {
           InputStream fileInputStream = file.getInputStream();
           //创建转换对象
           BpmnXMLConverter bpmnXMLConverter = new BpmnXMLConverter();
           //读取xml文件
           XMLInputFactory xmlInputFactory = XMLInputFactory.newInstance();
           XMLStreamReader xmlStreamReader = xmlInputFactory.createXMLStreamReader(fileInputStream);
           //将xml文件转换成BpmnModel
           BpmnModel bpmnModel = bpmnXMLConverter.convertToBpmnModel(xmlStreamReader);
           bpmnModel.getMainProcess().setId(type);
           bpmnModel.getMainProcess().setName(typeName);
           Deployment deployment = repositoryService.createDeployment()
                   .addBpmnModel(typeName + ".bpmn", bpmnModel)
                   .key(IdWorker.getIdStr())
                   .deploy();
           ProcessDefinition processDefinition = repositoryService.createProcessDefinitionQuery().deploymentId(deployment.getId()).singleResult();
           BpmnModel model = repositoryService.getBpmnModel(processDefinition.getId());
           if (model != null) {
               Collection<FlowElement> flowElementCollection = model.getMainProcess().getFlowElements();
               for (FlowElement e : flowElementCollection) {
                   LOGGER.info("flowelement id:" + e.getId() + "  name:" + e.getName() + "   class:" + e.getClass().toString());
               }
           }
           activitiRepository.updateActReProcdef(processDefinition.getId());
       } catch (Exception e) {
           LOGGER.error("导入流程定义失败:{}", e.getMessage(), e);
       }
   }

操作系统相关

什么是socket，网络传输有哪几个层级

什么是socket

网络上的两个程序通过一个双向的通信连接实现数据的交换，这个连接的一端称为一个socket。

建立网络通信连接至少要一对端口号(socket)。socket本质是编程接口(API)，对TCP/IP的封装，TCP/IP也要提供可供程序员做网络开发所用的接口，这就是Socket编程接口;HTTP是轿车，提供了封装或者显示数据的具体形式;Socket是发动机，提供了网络通信的能力。

Socket的英文原义是”孔”或”插座”。作为BSD UNIX的进程通信机制，取后一种意思。通常也称作”套接字”，用于描述IP地址和端口，是一个通信链的句柄，可以用来实现不同虚拟机或不同计算机之间的通信。在Internet上的主机一般运行了多个服务软件，同时提供几种服务。每种服务都打开一个Socket，并绑定到一个端口上，不同的端口对应于不同的服务。

Socket正如其英文原意那样，像一个多孔插座。一台主机犹如布满各种插座的房间，每个插座有一个编号，有的插座提供220伏交流电， 有的提供110伏交流电，有的则提供有线电视节目。 客户软件将插头插到不同编号的插座，就可以得到不同的服务。

网络传输有哪几个层级

所谓的协议就是双方进行数据传输的一种格式。

网络中，一帧以太网数据包的格式：

TCP/IP是传输层协议，主要解决数据如何在网络中传输；而HTTP是应用层协议，主要解决如何包装数据。

把IP想像成一种高速公路，它允许其它协议在上面行驶并找到到其它电脑的出口。TCP和UDP是高速公路上的“卡车”，它们携带的货物就是像HTTP，文件传输协议FTP这样的协议等。(可以这样理解:TCP和UDP都是用来传输其他协议的)

而Socket是对TCP/IP协议的封装，Socket本身并不是协议，而是一个调用接口（API），通过Socket，我们才能使用TCP/IP协议。

ip地址

每个IP地址包括两个标识码（ID），即网络ID和主机ID。同一个物理网络上的所有主机都使用同一个网络ID，网络上的一个主机（包括网络上工作站，服务器和路由器等）有一个主机ID与其对应。

Internet委员会定义了5种IP地址类型以适合不同容量的网络，即A类~E类。
其中A、B、C3类（如下表格）由InternetNIC在全球范围内统一分配，D、E类为特殊地址

TIME_WAIT是什么状态还记得吗，什么情况下网络会出现这个状态

出现原因分析

TCP连接的终止
TCP建立一个连接至少需要交换三个分组，也因此称之为TCP的三路握手（three-way handshake），然而在TCP终止连接时，由于双方都需要发送一个FIN分节给对端确认，因此TCP终止连接一般是需要交换四个分节。具体来看：

1、 应用进程（active close）首先调用close，于是导致TCP发送一个FIN分节，表示数据已分送完毕，请求关闭套接字。

2、 另一端应用进程（passive close）接受收到FIN，并由该端的TCP确认（确认的过程是TCP发送ACK分节给对端套接字）。FIN的接受也作为文件结束符传递给上层应用进程。这里的文件结束符并非应用进程的EOF，在TCP字节流中，EOF的读或写通过收发一个特殊的FIN分节来实现

3、 另端（passive close）应用进程在接受到文件束符后，会调用close关闭它的套接字，这导致该端的TCP也发送了一个FIN分节。

4、 主动关闭端（active close）接受到这个FIN后，TCP对它进行确认。（TCP发送ACK分节，值得注意的是主动关闭端在未接受到FIN之前，它的状态就是TIME_WAIT）。

综上所述：TIME_WAIT状态出现场景是主动关闭端在未接受到FIN之前，它的状态就是TIME_WAIT。

TCP为什么如此设计

1。防止上一次连接中的包(特别是最后一个ACK包)，迷路后重新出现，影响新连接 （经过2MSL(max segment lifetime)，上一次连接中所有的重复包都会消失）。

2。可靠的关闭TCP连接 在主动关闭方发送的最后一个 ack(fin) ，有可能丢失，这时被动方会重新发
fin, 如果这时主动方处于 CLOSED 状态 ，就会响应 rst 而不是 ack。所以 主动方要处于 TIME_WAIT 状态，而不能是 CLOSED 。TIME_WAIT 并不会占用很大资源的，除非受到攻击。还有，如果一方 send 或 recv 超时，就会直接进入 CLOSED 状态。

规避大量出现TIME_WAIT的方法

• net.ipv4.tcp_tw_reuse和net.ipv4.tcp_tw_recycle的开启都是为了回收处于TIME_WAIT状态的资源。
• net.ipv4.tcp_fin_timeout这个时间可以减少在异常情况下服务器从FIN-WAIT-2转到TIME_WAIT的时间。
• net.ipv4.tcp_keepalive_*一系列参数，是用来设置服务器检测连接存活的相关配置。

在服务器的日常维护过程中，会经常用到下面的命令：

1

netstat -n | awk '/^tcp/ {++S[$NF]} END {for(a in S) print a, S[a]}'

它会显示例如下面的信息：

1
2
3
4
5
6

TIME_WAIT 814
CLOSE_WAIT 1
FIN_WAIT1 1
ESTABLISHED 634
SYN_RECV 2
LAST_ACK 1

常用的三个状态是：ESTABLISHED 表示正在通信，TIME_WAIT 表示主动关闭，CLOSE_WAIT 表示被动关闭。

三次握手的详细描述

第一次握手：建立连接。客户端发送连接请求报文段，将SYN位置为1，Sequence Number为x；然后，客户端进入SYN_SEND状态，等待服务器的确认；

第二次握手：服务器收到SYN报文段。服务器收到客户端的SYN报文段，需要对这个SYN报文段进行确认，设置Acknowledgment Number为x+1(Sequence Number+1)；同时，自己自己还要发送SYN请求信息，将SYN位置为1，Sequence Number为y；服务器端将上述所有信息放到一个报文段（即SYN+ACK报文段）中，一并发送给客户端，此时服务器进入SYN_RECV状态；

第三次握手：客户端收到服务器的SYN+ACK报文段。然后将Acknowledgment Number设置为y+1，向服务器发送ACK报文段，这个报文段发送完毕以后，客户端和服务器端都进入ESTABLISHED状态，完成TCP三次握手。

为什么要三次握手

既然总结了TCP的三次握手，那为什么非要三次呢？怎么觉得两次就可以完成了。那TCP为什么非要进行三次连接呢？在谢希仁的《计算机网络》中是这样说的：

为了防止已失效的连接请求报文段突然又传送到了服务端，因而产生错误。

在书中同时举了一个例子，如下：

“已失效的连接请求报文段”的产生在这样一种情况下：client发出的第一个连接请求报文段并没有丢失，而是在某个网络结点长时间的滞留了，以致延误到连接释放以后的某个时间才到达server。本来这是一个早已失效的报文段。但server收到此失效的连接请求报文段后，就误认为是client再次发出的一个新的连接请求。于是就向client发出确认报文段，同意建立连接。假设不采用“三次握手”，那么只要server发出确认，新的连接就建立了。由于现在client并没有发出建立连接的请求，因此不会理睬server的确认，也不会向server发送数据。但server却以为新的运输连接已经建立，并一直等待client发来数据。这样，server的很多资源就白白浪费掉了。采用“三次握手”的办法可以防止上述现象发生。例如刚才那种情况，client不会向server的确认发出确认。server由于收不到确认，就知道client并没有要求建立连接。”

这就很明白了，防止了服务器端的一直等待而浪费资源。

四次挥手的详细描述

当客户端和服务器通过三次握手建立了TCP连接以后，当数据传送完毕，肯定是要断开TCP连接的啊。那对于TCP的断开连接，这里就有了神秘的“四次分手”。

1. 第一次分手：主机1（可以使客户端，也可以是服务器端），设置Sequence Number和Acknowledgment Number，向主机2发送一个FIN报文段；此时，主机1进入FIN_WAIT_1状态；这表示主机1没有数据要发送给主机2了；
2. 第二次分手：主机2收到了主机1发送的FIN报文段，向主机1回一个ACK报文段，Acknowledgment Number为Sequence Number加1；主机1进入FIN_WAIT_2状态；主机2告诉主机1，我也没有数据要发送了，可以进行关闭连接了；
3. 第三次分手：主机2向主机1发送FIN报文段，请求关闭连接，同时主机2进入CLOSE_WAIT状态；
4. 第四次分手：主机1收到主机2发送的FIN报文段，向主机2发送ACK报文段，然后主机1进入TIME_WAIT状态；主机2收到主机1的ACK报文段以后，就关闭连接；此时，主机1等待2MSL后依然没有收到回复，则证明Server端已正常关闭，那好，主机1也可以关闭连接了。

至此，TCP的四次分手就这么愉快的完成了。当你看到这里，你的脑子里会有很多的疑问，很多的不懂，感觉很凌乱；没事，我们继续总结。

为什么要四次分手

那四次分手又是为何呢？TCP协议是一种面向连接的、可靠的、基于字节流的运输层通信协议。TCP是全双工模式，这就意味着，当主机1发出FIN报文段时，只是表示主机1已经没有数据要发送了，主机1告诉主机2，它的数据已经全部发送完毕了；但是，这个时候主机1还是可以接受来自主机2的数据；当主机2返回ACK报文段时，表示它已经知道主机1没有数据发送了，但是主机2还是可以发送数据到主机1的；当主机2也发送了FIN报文段时，这个时候就表示主机2也没有数据要发送了，就会告诉主机1，我也没有数据要发送了，之后彼此就会愉快的中断这次TCP连接。如果要正确的理解四次分手的原理，就需要了解四次分手过程中的状态变化。

• FIN_WAIT_1: 这个状态要好好解释一下，其实FIN_WAIT_1和FIN_WAIT_2状态的真正含义都是表示等待对方的FIN报文。而这两种状态的区别是：FIN_WAIT_1状态实际上是当SOCKET在ESTABLISHED状态时，它想主动关闭连接，向对方发送了FIN报文，此时该SOCKET即进入到FIN_WAIT_1状态。而当对方回应ACK报文后，则进入到FIN_WAIT_2状态，当然在实际的正常情况下，无论对方何种情况下，都应该马上回应ACK报文，所以FIN_WAIT_1状态一般是比较难见到的，而FIN_WAIT_2状态还有时常常可以用netstat看到。（主动方）
• FIN_WAIT_2：上面已经详细解释了这种状态，实际上FIN_WAIT_2状态下的SOCKET，表示半连接，也即有一方要求close连接，但另外还告诉对方，我暂时还有点数据需要传送给你(ACK信息)，稍后再关闭连接。（主动方）
• CLOSE_WAIT：这种状态的含义其实是表示在等待关闭。怎么理解呢？当对方close一个SOCKET后发送FIN报文给自己，你系统毫无疑问地会回应一个ACK报文给对方，此时则进入到CLOSE_WAIT状态。接下来呢，实际上你真正需要考虑的事情是察看你是否还有数据发送给对方，如果没有的话，那么你也就可以 close这个SOCKET，发送FIN报文给对方，也即关闭连接。所以你在CLOSE_WAIT状态下，需要完成的事情是等待你去关闭连接。（被动方）
• LAST_ACK: 这个状态还是比较容易好理解的，它是被动关闭一方在发送FIN报文后，最后等待对方的ACK报文。当收到ACK报文后，也即可以进入到CLOSED可用状态了。（被动方）
• TIME_WAIT: 表示收到了对方的FIN报文，并发送出了ACK报文，就等2MSL后即可回到CLOSED可用状态了。如果FINWAIT1状态下，收到了对方同时带FIN标志和ACK标志的报文时，可以直接进入到TIME_WAIT状态，而无须经过FIN_WAIT_2状态。（主动方）
• CLOSED: 表示连接中断。

哪些典型的应用用的是UDP？

TCP应用

（1）FTP：文件传输协议；

（2）SSH：安全登录、文件传送(SCP)和端口重定向；

（3）Telnet：不安全的文本传送；

（4）SMTP：简单邮件传输协议Simple Mail Transfer Protocol (E-mail)；

（5）HTTP：超文本传送协议 (WWW)；

UDP应用

（1）流媒体

采用TCP，一旦发生丢包，TCP会将后续包缓存起来，等前面的包重传并接收到后再继续发送，延迟会越来越大。基于UDP的协议如WebRTC是极佳的选择。

（2）实时游戏

对实时要求较为严格的情况下，采用自定义的可靠UDP协议，比如Enet、RakNet（用户有sony online game、minecraft）等，自定义重传策略，能够把丢包产生的延迟降到最低，尽量减少网络问题对游戏性造成的影响。

采用UDP的经典游戏如FPS游戏Quake、CS，著名的游戏引擎Unity3D采用的也是RakNet。

（3）物联网

2014年google旗下的Nest建立Thread Group，推出了物联网通信协议Thread，完善物联网通信。

全球将近50%的人都在使用互联网，人们不断的追求更快、更好的服务，一切都在变化，在越来越多的领域，UDP将会抢占TCP的主导地位。

（4）QQ 文件传输、QQ语音、QQ视频

对于网络通讯质量要求不高的情况下，要求网络通讯速度能尽量快捷方便，就可以使用UDP技术。

内核态和用户态、cas 和 sout 哪个用到了内核态和用户态的切换

究竟什么是用户态，什么是内核态，这两个基本概念以前一直理解得不是很清楚，根本原因个人觉得是在于因为大部分时候我们在写程序时关注的重点和着眼的角度放在了实现的功能和代码的逻辑性上，先看一个例子：

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

void testfork(){  

    if(0 = = fork()){  

         printf(“create new process success!\n”);  

     }  

    printf(“testfork ok\n”);  

 }

这段代码很简单，从功能的角度来看，就是实际执行了一个fork()，生成一个新的进程，从逻辑的角度看，就是判断了如果fork()返回的是则打印相关语句，然后函数最后再打印一句表示执行完整个testfork()函数。代码的执行逻辑和功能上看就是如此简单，一共四行代码，从上到下一句一句执行而已，完全看不出来哪里有体现出用户态和进程态的概念。

如果说前面两种是静态观察的角度看的话，我们还可以从动态的角度来看这段代码，即它被转换成CPU执行的指令后加载执行的过程，这时这段程序就是一个动态执行的指令序列。而究竟加载了哪些代码，如何加载就是和操作系统密切相关了。

特权级

熟悉Unix/Linux系统的人都知道，fork的工作实际上是以系统调用的方式完成相应功能的，具体的工作是由sys_fork负责实施。其实无论是不是Unix或者Linux，对于任何操作系统来说，创建一个新的进程都是属于核心功能，因为它要做很多底层细致地工作，消耗系统的物理资源，比如分配物理内存，从父进程拷贝相关信息，拷贝设置页目录页表等等，这些显然不能随便让哪个程序就能去做，于是就自然引出特权级别的概念，显然，最关键性的权力必须由高特权级的程序来执行，这样才可以做到集中管理，减少有限资源的访问和使用冲突。

特权级显然是非常有效的管理和控制程序执行的手段，因此在硬件上对特权级做了很多支持，就Intel x86架构的CPU来说一共有0~3四个特权级，0级最高，3级最低，硬件上在执行每条指令时都会对指令所具有的特权级做相应的检查，相关的概念有 CPL、DPL和RPL，这里不再过多阐述。硬件已经提供了一套特权级使用的相关机制，软件自然就是好好利用的问题，这属于操作系统要做的事情，对于 Unix/Linux来说，只使用了0级特权级和3级特权级。也就是说在Unix/Linux系统中，一条工作在级特权级的指令具有了CPU能提供的最高权力，而一条工作在3级特权级的指令具有CPU提供的最低或者说最基本权力。

用户态和内核态

现在我们从特权级的调度来理解用户态和内核态就比较好理解了，当程序运行在3级特权级上时，就可以称之为运行在用户态，因为这是最低特权级，是普通的用户进程运行的特权级，大部分用户直接面对的程序都是运行在用户态；反之，当程序运行在级特权级上时，就可以称之为运行在内核态。

虽然用户态下和内核态下工作的程序有很多差别，但最重要的差别就在于特权级的不同，即权力的不同。运行在用户态下的程序不能直接访问操作系统内核数据结构和程序，比如上面例子中的testfork()就不能直接调用 sys_fork()，因为前者是工作在用户态，属于用户态程序，而sys_fork()是工作在内核态，属于内核态程序。

当我们在系统中执行一个程序时，大部分时间是运行在用户态下的，在其需要操作系统帮助完成某些它没有权力和能力完成的工作时就会切换到内核态，比如testfork()最初运行在用户态进程下，当它调用fork()最终触发 sys_fork()的执行时，就切换到了内核态。

用户态和内核态的转换

1）用户态切换到内核态的3种方式

a. 系统调用

这是用户态进程主动要求切换到内核态的一种方式，用户态进程通过系统调用申请使用操作系统提供的服务程序完成工作，比如前例中fork()实际上就是执行了一个创建新进程的系统调用。而系统调用的机制其核心还是使用了操作系统为用户特别开放的一个中断来实现，例如Linux的int 80h中断。

b. 异常

当CPU在执行运行在用户态下的程序时，发生了某些事先不可知的异常，这时会触发由当前运行进程切换到处理此异常的内核相关程序中，也就转到了内核态，比如缺页异常。

c. 外围设备的中断

当外围设备完成用户请求的操作后，会向CPU发出相应的中断信号，这时CPU会暂停执行下一条即将要执行的指令转而去执行与中断信号对应的处理程序，如果先前执行的指令是用户态下的程序，那么这个转换的过程自然也就发生了由用户态到内核态的切换。比如硬盘读写操作完成，系统会切换到硬盘读写的中断处理程序中执行后续操作等。

这3种方式是系统在运行时由用户态转到内核态的最主要方式，其中系统调用可以认为是用户进程主动发起的，异常和外围设备中断则是被动的。

2）具体的切换操作

从触发方式上看，可以认为存在前述3种不同的类型，但是从最终实际完成由用户态到内核态的切换操作上来说，涉及的关键步骤是完全一致的，没有任何区别，都相当于执行了一个中断响应的过程，因为系统调用实际上最终是中断机制实现的，而异常和中断的处理机制基本上也是一致的，关于它们的具体区别这里不再赘述。关于中断处理机制的细节和步骤这里也不做过多分析，涉及到由用户态切换到内核态的步骤主要包括：

[1] 从当前进程的描述符中提取其内核栈的ss0及esp0信息。

[2] 使用ss0和esp0指向的内核栈将当前进程的cs,eip,eflags,ss,esp信息保存起来，这个

过程也完成了由用户栈到内核栈的切换过程，同时保存了被暂停执行的程序的下一

条指令。

[3] 将先前由中断向量检索得到的中断处理程序的cs,eip信息装入相应的寄存器，开始

执行中断处理程序，这时就转到了内核态的程序执行了。

算法相关

• 递归使用中有什么需要注意的地方，递归写法一般可以用什么去替换？
• 如下图所示：
  
• 设计一个发号器，考虑集群和高并发的情况，要求发号器生成的id是递增趋势，通过id可以区分出来是今天生成的id还是昨天生成的id，但是生成的id中不能直接带有日期，要具有一定的混淆功能，白纸写代码
• 一个二位数组，每个元素都可以往上下左右四个方向走，寻找最长递增路径。如下图所示，最长递增路径即红色字体路径。白纸写代码。
  
• 反转链表，要求时间复杂度O(N)，空间复杂度O(1)
• 非递归实现斐波那契数列
• 这一周股市价格为[2,6,1,4,8]，求哪一天买入哪一天卖出，可获得最大收益，最大收益为多少
• 按照箭头方向查找二叉树
  
• 表a b c之间用ID关联，求阴影部分的数据
  
• 一个整形无序数组，里面三个数只和等于一个目标值，求这三个数
• 链表问题
  
• 扑克牌问题
  有十张扑克牌，从上面开始抽，抽出一张放桌子上，然后再抽出一张放扑克牌的最下面，这样循环往复的操作，直到手里的牌都没有了。这时，桌子上牌的顺序正好是1 2 3 4 5 6 7 8 9 10。要求写代码求出原顺序
• 手写大顶堆
• 手写LRU 算法
• 字符串相加
  两个数字类型的字符串，直接转int或者double肯定都放不下，然后求这两个数的和，返回值还是字符串，15分钟时间，要求无Bug
• 寻找目标值位置
  有一个二维数组，数组横向有序，纵向有序，求目标值的位置，10分钟时间
• 求字符串“efabcbaefehiabcba”中最长的回文数，不去重
• 反转int类型的值x，不要借用String，只用int 即可。&& 针对该程序，写出其应有的测试用例
• top K 问题

相关工具使用

Redis

Redis 中有几种类型 & 各自底层怎么实现的 & 项目中哪个地方用了什么类型，怎么使用的？

redis底层原理

• 使用 ziplist 存储链表，ziplist是一种压缩链表，它的好处是更能节省内存空间，因为它所存储的内容都是在连续的内存区域当中的。
• 使用 skiplist(跳跃表)来存储有序集合对象、查找上先从高Level查起、时间复杂度和红黑树相当，实现容易，无锁、并发性好。

1 单线程模型

Redis客户端对服务端的每次调用都经历了发送命令，执行命令，返回结果三个过程。其中执行命令阶段，由于Redis是单线程来处理命令的，所有每一条到达服务端的命令不会立刻执行，所有的命令都会进入一个队列中，然后逐个被执行。并且多个客户端发送的命令的执行顺序是不确定的。但是可以确定的是不会有两条命令被同时执行，不会产生并发问题，这就是Redis的单线程基本模型。

2 单线程模型每秒万级别处理能力的原因

（1）纯内存访问。数据存放在内存中，内存的响应时间大约是100纳秒，这是Redis每秒万亿级别访问的重要基础。

（2）非阻塞I/O，Redis采用epoll做为I/O多路复用技术的实现，再加上Redis自身的事件处理模型将epoll中的连接，读写，关闭都转换为了时间，不在I/O上浪费过多的时间。

（3）单线程避免了线程切换和竞态产生的消耗。

（4）Redis采用单线程模型，每条命令执行如果占用大量时间，会造成其他线程阻塞，对于Redis这种高性能服务是致命的，所以Redis是面向高速执行的数据库。

字符串string

字符串类型是Redis中最为基础的数据存储类型，是一个由字节组成的序列，他在Redis中是二进制安全的，这便意味着该类型可以接受任何格式的数据，如JPEG图像数据货Json对象描述信息等，是标准的key-value，一般来存字符串，整数和浮点数。Value最多可以容纳的数据长度为512MB
应用场景：很常见的场景用于统计网站访问数量，当前在线人数等。incr命令(++操作)

列表list

Redis的列表允许用户从序列的两端推入或者弹出元素，列表由多个字符串值组成的有序可重复的序列，是链表结构，所以向列表两端添加元素的时间复杂度为0(1)，获取越接近两端的元素速度就越快。这意味着即使是一个有几千万个元素的列表，获取头部或尾部的10条记录也是极快的。List中可以包含的最大元素数量是4294967295。
应用场景：1.最新消息排行榜。2.消息队列，以完成多程序之间的消息交换。可以用push操作将任务存在list中（生产者），然后线程在用pop操作将任务取出进行执行。（消费者）

散列hash

Redis中的散列可以看成具有String key和String value的map容器，可以将多个key-value存储到一个key中。每一个Hash可以存储4294967295个键值对。
应用场景：例如存储、读取、修改用户属性（name，age，pwd等）

集合set

Redis的集合是无序不可重复的，和列表一样，在执行插入和删除和判断是否存在某元素时，效率是很高的。集合最大的优势在于可以进行交集并集差集操作。Set可包含的最大元素数量是4294967295。
应用场景：1.利用交集求共同好友。2.利用唯一性，可以统计访问网站的所有独立IP。3.好友推荐的时候根据tag求交集，大于某个threshold（临界值的）就可以推荐。

有序集合sorted set

和set很像，都是字符串的集合，都不允许重复的成员出现在一个set中。他们之间差别在于有序集合中每一个成员都会有一个分数(score)与之关联，Redis正是通过分数来为集合中的成员进行从小到大的排序。尽管有序集合中的成员必须是卫衣的，但是分数(score)却可以重复。
应用场景：可以用于一个大型在线游戏的积分排行榜，每当玩家的分数发生变化时，可以执行zadd更新玩家分数(score)，此后在通过zrange获取几分top ten的用户信息。

key的通用操作

所有的数据类型都可以使用的

Redis如何实现分布式锁，zk如何实现分布式锁，两者的区别。如果service还没执行完，分布式锁在Redis中已经过期了，怎么解决这种问题？

解决redis分布式锁过期时间到了业务没执行完问题

很多同学在用分布式锁时,都是直接百度搜索找一个Redis分布式锁工具类就直接用了，其实Redis分布式锁比较正确的姿势是采用redisson这个客户端工具

默认情况下,加锁的时间是30秒.如果加锁的业务没有执行完,那么到 30-10 = 20秒的时候,就会进行一次续期,把锁重置成30秒.那这个时候可能又有同学问了,那业务的机器万一宕机了呢?宕机了定时任务跑不了,就续不了期,那自然30秒之后锁就解开了呗。

Redisson分布式锁的底层原理

1）加锁机制

咱们来看上面那张图，现在某个客户端要加锁。如果该客户端面对的是一个redis cluster集群，他首先会根据hash节点选择一台机器。

这里注意，仅仅只是选择一台机器！这点很关键！

紧接着，就会发送一段lua脚本到redis上，那段lua脚本如下所示：

为啥要用lua脚本呢？

因为一大坨复杂的业务逻辑，可以通过封装在lua脚本中发送给redis，保证这段复杂业务逻辑执行的原子性。

那么，这段lua脚本是什么意思呢？

KEYS[1]代表的是你加锁的那个key，比如说：

1

RLock lock = redisson.getLock("myLock");

这里你自己设置了加锁的那个锁key就是“myLock”。

ARGV[1]代表的就是锁key的默认生存时间，默认30秒。

ARGV[2]代表的是加锁的客户端的ID，类似于下面这样：

8743c9c0-0795-4907-87fd-6c719a6b4586:1

给大家解释一下，第一段if判断语句，就是用“exists myLock”命令判断一下，如果你要加锁的那个锁key不存在的话，你就进行加锁。

如何加锁呢？很简单，用下面的命令：

1
2

hset myLock 
8743c9c0-0795-4907-87fd-6c719a6b4586:1 1

通过这个命令设置一个hash数据结构，这行命令执行后，会出现一个类似下面的数据结构：

上述就代表“8743c9c0-0795-4907-87fd-6c719a6b4586:1”这个客户端对“myLock”这个锁key完成了加锁。

接着会执行“pexpire myLock 30000”命令，设置myLock这个锁key的生存时间是30秒。

好了，到此为止，ok，加锁完成了。

（2）锁互斥机制

那么在这个时候，如果客户端2来尝试加锁，执行了同样的一段lua脚本，会咋样呢？

很简单，第一个if判断会执行“exists myLock”，发现myLock这个锁key已经存在了。

接着第二个if判断，判断一下，myLock锁key的hash数据结构中，是否包含客户端2的ID，但是明显不是的，因为那里包含的是客户端1的ID。

所以，客户端2会获取到pttl myLock返回的一个数字，这个数字代表了myLock这个锁key的剩余生存时间。比如还剩15000毫秒的生存时间。

此时客户端2会进入一个while循环，不停的尝试加锁。

（3）watch dog自动延期机制

客户端1加锁的锁key默认生存时间才30秒，如果超过了30秒，客户端1还想一直持有这把锁，怎么办呢？

简单！只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会每隔10秒检查一下，如果客户端1还持有锁key，那么就会不断的延长锁key的生存时间。

（4）可重入加锁机制

那如果客户端1都已经持有了这把锁了，结果可重入的加锁会怎么样呢？

比如下面这种代码：

这时我们来分析一下上面那段lua脚本。

第一个if判断肯定不成立，“exists myLock”会显示锁key已经存在了。

第二个if判断会成立，因为myLock的hash数据结构中包含的那个ID，就是客户端1的那个ID，也就是8743c9c0-0795-4907-87fd-6c719a6b4586:1

此时就会执行可重入加锁的逻辑，他会用：

1
2
3

incrby myLock 

 8743c9c0-0795-4907-87fd-6c71a6b4586:1 1

通过这个命令，对客户端1的加锁次数，累加1。

此时myLock数据结构变为下面这样：

（5）释放锁机制

如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑也是非常简单的。

其实说白了，就是每次都对myLock数据结构中的那个加锁次数减1。

如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用：

“del myLock”命令，从redis里删除这个key。

然后呢，另外的客户端2就可以尝试完成加锁了。

这就是所谓的分布式锁的开源Redisson框架的实现机制。

一般我们在生产系统中，可以用Redisson框架提供的这个类库来基于redis进行分布式锁的加锁与释放锁。

（6）上述Redis分布式锁的缺点

其实上面那种方案最大的问题，就是如果你对某个redis master实例，写入了myLock这种锁key的value，此时会异步复制给对应的master slave实例。

但是这个过程中一旦发生redis master宕机，主备切换，redis slave变为了redis master。

接着就会导致，客户端2来尝试加锁的时候，在新的redis master上完成了加锁，而客户端1也以为自己成功加了锁。

此时就会导致多个客户端对一个分布式锁完成了加锁。

这时系统在业务语义上一定会出现问题，导致各种脏数据的产生。

所以这个就是redis cluster，或者是redis master-slave架构的主从异步复制导致的redis分布式锁的最大缺陷：在redis master实例宕机的时候，可能导致多个客户端同时完成加锁。

Ehcache支持哪些缓存？

Ehcache

EhCache 是一个纯Java的进程内缓存框架，具有快速、精干等特点，是Hibernate中默认的CacheProvider。Ehcache是一种广泛使用的开源Java分布式缓存。

优点：

• 快速
• 简单
• 缓存数据有两级：内存和磁盘，因此无需担心容量问题
• 缓存数据会在虚拟机重启的过程中写入磁盘
• 可以通过RMI、可插入API等方式进行分布式缓存
• 具有缓存和缓存管理器的侦听接口
• 支持多缓存管理器实例，以及一个实例的多个缓存区域
• 提供Hibernate的缓存实现
• 多种缓存策略，Ehcache提供了对大数据的内存和硬盘的存储，最近版本允许多实例、保存对象高灵活性、提供LRU、LFU、FIFO淘汰算法，基础属性支持热配置、支持的插件多

缺点：

• 使用磁盘Cache的时候非常占用磁盘空间；
• 不能保证数据的安全

memcache

memcache 是一种高性能、分布式对象缓存系统，最初设计于缓解动态网站数据库加载数据的延迟性，你可以把它想象成一个大的内存HashTable，就是一个key-value键值缓存。

memcache C语言所编写，依赖于最近版本的GCC和libevent。多线程支持

优点：

一.部分容灾

假设只用一台memcache，如果这台memcache服务器挂掉了，那么请求将不断的冲击数据库，这样有可能搞死数据库，从而引发”雪崩“。如果使用多台memcache服务器，由于memcache使用一致性哈希算法，万一其中一台挂掉了，部分请求还是可以在memcache中命中，为修复系统赢得一些时间。

二.容量问题

一台memcache服务器的容量毕竟有限，可以使用多台memcache服务器，增加缓存容量。

三.均衡请求

使用多台memcache服务器，可以均衡请求，避免所有请求都冲进一台memcache服务器，导致服务器挂掉。

四.利用memcache分布式特性

使用一台memcache服务器，并没有利用memcache的数据分布式特性。

缺点：

• 不能持久化存储
• 存储数据有限制：1M 【大于1M，认为就行分割】（内存碎片）
• mm存储数据只能key-value
• 集群数据没有复制和同步机制 【崩溃不会影响程序，会从数据库中取数据】
• 内存回收不能及时 LRU(算法)：未使用内存》过期内存》最近最少使用内存 这是惰性删除

redis

单线程、读写性能优异、支持数据持久化，支持AOF和RDB两种持久化方式、支持主从复制，主机会自动将数据同步到从机，可以进行读写分离；数据结构丰富：除了支持string类型的value外还支持string、hash、set、sortedset、list等数据结构。

缺点：

1 Redis不具备自动容错和恢复功能，主机从机的宕机都会导致前端部分读写请求失败，需要等待机器重启或者手动切换前端的IP才能恢复。

2 主机宕机，宕机前有部分数据未能及时同步到从机，切换IP后还会引入数据不一致的问题，降低了系统的可用性。

3 Redis的主从复制采用全量复制，复制过程中主机会fork出一个子进程对内存做一份快照，并将子进程的内存快照保存为文件发送给从机，这一过程需要确保主机有足够多的空余内存。若快照文件较大，对集群的服务能力会产生较大的影响，而且复制过程是在从机新加入集群或者从机和主机网络断开重连时都会进行，也就是网络波动都会造成主机和从机间的一次全量的数据复制，这对实际的系统运营造成了不小的麻烦。

4 Redis较难支持在线扩容，在集群容量达到上限时在线扩容会变得很复杂。为避免这一问题，运维人员在系统上线时必须确保有足够的空间，这对资源造成了很大的浪费。

ehcache直接在jvm虚拟机中缓存，速度快，效率高；但是缓存共享麻烦，集群分布式应用不方便。
redis是通过socket访问到缓存服务，效率比ecache低，比数据库要快很多，处理集群和分布式缓存方便，有成熟的方案。
如果是单个应用或者对缓存访问要求很高的应用，用ehcache。
如果是大型系统，存在缓存共享、分布式部署、缓存内容很大的，建议用redis。

Redis是单线程的还是多线程的，为什么这么快？

Redis是一个开源的内存中的数据结构存储系统，它可以用作：数据库、缓存和消息中间件。

它支持多种类型的数据结构，如字符串（String），散列（Hash），列表（List），集合（Set），有序集合（Sorted Set或者是ZSet）与范围查询，Bitmaps，Hyperloglogs 和地理空间（Geospatial）索引半径查询。其中常见的数据结构类型有：String、List、Set、Hash、ZSet这5种。

Redis 内置了复制（Replication），LUA脚本（Lua scripting）， LRU驱动事件（LRU eviction），事务（Transactions） 和不同级别的磁盘持久化（Persistence），并通过 Redis哨兵（Sentinel）和自动分区（Cluster）提供高可用性（High Availability）。

Redis也提供了持久化的选项，这些选项可以让用户将自己的数据保存到磁盘上面进行存储。根据实际情况，可以每隔一定时间将数据集导出到磁盘（快照），或者追加到命令日志中（AOF只追加文件），他会在执行写命令时，将被执行的写命令复制到硬盘里面。您也可以关闭持久化功能，将Redis作为一个高效的网络的缓存数据功能使用。

Redis不使用表，他的数据库不会预定义或者强制去要求用户对Redis存储的不同数据进行关联。

数据库的工作模式按存储方式可分为：硬盘数据库和内存数据库。Redis 将数据储存在内存里面，读写数据的时候都不会受到硬盘 I/O 速度的限制，所以速度极快。

（1）硬盘数据库的工作模式：

（2）内存数据库的工作模式：

Redis为什么这么快

1、完全基于内存，绝大部分请求是纯粹的内存操作，非常快速。数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)；

2、数据结构简单，对数据操作也简单，Redis中的数据结构是专门进行设计的；

3、采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；

4、使用多路I/O复用模型，非阻塞IO；

5、使用底层模型不同，它们之间底层实现方式以及与客户端之间通信的应用协议不一样，Redis直接自己构建了VM 机制 ，因为一般的系统调用系统函数的话，会浪费一定的时间去移动和请求；

以上几点都比较好理解，下边我们针对多路 I/O 复用模型进行简单的探讨：

（1）多路 I/O 复用模型

多路I/O复用模型是利用 select、poll、epoll 可以同时监察多个流的 I/O 事件的能力，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有 I/O 事件时，就从阻塞态中唤醒，于是程序就会轮询一遍所有的流（epoll 是只轮询那些真正发出了事件的流），并且只依次顺序的处理就绪的流，这种做法就避免了大量的无用操作。

这里“多路”指的是多个网络连接，“复用”指的是复用同一个线程。采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求（尽量减少网络 IO 的时间消耗），且 Redis 在内存中操作数据的速度非常快，也就是说内存内的操作不会成为影响Redis性能的瓶颈，主要由以上几点造就了 Redis 具有很高的吞吐量。

那么为什么Redis是单线程的

我们首先要明白，上边的种种分析，都是为了营造一个Redis很快的氛围！官方FAQ表示，因为Redis是基于内存的操作，CPU不是Redis的瓶颈，Redis的瓶颈最有可能是机器内存的大小或者网络带宽。既然单线程容易实现，而且CPU不会成为瓶颈，那就顺理成章地采用单线程的方案了（毕竟采用多线程会有很多麻烦！）。

看到这里，你可能会气哭！本以为会有什么重大的技术要点才使得Redis使用单线程就可以这么快，没想到就是一句官方看似糊弄我们的回答！但是，我们已经可以很清楚的解释了为什么Redis这么快，并且正是由于在单线程模式的情况下已经很快了，就没有必要在使用多线程了！

但是，我们使用单线程的方式是无法发挥多核CPU 性能，不过我们可以通过在单机开多个Redis 实例来完善！

警告1：这里我们一直在强调的单线程，只是在处理我们的网络请求的时候只有一个线程来处理，一个正式的Redis Server运行的时候肯定是不止一个线程的，这里需要大家明确的注意一下！例如Redis进行持久化的时候会以子进程或者子线程的方式执行（具体是子线程还是子进程待读者深入研究）；例如我在测试服务器上查看Redis进程，然后找到该进程下的线程：

Redis Hash中某个key过大，变为String类型的大key，怎么处理，使用中如何避免出现这种问题?

由于redis是单线程运行的，如果一次操作的value很大会对整个redis的响应时间造成负面影响，所以，业务上能拆则拆，下面举几个典型的分拆方案。

单个简单的key存储的value很大

1.1、 改对象需要每次都整存整取
可以尝试将对象分拆成几个key-value， 使用multiGet获取值，这样分拆的意义在于分拆单次操作的压力，将操作压力平摊到多个redis实例中，降低对单个redis的IO影响；

1.2、该对象每次只需要存取部分数据
可以像第一种做法一样，分拆成几个key-value， 也可以将这个存储在一个hash中，每个field代表一个具体的属性，使用hget,hmget来获取部分的value，使用hset，hmset来更新部分属性

hash， set，zset，list 中存储过多的元素

类似于场景一种的第一个做法，可以将这些元素分拆。

以hash为例，原先的正常存取流程是 hget(hashKey, field) ; hset(hashKey, field, value)
现在，固定一个桶的数量，比如 10000， 每次存取的时候，先在本地计算field的hash值，模除 10000， 确定了该field落在哪个key上。

1
2
3

newHashKey  =  hashKey + (*hash*(field) % 10000）;   
hset (newHashKey, field, value) ;  
hget(newHashKey, field)

set, zset, list 也可以类似上述做法.

但有些不适合的场景，比如，要保证 lpop 的数据的确是最早push到list中去的，这个就需要一些附加的属性，或者是在 key的拼接上做一些工作（比如list按照时间来分拆）。

哨兵机制、Redis两种备份方式的区别，项目中用的哪种，为什么？

持久化机制

　为了解决一旦断电或者宕机，内存数据库中的数据将会全部丢失这个缺点，Redis提供了将内存数据持久化到硬盘，以及用持久化文件来恢复数据的功能。Redis 支持两种形式的持久化，一种是RDB快照（snapshotting），另外一种是AOF（append-only-file）。
　
　（1）RDB是把当前内存中的数据集快照写入磁盘，也就是 Snapshot 快照（数据库中所有键值对数据）。恢复时是将快照文件直接读到内存里，RDB持久化有两种触发机制，分别是自动触发和手动触发。在redis.windows.conf(linux就是redis.conf)文件的SNAPSHOTTING 下有个自动触发rdb持久化的策略：
　
　
　其中分别表示900s内，如果至少有一个key值变化，则保存到rdb；300s内，至少有10个key变化就保存；60s内至少有10000个key变化就保存。如果只需要redis的缓存功能那么就可以关掉rdb持久化，使用空串停用，如：save “”“”。

在这个配置下面还有几个关于rdb持久化的配置：

• stop-writes-on-bgsave-error yes：默认yes，表示在后台通过rdb保存数据失败之后是否停止向redis写入数据（接收数据）。这样可以让用户意识到后台持久化失败了，避免后面数据不能持久化。
• rdbcompression yes：默认yes，表示存储的rdb快照是否进行压缩存储，如果关闭则快照会比较大。
• rdbchecksum yes：默认yes，存储之后是否对数据进行校验，若希望提升redis性能可以关闭。
• dbfilename dump.rdb：存储的rdb快照文件名。
• dir ./：设置快照存放路径，必须是目录，默认和当前配置文件在同一目录。

手动触发rdb快照保存可以使用save和bgsave命令。save会阻塞当前redis，redis不能处理其他命令直到rdb过程完成，而bgsave会在后台异步进行保存（redis会执行fork操作创建一个子进程），阻塞只会在fork短时间内，redis内部rdb自动保存都是采用bgsave命令。

将备份文件dump.rdb放到配置文件指定的目录（默认是和redis配置文件同一目录）下，启动redis就会自动将数据加载到内存中。

（2）AOF 持久化是通过保存Redis服务器所执行的写命令来记录数据库状态。在redis的配置文件中APPEND ONLY MODE的下有关于AOF持久化的相关配置。

• appendonly：表示是否开启AOF持久化，因为redis默认的是rdb方式，打开aof需要改为yes。
• appendfilename：aof文件名。
• appendsync：aof持久化策略配置。no表示不执行fsync，有系统保证数据同步到磁盘，速度最快但不安全；always表示每次写入都要执行fsync，以保证数据完全同步到磁盘，效率低；everysec则是每秒保存一次，可能会丢失者1s数据，兼顾安全和效率。
• no-appendfsync-on-rewrite：设置为yes表示rewrite期间对新写操作不fsync,暂时存在内存中,等rewrite完成后再写入，默认为no，建议yes。Linux的默认fsync策略是30秒。可能丢失30秒数据。默认值为no。
• auto-aof-rewrite-percentage：默认值为100。aof自动重写配置，当目前aof文件大小超过上一次重写的aof文件大小的百分之多少进行重写，即当aof文件增长到一定大小的时候，Redis能够调用bgrewriteaof对日志文件进行重写。
• auto-aof-rewrite-min-size：64mb。设置允许重写的最小aof文件大小，避免了达到约定百分比但尺寸仍然很小的情况还要重写。
• aof-load-truncated：aof文件可能在尾部是不完整的（宕机或者断电等），如果选择的是yes，当截断的aof文件被导入的时候，会自动发布一个log给客户端然后load。如果是no，用户必须手动redis-check-aof修复AOF文件才可以。默认值为 yes

AOF文件不完整需要恢复可以使用命令：redis-check-aof –fix 进行修复。aof文件过大的时候也需要重写，和rdb的bgsave模式相似，都是创建子进程，设置重写缓冲区，在重写完成之后再将缓冲区文件写入aof文件。

总结

AOF 持久化的方法提供了多种的同步频率，即使使用默认的同步频率每秒同步一次，Redis 最多也就丢失 1 秒的数据而已。

AOF 文件使用 Redis 命令追加的形式来构造，因此，即使 Redis 只能向 AOF 文件写入命令的片断，使用 redis-check-aof 工具也很容易修正 AOF 文件。

AOF 文件的格式可读性较强，这也为使用者提供了更灵活的处理方式。例如，如果我们不小心错用了 FLUSHALL 命令，在重写还没进行时，我们可以手工将最后的 FLUSHALL 命令去掉，然后再使用 AOF 来恢复数据。

对于具有相同数据的的 Redis，AOF 文件通常会比 RDB文件体积更大。

虽然 AOF 提供了多种同步的频率，默认情况下，每秒同步一次的频率也具有较高的性能。但在 Redis 的负载较高时，RDB 比 AOF 具好更好的性能保证。

RDB 使用快照的形式来持久化整个 Redis 数据，而 AOF 只是将每次执行的命令追加到 AOF 文件中，因此从理论上说，RDB 比 AOF 方式更健壮。官方文档也指出，AOF 的确也存在一些 BUG，这些 BUG 在 RDB 没有存在。

主从复制

复制三份配置文件，分别更改端口号，并且配置文件的名字以端口号区分。

之后再更改每份配置文件：rdb文件名（dbfilename）、log日志文件名（logfile），之后使用下面三个命令启动三个redis服务端。

1
2
3

redis-server.exe redis.windows.conf
redis-server.exe redis.windows-10087.conf 
redis-server.exe redis.windows-10088.conf

通过info replication查看各个节点信息。

此时都是master节点，接着设置slave节点。使用slaveof命令把10087和10088设为slave。

此时设为slave节点之后再使用info replication命令查看就发现role已经变为了slave了。现再在master节点写，在slave读。

可以看见现在主从关系已经成功建立了。注：

（1）如果master以前还存在一些key，那么slave节点也是会有的，因为master会全量复制到slave。

（2）默认从节点是不能够执行写命令的，配置文件中slave-read-only默认是yes。

（3）主节点down掉之后，另外两个slave角色依然不变，并且在master恢复之后，仍然是master并且有两个slave节点。

哨兵模式

哨兵模式就是不时地监控redis是否按照预期良好地运行（至少是保证主节点是存在的），若一台主机出现问题时，哨兵会自动将该主机下的某一个从机设置为新的主机，并让其他从机和新主机建立主从关系（如果监控主机发生故障，就根据投票数自动将从库转化为主库）。首先启动多个redis，并配置主从关系，下面再配置哨兵监控master。

（1）在配置文件目录下面使用touch命令新建sentinel.conf文件，然后配置内容：

sentinel monitor 被监控主机名（自己起） ip地址 端口 得票数 如：sentinel monitor my6379 127.0.0.1 6379 1。1表示当主机挂掉之后得票数>=1便成为主机。

（2）启动哨兵监控

使用命令：./redis-sentinel /mrliu_project/redis/sentinel.conf 注：/mrliu_project/redis/sentinel.conf 是配置文件所在目录

（3）exit退出master主机，之后查看哨兵控制台打印的日志，会发现在重新选择master。

在redis稳定版之后，挂掉的主机在选举之后重新连接上，会设置为新选举的msater的从节点。

启动三台带有redis的服务器133，132，130
分别更改redis.conf文件，并指定配置启动redis-server和redis-sentinel

进入redis的src目录：./redis-server ../redis.config ./redis-sentinel ../sentinel.conf(哨兵配置需要自己touch)

133启动一个master和哨兵的服务器，

132启动两个redis服务(两份配置文件)，分别设为133的slave

130启动一个redis服务，设为133的slave

redis-config更改了一些目录和端口等，我大概改了如下内容：

1
2
3
4
5

daemonize yes
logfile 
port 
 #bind.1 注释掉了
protected-mode no //关闭了

注意（出现设置了slaveof之后发现主机一直未down可能是如下原因）：

（1）每个启动的redis服务需要注释掉bind(代表可以访问的主机)，不然其他redis服务器不能感知到该redis服务的存在

(2)每个redis服务需要更改protected-mode 为no，

（3）每台linux服务器需要打开端口，不然直接关闭firewall或者是iptables也行，命令systemctl stop firewalld和service iptables stop。 永久关闭防火墙使用disable。

（4）建议redis-server都配置为守护进程，daemon设为yes即可。

哨兵需要自建配置文件，在redis的src目录下使用 ./redis-sentinel ../sentinel_liu.conf 启动，其中后面的是自己的哨兵配置文件，简易配置如下:

1
2
3
4
5

port 26379
 
sentinel monitor liu_master 192.168.15.130 6379 1//监控的主机，后面代表超过机票就成为leader
sentinel down-after-milliseconds liu_master 10000//可选，多久没心跳就认为down
sentinel failover-timeout liu_master 10000//可选，代表每次选举间隔

几台服务器如下:

之后进入133使用info确认从节点有三个，

之后查看哨兵启动日志

注：sdown代表哨兵主观任务下线，（后面是master断开之后的日志）odown是客观下线，当出现odown的时候，哨兵将不会监控该服务，任务确实下线了，然后开始选举。

断开133的master，观察哨兵日志
可以进入133的客户端，使用shutdown关掉，也可以在redis的src目录下使用./redis-cli shutdown关，也可以杀redis进程关。

哨兵日志如下（我这儿选了这么多次原因是其他机器防火墙开着，所以关了就选举成功了）:

出现failover-end即是选举成功了，switch代表切换master到了130服务器节点，随后发现 +fix-slave-config日志，代表将132的6379redis服务和132的6380redis服务设为了130的slave节点，这时候130成为master，132的两个redis服务是slave节点。

重新上线之前断开的133master，查看哨兵日志

发现新上线的133设为了130redis服务的slave，下面去130服务器登上redis客户端输入info验证

发现回归后的master变为了新选举的master的slave节点。