作者rechen2020·2023-04-27 10:37

系统架构师·某大型银行

银行行业Kafka容器化服务测试报告

字数 14045阅读 6241评论 2赞 10

内容摘要：

Kafka在金融银行应用广泛，容器化部署可提高敏捷交付成功率、弹性伸缩能力、业务迁移扩展等。存储选型需考虑高性能、高可用、高扩展性。测试验证了多集群多副本架构、华为企业级NAS存储与OpenShift容器云平台的组合，实现了高可用Kafka集群实例在容器云平台节点上的独立部署，验证了卷在线扩容、节点在线扩容、节点再均衡等功能，以及节点故障的集群快速自愈能力和单副本下的性能提升。

当前摘要为AI生成

# 1 背景和需求
银行业务应用系统每天有大量的交易发生，这些交易很多需要进行实时的业务核查与处理，譬如一笔交易发生，这笔交易是否是欺诈交易，这笔交易是否已经达到限额了，相关用户是否在黑名单中，交易结果联动短信、Email渠道通知用户等等，典型的一笔简单的资金转账交易，就要与反欺诈系统、黑名单系统、限额或授信管控系统、短信邮件系统进行通讯。另外，银行业务应用系统在投产后的运行时，如果出现了问题，需要采集日志及时做故障定位；为了及时发现问题，要将处理结果实时发送到监控平台；为了预防突发事件发生，还需要备份系统做高可用处理。这就需要业务应用系统与运维体系中的日志系统、监控系统、备份系统进行数据交互。这些处理须在极短时间内完成，否则用户体验就会受损和卡顿。

解决这类问题的成熟技术方案是在各个银行业务应用系统之间建设一个消息分发的总控系统，连接总控系统有2种角色：
- 消息供应方（Producer）：产生消息，并将消息送给消息分发系统
- 消息消费方（Consumer）：从消息分发系统获取消息进行功能处理。
消息分发系统起到消息中介（Message Broker）的作用。

## 1.1 Kafka简介
Kafka开源项目最初由Linkedin公司开发，是业界在消息分发领域的成熟技术方案，在金融银行有广泛的应用场景。Kafka是基于Zookeeper协调的分布式架构，是一个企业级的的消息分发处理系统，可以很好地进行多个业务应用系统间的消息处理事务，同时支持实时的流处理，其设计目标为：

- 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
- 高吞吐率，即使在低配的商用机器上也能做到单机支持每秒100K条消息的传输。
- 支持Kafka Server间的消息分区及分布式消费，保证每个partition内的消息顺序传输。
- 支持离线数据处理和实时数据处理。
- Scale out：支持在线水平扩展。

**Kafka功能架构图如下所示：**
![](/home/attachment/202304/27/1241241_168256063226718.png)

一个典型的 Kafka 集群中包含若干 Producer，若干 broker（注：Kafka 支持水平扩展，一般 broker 数量越多，集群吞吐率越高），若干 Consumer Group，以及一个 Zookeeper 集群。Kafka 通过 Zookeeper 管理集群配置，选举Leader，以及在 Consumer Group 发生变化时进行数据平衡。Producer 使用 push 模式将消息发布到 broker，Consumer 使用 pull 模式从 broker 订阅并消费消息。

对Kafka服务的使用方式，有作为消息队列和数据处理二类。Kafka在金融行业的具体应用场景分为如下4种：

1.       日志处理场景：使用Kafka收集各种业务系统的日志数据，通过Kafka以统一接口服务的方式开放给下游系统消费，进行实时处理和分析。

2.       用户数据采集和分发场景：譬如在建设用户活动跟踪平台时，使用Kafka记录Web用户或者App用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个业务系统服务器发布到Kafka的Topic主题中，然后下游消费者通过订阅这些Topic主题来做实时的监控分析，亦可保存到数据库。

3.       大数据实时计算场景：Kafka作为实时数据的输入源，通过流处理框架如 Apache Spark 和 Apache Flink 进行实时分析和计算。

4.       应用程序事件聚合场景：Kafka可以作为多个应用程序间事件的聚合平台，通过主题将事件数据从不同的应用程序分发到消费者，以解耦生产者和消费者、缓存消息等。

## 1.2 Kafka容器化部署需求
Kafka的部署架构主要有二种，一种是虚拟机+云硬盘，另一种是物理机+本地盘，高可用设计采用主备集群模式。但是随着业务系统的云原生转型过程中，特别系统分级保障和资源精细化治理工作，Kafka集群管理也遇到了敏捷交付成功率有待提高、弹性伸缩能力不足、资源使用率低、高可用保障有待优化等新的挑战。为此，银行需通过对Kafka的发展趋势展开调研，参考公有云厂商的成熟经验，探索采用新的技术架构解决问题。

银行通过调研，发现业界出现了Kafka容器化部署的趋势，具体包括：
- 容器镜像天然支持标准化，与硬件无关，能解决Kafka运维的版本问题。
- 在Kubernetes环境可通过 StatefulSet 部署Kafka集群，使用 configmap 管理配置文件，新集群部署时间快只需要几分钟，可以有效提升敏捷交付成功率;
- 在Kubernetes环境中使用Operator的方式部署Kafka集群还能提高部署密度和提升弹性伸缩能力。

Kafka 容器化部署的优点非常多，具体有如下优势：
−        轻量部署，节约服务器物理资源。
−        秒级拉起业务，可以实现快速交付和部署。
−        迁移扩展，结合企业级NAS存储，可以实现业务快速迁移，集群横向和纵向快速扩容。
−        业务隔离，一台主机运行不同业务，资源隔离，互不影响。
−        故障自愈，利用K8S平台的故障恢复机制，通过POD重启、漂移等方式快速规避故障，将故障对业务的影响降低到最小。

## 1.3 Kafka容器化存储选型思考
Kafka容器化部署的落地实施过程中，需要面临的一个核心问题是做好Kafka存储的技术选型。Kafka的存储需求，在功能面是存储的数据是消息流，数据量级非常大；数据CRUD操作足够简单，主要为并发非常高和百万级TPS的追加写、无需更改、能根据消费位移offset和时间戳timestamp查询消息、能定期删除过期的消息；在非功能性面是高性能要求、高可用的稳定性要求、高扩展性要求。

Kafka的存储是「分区 + 分段 + 索引」的三层结构，其设计原则是Append 追加写日志 + 稀疏的哈希索引，如下图所示：
![](/home/attachment/202304/27/1241241_168256080620844.png)

其中：每个 Topic 被分成多个 Partition，Partition 从物理上可以理解成一个文件夹。每个 Partition 又被分成了多个 Segment，Segment 从物理上可以理解成一个「数据文件 + 索引文件」。在做历史消息删除时，只需将旧的 Segment 文件删除即可，保证了每个 Segment 的顺序写。

在存储的选择过程中，我们对比了服务器本地盘、外置SAN存储和外置NAS存储三种选择。
- 服务器本地盘性能较好，但存在故障率高、故障后恢复慢、不同节点容量使用不均衡、容量使用率低、扩容难等问题；而且容器化部署后，使用本地盘会限制POD漂移能力，影响组件的高可用。
- 在SAN和NAS的选择上，考虑到多个Kafka集群间需要做数据隔离；所以我们选择了具备多租户能力的高性能全闪存NAS来承载Kafka数据持久化需求。不同NAS租户对接不同的Kafka集群，实现多集群共享存储资源池。
- 在Kafka集群节点故障时，POD容器漂移后可直接在目的端重新挂载存储，因存储数据全局共享，不需要拷贝数据，恢复时长降低至分钟级别，可快速实现故障自动切换，满足高可用性的稳定性需求。

本文就Kafka容器化部署改造的可行性验证工作中，介绍一种基于K8S+NAS存储的技术架构的高可靠、高性能的Kafka容器化集群的测试实践和经验。

# 2 测试方案设计
测试目标：为了模拟真实生产环境，我们分别验证了单集群/双集群多broker、多副本的架构，Kafka 集群的架构, 测试硬件资源和软件版本与生产环境尽量保持一致。

## 2.1测试硬件资源清单
![](/home/attachment/202304/27/1241241_168256088246378.png)

**表2-1 测试资源清单**
![](/home/attachment/202304/27/1241241_168256091256207.png)

## 2.2  测试软件版本清单
![](/home/attachment/202304/27/1241241_168256097580629.png)

# 3 Kafka容器集群部署
## 3.1 单集群部署

1.在OKD中创建一个名为kafka的namespace，将Kafka的资源与其他业务隔离
kubectl create namespace kafka

2.编写Kafka Operator的yaml文件，拉起Operator镜像，定义自定义资源对象。具体内容可以参考官网https://strimzi.io/quickstarts/。
![](/home/attachment/202304/27/1241241_168256104763554.png)
3.编写Zookeeper和Kafka集群的yaml文件，定义端口、PVC资源、CPU和内存资源、副本数等，拉起对应的POD
```asp
[root@bastion kafka_ope]# cat kafka-cr.yaml 
apiVersion: kafka.strimzi.io/v1beta2 
kind: Kafka 
metadata: 
  name: my-cluster 
  namespace: kafka 
spec: 
  kafka: 
    version: 3.2.3 
    replicas: 3 
    listeners: 
      ...... 
    config: 
      ...... 
    resources: 
      ...... 
    template: 
      ...... 
 
    storage: 
      type: persistent-claim  #指定存储资源类型 
      class: scnodedup  #指定StorageClass 
      size: 100Gi  #指定创建PVC的容量大小     
      deleteClaim: false  #指定删除Pod时是否一起删除对应的PVC 
    ...... 
  zookeeper: 
    replicas: 1 
    resources: 
      ...... 
    template: 
      ...... 
    storage: 
      type: persistent-claim  #指定存储资源类型 
      class: scnodedup  #指定StorageClass 
      size: 10Gi  #指定创建PVC的容量大小 
      deleteClaim: false  #指定删除Pod时是否一起删除对应的PVC 
  entityOperator: {}

[root@bastion kafka_ope]# oc apply -f kafka_cluster_multi.yaml -n kafka

```
![](/home/attachment/202304/27/1241241_168256121236895.png)
4.集群创建成功。

## 3.2 双集群部署

在OKD中创建两个namespace分别为kafka和kafka1；其他操作同上。
集群均创建成功。

# 4 测试过程
## 4.1 相关功能测试
**4.1.1 验证Kafka集群Broker数量扩容功能**
Broker数量扩容功能主要用于Broker处理能力不足时增加新的Broker；扩容后，用户可以根据需要决定是否在多个Broker之间进行再均衡。

1.为Kafka集群扩容3个Broker，检查是否扩容成功；
修改cr的yaml文件，将副本数改为6后执行
oc apply -f  kafka-cr.yaml -n kafka
查看pod状态，已成功创建。

![](/home/attachment/202304/27/1241241_168256136237040.png)
2.检查确认扩容过程中业务未中断；
![](/home/attachment/202304/27/1241241_168256139036987.png)
3.检查存储上新增的PVC文件系统中没有数据；
![](/home/attachment/202304/27/1241241_168256143720002.png)
4.执行数据均衡操作，记录数据均衡时间；
创建一个kafka-balance 容器
```asp
[root@bastion balance]# oc -n kafka run kafka-balance -ti --
image=registry.example.com:8443/kafka/kafka:0.31.1-kafka-3.2.3  --
restart=Never /bin/bash
```
创建一个topic-to-move.json文件，指定需要均衡的topic
```asp
[root@bastion balance]# cat topic-to-move.json 
{ 
    "topics": [ 
        {"topic": "mytopic"} 
    ], 
    "version":1 
}

```
将topic-to-move.json拷贝到kafka-balance 容器中的/tmp目录下:
```asp
[root@bastion balance]# oc cp -n kafka topic-to-move.json kafka-balance:/tmp
```
在kafka-balance容器中执行以下命令，得到需要均衡后分区布局
```asp
[kafka@kafka-balance kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topics-to-move-json-file /tmp/topic-to-move.json --broker-list "0,1,2,3,4,5" --generate
```
![](/home/attachment/202304/27/1241241_168256163633150.png)
创建一个new.json文件，将上面生成分区布局拷贝到该文件中
```asp
[root@bastion balance]# cat new.json 
{"version":1,"partitions":[{"topic":"mytopic","partition":0,"replicas":[4,5,0],"log_dirs":["any","any","any"]},{"topic":"mytopic","partition":1,"replicas":[5,0,1],"log_dirs":["any","any","any"]},{"topic":"mytopic","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"mytopic","partition":3,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"mytopic","partition":4,"replicas":[2,3,4],"log_dirs":["any","any","any"]},{"topic":"mytopic","partition":5,"replicas":[3,4,5],"log_dirs":["any","any","any"]}]}

```
将new.json文件拷贝到kafka-balance容器中的/tmp目录下:
```asp
[root@bastion balance]# oc cp -n kafka new.json kafka-balance:/tmp
```
执行以下命令开始均衡数据到所有broker上
```asp
[kafka@kafka-balance kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --reassignment-json-file /tmp/new.json --execute
```
![](/home/attachment/202304/27/1241241_168256172656071.png)
执行以下命令查看均衡情况:
```asp
[kafka@kafka-balance kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --reassignment-json-file /tmp/new.json --verify
```
![](/home/attachment/202304/27/1241241_168256176770887.png)

5.60G数据量由3个broker均衡到6个broker上，耗时约1分50秒。数据均衡会触发Broker和 Partition的对应关系发生变化，导致业务中断约40s，建议在非核心时间执行该操作。
**注意**
1、在使用Strimzi Operator时，Kafka的集群可以使用K8S自带的资源控制器statefulset和Operator自定义的资源控制器UseStrimziPodSets两种方式部署。如果采用statefulset部署，则在增加broker数量的过程中，会滚动更新，此时会对业务造成影响；如果采用UseStrimziPodSets方式部署Kafka集群，可以直接扩容，不会重启前面的Pod。具体情况参考：https://github.com/strimzi/strimzi-kafka-operator/issues/4913。

2、在执行数据均衡时，两种部署方式都会造成业务中断，因此建议在无业务的情况下执行数据均衡。

**4.1.2 验证Kafka集群下的卷扩容功能**
卷扩容功能可在持久化卷（PV）容量不足时实现快速在线扩容，业务无感知。
由于单服务器的存储扩容能力有限，在使用本地盘作为持久化存储时，常会出现由于盘容量限制导致无法扩容的情况；使用外置存储则可以按需轻松扩容。
![](/home/attachment/202304/27/1241241_168256185227556.png)

1.检查每个Broker文件系统容量大小；
![](/home/attachment/202304/27/1241241_168256187374153.png)
2.依次将3个Broker的NAS卷扩容20%；
修改kafka-cr的yaml文件中broker的PVC大小，并执行
![](/home/attachment/202304/27/1241241_168256189496670.png)
3.检查存储文件系统容量也变为120G大小，卷扩容成功。
![](/home/attachment/202304/27/1241241_168256192577339.png)

**4.1.3 使用NAS多租户对两个Kafka集群做数据隔离**
NAS多租户功能简介：存储支持按需创建多个租户，在租户内再创建文件系统；归属不同租户的NAS客户端通过指定的LIF只能访问该LIF归属租户下的文件系统、快照等资源；租户之间数据面、网络面和管理面均实现了隔离。
![](/home/attachment/202304/27/1241241_168256196263612.png)

在应用层，K8S可以通过Namespace完成各个应用的资源隔离；在存储层，使用华为OceanStor Dorado系列存储，利用NAS多租户功能，不同的应用对应的不同的租户，而不同的租户管理员只能看到自己的资源，真正做到数据逻辑隔离。

步骤 1       在存储上创建两个租户分别为tenant和tenant1；
![](/home/attachment/202304/27/1241241_168256200628152.png)

步骤 2       分别为两个租户tenant和tenant1创建两个租户管理员；
![](/home/attachment/202304/27/1241241_168256204299383.png)

步骤 3       在K8S创建两个namespace分别为kafka和kafka1
![](/home/attachment/202304/27/1241241_168256207044891.png)

步骤 4       分别创建对应租户tenant和tenant1的存储后端命名为backend_tenant和backend_tenant1；
![](/home/attachment/202304/27/1241241_168256209932252.png)

步骤 5      分别使用backend_tenant和backend_tenant1存储后端创建两个StorageClass命名为sc-tenant和sc-tenant1；
![](/home/attachment/202304/27/1241241_168256213091679.png)

步骤 6       使用sc-tenant和sc-tenant1在分别两个namespace kafka和kafka1中创建PVC拉起POD应用；
![](/home/attachment/202304/27/1241241_168256215825061.png)

步骤 7       通过租户管理IP地址，登录存储管理界面，只能查看当前租户下的文件系统资源。不同应用的资源和数据可以完全隔离，对外隐藏。
![](/home/attachment/202304/27/1241241_168256218323101.png)
![](/home/attachment/202304/27/1241241_168256220374766.png)

**4.1.14 配置存储租户配额简化容量管理**
租户配额简介：在存储上可以对每个租户设置容量配额，限制该租户可以使用的容量上限，防止租户过度使用存储空间。在使用多租户功能保障各个业务资源隔离的同时，使用租户配额功能，简化了存储管理员的容量管理工作，避免了业务之间互相影响。
![](/home/attachment/202304/27/1241241_168256223227761.png)

步骤 1    在存储侧创建租户tenant，并设置租户空间配额为3TB；
![](/home/attachment/202304/27/1241241_168256225321934.png)

步骤 2    在K8S上使用该存储租户创建一个Kafka集群，Broker数量为3，每个Broker所申请的空间为1TB；
![](/home/attachment/202304/27/1241241_168256228071298.png)

步骤 3    配额的3TB用完后，无法再继续创建新的PVC申请存储空间，因此创建PVC失败;
![](/home/attachment/202304/27/1241241_168256230731395.png)

步骤 4    在存储上调整租户空间配额为5TB；
![](/home/attachment/202304/27/1241241_168256233431562.png)

步骤 5    再次水平扩容Broker数量到5个，每个Broker所申请的空间为1TB，扩容成功。
![](/home/attachment/202304/27/1241241_168256235815437.png)

**4.1.5 验证Kafka节点故障后的POD飘移能力**
POD漂移能力是容器的关键功能之一，可以实现业务在集群内的灵活调度；同时在节点资源不足、节点故障等场景下可以通过漂移实现业务的快速恢复。如果使用本地盘存储，由于目的节点没有数据，POD无法跨节点漂移；使用共享存储则可以轻松实现漂移。本节将验证在节点故障时，如何利用POD漂移能力快速恢复Kafka实例的业务。
![](/home/attachment/202304/27/1241241_168256239055689.png)
1.       配置文件系统PV通过RWX的方式挂载给Kafka使用；
![](/home/attachment/202304/27/1241241_168256241179665.png)
2.       在Kafka集群创建1个Topic（6分区，3副本）；
```asp
[root@bastion kafka_ope]# oc -n kafka run kafka-topic -ti --image=registry.example.com:8443/kafka/kafka:0.31.1-kafka-3.2.3 --rm=true --restart=Never -- bin/kafka-topics.sh  --bootstrap-server bootstrap-server=my-cluster-kafka-bootstrap:9092 --create --topic mytopic --partitions 6 --replication-factor 3
```
![](/home/attachment/202304/27/1241241_168256245494422.png)
```asp
[root@bastion kafka_ope]# oc -n kafka run kafka-topic -ti --image=registry.example.com:8443/kafka/kafka:0.31.1-kafka-3.2.3 --rm=true --restart=Never -- bin/kafka-topics.sh  --bootstrap-server bootstrap-server=my-cluster-kafka-bootstrap:9092 --describe --topic mytopic
```
![](/home/attachment/202304/27/1241241_168256247872924.png)
3.       使用性能脚本往集群写入20GB数据；
```asp
[root@bastion kafka_ope]# oc -n kafka run kafka-topic -ti --image=registry.example.com:8443/kafka/kafka:0.31.1-kafka-3.2.3 --rm=true --restart=Never -- bin/kafka-producer-perf-test.sh --topic mytopic --num-records 1048576 --record-size 20480 --throughput -1 --producer-props bootstrap.servers=my-cluster-kafka-bootstrap:9092
```
![](/home/attachment/202304/27/1241241_168256260126974.png)
4.       下电一个broker1所在的节点worker0.okd4b.example.com；
![](/home/attachment/202304/27/1241241_168256262523746.png)
5.       查看该broker的Pod漂移到master0节点上；
![](/home/attachment/202304/27/1241241_168256264333621.png)
6.       记录漂移时间：漂移时间约为2分钟30秒

当Kafka通过容器化部署，且数据存储在外置存储上时，利用K8S中Pod故障漂移特性，可以快速在正常的node上拉起业务，使应用的可靠性不至于因为服务器架构的不可靠性而降低。

除此之外，测试中还对比了使用外置SAN存储的场景下，node故障触发Pod漂移时间约为9分钟左右。相较于容器场景下服务器本地盘Pod无法漂移和外置SAN存储漂移时间约为9分钟，外置NAS存储的漂移时间有较大优势。

## 4.2性能测试
本次性能测试主要参数配置说明：
![](/home/attachment/202304/27/1241241_168256273679135.png)

**性能结果如下：**
- 表4-1 使用本地盘与OceanStor Dorado的性能对比
![](/home/attachment/202304/27/1241241_168256277112830.png)
说明：使用本地盘时，每Broker配置一块1.92TB NVMe SSD盘。

测试结果：同样配置三副本的情况下，使用OceanStor Dorado吞吐量与本地盘相当，但时延略高。考虑到外置存储保证了数据的高可靠，我们测试了Kafka单副本配置下的性能，发现无论是吞吐量和时延都有3倍的提升，收益明显。

- 表4-2基于OceanStor Dorado的单Topic（三副本）性能测试数据
![](/home/attachment/202304/27/1241241_168256281082884.png)

- 表4-3 基于OceanStor Dorado的三Topic（三副本）性能测试数据
![](/home/attachment/202304/27/1241241_168256284126593.png)

- 表4-4 基于OceanStor Dorado的六Topic（三副本）性能测试数据
![](/home/attachment/202304/27/1241241_168256286795952.png)

**测试结果：**在保证端到端100ms时延左右时，OceanStor Dorado NAS可以提供高达数百MB/s的吞吐量，性能测试结果均达到预期，完全满足业务需求。

# 5 测试总结
本次测试模拟生产环境 Kafka 集群架构，实现了高可用Kafka集群实例在容器云平台节点上的独立部署，使用企业级NAS存储为 Kafka 集群提供了Topic分区数据、配置信息等关键信息的持久化。

•         本次测试中验证了卷在线扩容、节点在线扩容、节点再均衡等功能，均达到了预期的测试效果；

•         验证了节点故障的集群快速自愈能力，发现企业级NAS在该场景下可以实现高效的POD漂移；

•         考虑到企业级存储本身保证了数据的高可靠，可以减少Kafka消息副本数；测试了单副本下的性能，发现性能和时延均实现了三倍的提升，收益显著；

•         分别创建不同数量的Topic和Broker对集群进行了性能压测，发现OceanStor Dorado均可以达到接近本地盘的吞吐量和时延表现。

本测试方案验证了华为企业级NAS存储与红帽 OpenShift 容器云平台的组合，是性能卓越、稳定的运行平台，其中 Kafka 集群容器化部署表现出与非容器部署接近的性能，达到了对Kafka服务容器化改造的技术可行性验证的预期效果。

华为NAS kafka容器化容器云平台 kafka

著作权归作者所有

如果觉得我的文章对您有用，请点赞。您的支持将鼓励我继续创作！

添加新评论2 条评论

xiangxiang1999系统架构师北京某技术公司
2023-08-03 10:46

内容详尽，感谢分享！

menglunyang

系统工程师中国银行
2023-04-28 10:42

文章介绍了银行行业的Kafka容器化服务的测试目的、测试环境、测试内容、测试结果和测试总结，结构清晰，内容完整。可以在测试环境中介绍一下Kafka容器化的原理和实现方式，以及与传统Kafka部署方式的区别和优势，增加文章的深度和背景知识。感谢老师详细的文章，使我们对kafka容器化有了更深入的理解。

Ctrl+Enter 发表

匿名评论

银行行业Kafka容器化服务测试报告

添加新评论2 条评论

文章目录收起目录

本文隶属于专栏

作者其他文章

相关文章

相关问题

相关资料