Flink 原理与实现: Flink流计算常用算子（Flink算子大全）

2022-02-15

字数统计: 5.3k字 | 阅读时长≈ 22分

Flink和Spark类似，也是一种一站式处理的框架；既可以进行批处理（DataSet），也可以进行实时处理（DataStream）。

所以下面将Flink的算子分为两大类：一类是DataSet，一类是DataStream。

Flink官网: https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/datastream/operators/overview/

案例代码: https://github.com/ShangBaiShuYao/flink-learning-from-zhisheng/tree/main/flink-1.12.0-Demo/src/main/java/com/shangbaishuyao/demo

我们列举了一些Flink自带且常用的transformation算子，例如map、flatMap等。在Flink的编程体系中，我们获取到数据源之后，需要经过一系列的处理即transformation操作，再将最终结果输出到目的Sink（ES、mysql或者hdfs），使数据落地。因此，除了正确的继承重写RichSourceFunction<>和RichSinkFunction<>之外，最终要的就是实时处理这部分，下面的图介绍了Flink代码执行流程以及各模块的组成部分。

在Flink中，Transformation算子就是将一个或多个DataStream转换为新的DataStream，可以将多个转换组合成复杂的数据流拓扑。如下图所示，DataStream会由不同的Transformation操作，转换、过滤、聚合成其他不同的流，从而完成我们的业务要求。

DataSet

一、Source算子

1. fromCollection

fromCollection：从本地集合读取数据

例：

val env = ExecutionEnvironment.getExecutionEnvironment
val textDataSet: DataSet[String] = env.fromCollection(
  List("1,张三", "2,李四", "3,王五", "4,赵六")
)

2. readTextFile

readTextFile：从文件中读取

1	val textDataSet: DataSet[String] = env.readTextFile("/data/a.txt")

3. readTextFile：遍历目录

readTextFile可以对一个文件目录内的所有文件，包括所有子目录中的所有文件的遍历访问方式

val parameters = new Configuration
// recursive.file.enumeration 开启递归
parameters.setBoolean("recursive.file.enumeration", true)
val file = env.readTextFile("/data").withParameters(parameters)

4. readTextFile：读取压缩文件

对于以下压缩类型，不需要指定任何额外的inputformat方法，flink可以自动识别并且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。

1	val file = env.readTextFile("/data/file.gz")

二、Transform转换算子

因为Transform算子基于Source算子操作，所以首先构建Flink执行环境及Source算子，后续Transform算子操作基于此：

val env = ExecutionEnvironment.getExecutionEnvironment
val textDataSet: DataSet[String] = env.fromCollection(
  List("张三,1", "李四,2", "王五,3", "张三,4")
)

1. map

将DataSet中的每一个元素转换为另外一个元素

// 使用map将List转换为一个Scala的样例类

case class User(name: String, id: String)

val userDataSet: DataSet[User] = textDataSet.map {
  text =>
    val fieldArr = text.split(",")
    User(fieldArr(0), fieldArr(1))
}
userDataSet.print()

2. flatMap

将DataSet中的每一个元素转换为0…n个元素。

// 使用flatMap操作，将集合中的数据：
// 根据第一个元素，进行分组
// 根据第二个元素，进行聚合求值 

val result = textDataSet.flatMap(line => line)
      .groupBy(0) // 根据第一个元素，进行分组
      .sum(1) // 根据第二个元素，进行聚合求值
      
result.print()

3. mapPartition

将一个分区中的元素转换为另一个元素

// 使用mapPartition操作，将List转换为一个scala的样例类

case class User(name: String, id: String)

val result: DataSet[User] = textDataSet.mapPartition(line => {
      line.map(index => User(index._1, index._2))
    })
    
result.print()

4. filter

过滤出来一些符合条件的元素，返回boolean值为true的元素

1
2
3

val source: DataSet[String] = env.fromElements("java", "scala", "java")
val filter:DataSet[String] = source.filter(line => line.contains("java"))//过滤出带java的数据
filter.print()

5. reduce

可以对一个dataset或者一个group来进行聚合计算，最终聚合成一个元素

// 使用 fromElements 构建数据源
val source = env.fromElements(("java", 1), ("scala", 1), ("java", 1))
// 使用map转换成DataSet元组
val mapData: DataSet[(String, Int)] = source.map(line => line)
// 根据首个元素分组
val groupData = mapData.groupBy(_._1)
// 使用reduce聚合
val reduceData = groupData.reduce((x, y) => (x._1, x._2 + y._2))
// 打印测试
reduceData.print()

6. reduceGroup

将一个dataset或者一个group聚合成一个或多个元素。
reduceGroup是reduce的一种优化方案；
它会先分组reduce，然后在做整体的reduce；这样做的好处就是可以减少网络IO

// 使用 fromElements 构建数据源
val source: DataSet[(String, Int)] = env.fromElements(("java", 1), ("scala", 1), ("java", 1))
// 根据首个元素分组
val groupData = source.groupBy(_._1)
// 使用reduceGroup聚合
val result: DataSet[(String, Int)] = groupData.reduceGroup {
      (in: Iterator[(String, Int)], out: Collector[(String, Int)]) =>
        val tuple = in.reduce((x, y) => (x._1, x._2 + y._2))
        out.collect(tuple)
    }
// 打印测试
result.print()

7. minBy和maxBy

选择具有最小值或最大值的元素

// 使用minBy操作，求List中每个人的最小值
// List("张三,1", "李四,2", "王五,3", "张三,4")

case class User(name: String, id: String)
// 将List转换为一个scala的样例类
val text: DataSet[User] = textDataSet.mapPartition(line => {
      line.map(index => User(index._1, index._2))
    })
    
val result = text
          .groupBy(0) // 按照姓名分组
          .minBy(1)   // 每个人的最小值

8. Aggregate

在数据集上进行聚合求最值（最大值、最小值）

val data = new mutable.MutableList[(Int, String, Double)]
    data.+=((1, "yuwen", 89.0))
    data.+=((2, "shuxue", 92.2))
    data.+=((3, "yuwen", 89.99))
// 使用 fromElements 构建数据源
val input: DataSet[(Int, String, Double)] = env.fromCollection(data)
// 使用group执行分组操作
val value = input.groupBy(1)
            // 使用aggregate求最大值元素
            .aggregate(Aggregations.MAX, 2) 
// 打印测试
value.print()

Aggregate只能作用于元组上

注意：
要使用aggregate，只能使用字段索引名或索引名称来进行分组 groupBy(0) ，否则会报一下错误:
Exception in thread “main” java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.

9. distinct

去除重复的数据

// 数据源使用上一题的
// 使用distinct操作，根据科目去除集合中重复的元组数据

val value: DataSet[(Int, String, Double)] = input.distinct(1)
value.print()

10. first

取前N个数

1	input.first(2) // 取前两个数

11. join

将两个DataSet按照一定条件连接到一起，形成新的DataSet

// s1 和 s2 数据集格式如下：
// DataSet[(Int, String,String, Double)]

 val joinData = s1.join(s2)  // s1数据集 join s2数据集
             .where(0).equalTo(0) {     // join的条件
      (s1, s2) => (s1._1, s1._2, s2._2, s1._3)
    }

12. leftOuterJoin

左外连接,左边的Dataset中的每一个元素，去连接右边的元素

此外还有：

rightOuterJoin：右外连接,左边的Dataset中的每一个元素，去连接左边的元素

fullOuterJoin：全外连接,左右两边的元素，全部连接

下面以 leftOuterJoin 进行示例：

 val data1 = ListBuffer[Tuple2[Int,String]]()
    data1.append((1,"zhangsan"))
    data1.append((2,"lisi"))
    data1.append((3,"wangwu"))
    data1.append((4,"zhaoliu"))

val data2 = ListBuffer[Tuple2[Int,String]]()
    data2.append((1,"beijing"))
    data2.append((2,"shanghai"))
    data2.append((4,"guangzhou"))

val text1 = env.fromCollection(data1)
val text2 = env.fromCollection(data2)

text1.leftOuterJoin(text2).where(0).equalTo(0).apply((first,second)=>{
      if(second==null){
        (first._1,first._2,"null")
      }else{
        (first._1,first._2,second._2)
      }
    }).print()

13. cross

交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集

和join类似，但是这种交叉操作会产生笛卡尔积，在数据比较大的时候，是非常消耗内存的操作

val cross = input1.cross(input2){
      (input1 , input2) => (input1._1,input1._2,input1._3,input2._2)
    }

cross.print()

14. union

联合操作，创建包含来自该数据集和其他数据集的元素的新数据集,不会去重

1
2
3

val unionData: DataSet[String] = elements1.union(elements2).union(elements3)
// 去除重复数据
val value = unionData.distinct(line => line)

15. rebalance

Flink也有数据倾斜的时候，比如当前有数据量大概10亿条数据需要处理，在处理过程中可能会发生如图所示的状况：

这个时候本来总体数据量只需要10分钟解决的问题，出现了数据倾斜，机器1上的任务需要4个小时才能完成，那么其他3台机器执行完毕也要等待机器1执行完毕后才算整体将任务完成；所以在实际的工作中，出现这种情况比较好的解决方案就是接下来要介绍的—rebalance（内部使用round robin方法将数据均匀打散。这对于数据倾斜时是很好的选择。）

1 2	// 使用rebalance操作，避免数据倾斜 val rebalance = filterData.rebalance()

16. partitionByHash

按照指定的key进行hash分区

val data = new mutable.MutableList[(Int, Long, String)]
data.+=((1, 1L, "Hi"))
data.+=((2, 2L, "Hello"))
data.+=((3, 2L, "Hello world"))

val collection = env.fromCollection(data)
val unique = collection.partitionByHash(1).mapPartition{
  line =>
    line.map(x => (x._1 , x._2 , x._3))
}

unique.writeAsText("hashPartition", WriteMode.NO_OVERWRITE)
env.execute()

17. partitionByRange

根据指定的key对数据集进行范围分区

val data = new mutable.MutableList[(Int, Long, String)]
data.+=((1, 1L, "Hi"))
data.+=((2, 2L, "Hello"))
data.+=((3, 2L, "Hello world"))
data.+=((4, 3L, "Hello world, how are you?"))

val collection = env.fromCollection(data)
val unique = collection.partitionByRange(x => x._1).mapPartition(line => line.map{
  x=>
    (x._1 , x._2 , x._3)
})
unique.writeAsText("rangePartition", WriteMode.OVERWRITE)
env.execute()

18. sortPartition

根据指定的字段值进行分区的排序

val data = new mutable.MutableList[(Int, Long, String)]
    data.+=((1, 1L, "Hi"))
    data.+=((2, 2L, "Hello"))
    data.+=((3, 2L, "Hello world"))
    data.+=((4, 3L, "Hello world, how are you?"))

val ds = env.fromCollection(data)
    val result = ds
      .map { x => x }.setParallelism(2)
      .sortPartition(1, Order.DESCENDING)//第一个参数代表按照哪个字段进行分区
      .mapPartition(line => line)
      .collect()

println(result)

三、Sink算子

1. collect

将数据输出到本地集合

1	result.collect()

2. writeAsText

将数据输出到文件

Flink支持多种存储设备上的文件，包括本地文件，hdfs文件等

Flink支持多种文件的存储格式，包括text文件，CSV文件等

// 将数据写入本地文件
result.writeAsText("/data/a", WriteMode.OVERWRITE)

// 将数据写入HDFS
result.writeAsText("hdfs://node01:9000/data/a", WriteMode.OVERWRITE)

DataStream

和DataSet一样，DataStream也包括一系列的Transformation操作

一、Source算子

Flink可以使用 StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源。
Flink 已经提供了若干实现好了的 source functions，当然我们也可以通过实现 SourceFunction 来自定义非并行的source或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source。

Flink在流处理上的source和在批处理上的source基本一致。大致有4大类：

基于本地集合的source（Collection-based-source）
基于文件的source（File-based-source）- 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回
基于网络套接字的source（Socket-based-source）- 从 socket 读取。元素可以用分隔符切分。
自定义的source（Custom-source）

下面使用addSource将Kafka数据写入Flink为例：

如果需要外部数据源对接，可使用addSource，如将Kafka数据写入Flink，先引入依赖：

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11 -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
    <version>1.10.0</version>
</dependency>

将Kafka数据写入Flink：

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("group.id", "consumer-group")
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("auto.offset.reset", "latest")

val source = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

基于网络套接字的：

1	val source = env.socketTextStream("IP", PORT)

二、Transform转换算子

1. map

将DataSet中的每一个元素转换为另外一个元素

1	dataStream.map { x => x * 2 }

2. FlatMap

采用一个数据元并生成零个，一个或多个数据元。将句子分割为单词的flatmap函数

1	dataStream.flatMap { str => str.split(" ") }

3. Filter

计算每个数据元的布尔函数，并保存函数返回true的数据元。过滤掉零值的过滤器

1	dataStream.filter { _ != 0 }

4. KeyBy

逻辑上将流分区为不相交的分区。具有相同Keys的所有记录都分配给同一分区。在内部，keyBy（）是使用散列分区实现的。指定键有不同的方法。

此转换返回KeyedStream，其中包括使用被Keys化状态所需的KeyedStream。

1	dataStream.keyBy(0)

注意:

这个keyBy,你给我一个dataStream,他调用keyBy之后得到一个KeyedStream,这个KeyedStream就相当于是Spark里面的,包含键值对的RDD.说白了就是元素是二元组的那个RDD.不过在我们Flink里面,没有所谓二元组的说法.不一定说二元组就是键值对.在Flink中没有这个说法.只有在Spark RDD中才有这个说法.意思就是说:你看到一个Dstream里面的元素是二元组类型,千万别认为这就是一个键值对的Dstream.这是一个非常普通的Dstream.除非他不是Dstream,而是一个KeyedStream.所以在Flink中到底判断他是不是一个键值对的Stream,就看他类型是不是KeyedStream,跟里面的元素没有关系.

5. 滚动聚合算子（Rolling Aggregation）

Aggregations 是 KeyedDataStream 接口提供的聚合算子，根据指定的字段进行聚合操作，滚动地产生一系列数据聚合结果。其实是将 Reduce 算子中的函数进行了封装，封装的聚合操作有sum,min,max 等，这样就不需要用户自己定义 Reduce 函数。
如下代码所示，指定数据集中第一个字段作为 key，用第二个字段作为累加字段，然后滚动地对第二个字段的数值进行累加并输出。

这些算子可以针对KeyedStream的每一个支流做聚合。

sum()
min()
max()
minBy()
maxBy()

6. Reduce

被Keys化数据流上的“滚动”Reduce。将当前数据元与最后一个Reduce的值组合并发出新值

1	keyedStream.reduce { _ + _ }

KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。

val stream2 = env.readTextFile("YOUR_PATH\\sensor.txt")
  .map( data => {
    val dataArray = data.split(",")
    SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
  })
  .keyBy("id")
  .reduce( (x, y) => SensorReading(x.id, x.timestamp + 1, y.temperature) )

7. Fold

具有初始值的被Keys化数据流上的“滚动”折叠。将当前数据元与最后折叠的值组合并发出新值

1
2
3

val result: DataStream[String] =  keyedStream.fold("start")((str, i) => { str + "-" + i }) 

// 解释：当上述代码应用于序列（1,2,3,4,5）时，输出结果“start-1”，“start-1-2”，“start-1-2-3”，...

8. Aggregations

在被Keys化数据流上滚动聚合。min和minBy之间的差异是min返回最小值，而minBy返回该字段中具有最小值的数据元（max和maxBy相同）。

keyedStream.sum(0);

keyedStream.min(0);

keyedStream.max(0);

keyedStream.minBy(0);

keyedStream.maxBy(0);

9. Window

可以在已经分区的KeyedStream上定义Windows。Windows根据某些特征（例如，在最后5秒内到达的数据）对每个Keys中的数据进行分组。这里不再对窗口进行详解，有关窗口的完整说明，请查看这篇文章： Flink 中极其重要的 Time 与 Window 详细解析

1	dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));

10. WindowAll

Windows可以在常规DataStream上定义。Windows根据某些特征（例如，在最后5秒内到达的数据）对所有流事件进行分组。

注意：在许多情况下，这是非并行转换。所有记录将收集在windowAll 算子的一个任务中。

1	dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))

11. Window Apply

将一般函数应用于整个窗口。

注意：如果您正在使用windowAll转换，则需要使用AllWindowFunction。

下面是一个手动求和窗口数据元的函数

1
2
3

windowedStream.apply { WindowFunction }

allWindowedStream.apply { AllWindowFunction }

12. Window Reduce

将函数缩减函数应用于窗口并返回缩小的值

1	windowedStream.reduce { _ + _ }

13. Window Fold

将函数折叠函数应用于窗口并返回折叠值

1
2
3

val result: DataStream[String] = windowedStream.fold("start", (str, i) => { str + "-" + i }) 

// 上述代码应用于序列（1,2,3,4,5）时，将序列折叠为字符串“start-1-2-3-4-5”

14. Union(真正意义上的汇合)

两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次数据元

1	dataStream.union(otherStream1, otherStream2, ...)

DataStream → DataStream：对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream。

Connect与 Union 区别：

1． Union之前两个流的类型必须是一样，Connect可以不一样，在之后的coMap中再去调整成为一样的。

Connect只能操作两个流，Union可以操作多个。

15. Window Join

在给定Keys和公共窗口上连接两个数据流

dataStream.join(otherStream)
    .where(<key selector>).equalTo(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...})

16. Interval Join

在给定的时间间隔内使用公共Keys关联两个被Key化的数据流的两个数据元e1和e2，以便e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound

am.intervalJoin(otherKeyedStream)
    .between(Time.milliseconds(-2), Time.milliseconds(2)) 
    .upperBoundExclusive(true) 
    .lowerBoundExclusive(true) 
    .process(new IntervalJoinFunction() {...})

案例代码:

https://github.com/ShangBaiShuYao/flink-learning-from-zhisheng/blob/main/Flink-shangbaishuyao-RTDW-gmall-realtime/src/main/java/com/shangbaishuyao/gmall/realtime/app/DWM/OrderWideApp.java

17. Window CoGroup

在给定Keys和公共窗口上对两个数据流进行Cogroup

dataStream.coGroup(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new CoGroupFunction () {...})

18. Connect

“连接”两个保存其类型的数据流。连接允许两个流之间的共享状态

1
2
3

DataStream<Integer> someStream = ... DataStream<String> otherStream = ... ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream)

// ... 代表省略中间操作

19. CoMap，CoFlatMap

Connect做连接的,把两个流连在一起.也可以理解为做回合,把两个流汇合在一起.

Connect把两个流汇合在一起有条件吗?stream1是整形,stream2是字符串的这样的两个流可以汇合在一起吗?他是可以的.他不管你两个流是否类型一致.他都可以汇合在一起.connect会和其实和上面的splitStream一样,都没有被真正的汇合或切分.

类似于连接数据流上的map和flatMap

connectedStreams.map(
    (_ : Int) => true,
    (_ : String) => false)connectedStreams.flatMap(
    (_ : Int) => true,
    (_ : String) => false)

DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被Connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

CoMap,CoFlatMap

从这张图可以看出,他只有调用map或者FlatMap才真正将两个流汇合起来的.
为什么要执行map或者FlatMap呢?因为执行map算子或者flatmap算子之后他就可以在map算子或者FlatMap算子中把这两个流分别做处理.分别处理成同样类型的这时返回得到一个新的对象,他就不会出现两个流了,他就变成一个流了.

ConnectedStreams → DataStream：作用于ConnectedStreams上，功能与map和flatMap一样，对ConnectedStreams中的每一个Stream分别进行map和flatMap处理。

20. Split(不是真正意义上切分流,只是打了标记)

根据某些标准将流拆分为两个或更多个流

val split = someDataStream.split(
  (num: Int) =>
    (num % 2) match {
      case 0 => List("even")
      case 1 => List("odd")
    })

21. Select(真正意义上的切分流)

从拆分流中选择一个或多个流

1	SplitStream<Integer> split;DataStream<Integer> even = split.select("even");DataStream<Integer> odd = split.select("odd");DataStream<Integer> all = split.select("even","odd")

Split 和 Select:

DataStream → SplitStream：根据某些特征把一个DataStream拆分成两个或者多个DataStream。

SplitStream→DataStream：从一个SplitStream中获取一个或者多个DataStream。

需求：传感器数据按照温度高低（以30度为界），拆分成两个流。

val splitStream = stream2
  .split( sensorData => {
    if (sensorData.temperature > 30) Seq("high") else Seq("low")
  } )

val high = splitStream.select("high")
val low = splitStream.select("low")
val all = splitStream.select("high", "low")

三、Sink算子

支持将数据输出到：

本地文件(参考批处理)
本地集合(参考批处理)
HDFS(参考批处理)

除此之外，还支持：

sink到kafka
sink到mysql
sink到redis

下面以sink到kafka为例：

val sinkTopic = "test"

//样例类
case class Student(id: Int, name: String, addr: String, sex: String)
val mapper: ObjectMapper = new ObjectMapper()

//将对象转换成字符串
def toJsonString(T: Object): String = {
    mapper.registerModule(DefaultScalaModule)
    mapper.writeValueAsString(T)
}

def main(args: Array[String]): Unit = {
    //1.创建流执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.准备数据
    val dataStream: DataStream[Student] = env.fromElements(
      Student(8, "xiaoming", "beijing biejing", "female")
    )
    //将student转换成字符串
    val studentStream: DataStream[String] = dataStream.map(student =>
      toJsonString(student) // 这里需要显示SerializerFeature中的某一个，否则会报同时匹配两个方法的错误
    )
    //studentStream.print()
    val prop = new Properties()
    prop.setProperty("bootstrap.servers", "node01:9092")

    val myProducer = new FlinkKafkaProducer011[String](sinkTopic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), prop)
    studentStream.addSink(myProducer)
    studentStream.print()
    env.execute("Flink add sink")
}

知识源于积累,登峰造极源于自律!

好文章就得收藏慢慢品, 文章转载于:https://zhuanlan.zhihu.com/p/356616078 在此基础上做的CRUD.

本文作者： xubatian
本文链接： http://xubatian.cn/Flink-原理与实现-Flink流计算常用算子（Flink算子大全）/
版权声明： 本博客所有文章除特别声明外均为原创，采用 CC BY 4.0 CN协议许可协议。转载请注明出处:https://www.xubatian.cn/