基于Spark的地铁客流分析系统：架构、实践与避坑指南

简介面向计算机专业毕业设计的完整项目基于Spark的地铁大数据客流分析系统以城市地铁客流数据为分析对象覆盖数据采集、清洗、存储、分析、可视化与客流预测等环节适合大数据方向学生用于课程设计、毕设参考或技术实战练习。压缩包共194个文件包含Java/Scala源码、Spark作业脚本、SQL/HBase数据库脚本、XML/YAML配置、CSV测试数据以及项目PPT、PNG图像和日志文件等整体42.6MB按模块组织、目录清晰。已有259人学习/下载。通过系统可掌握Spark SQL、Streaming、MLlib等核心组件的应用理解客流统计、时间序列分析与结果展示思路文档部分涵盖需求分析、系统设计、数据库设计等软件工程材料日志配置、接口调用记录与Shell脚本则有助于环境搭建和排错方便完整复现毕业设计全过程并开展二次开发。1. 基于Spark的地铁客流分析毕设为什么值得拆开看地铁客流数据有三个和大数据场景天然贴合的特征全天候产生、时间连续、空间分布不均。高峰期一个站点的十分钟客流量能抵低谷期一天的量这种数据跑在单机MySQL上做一次小时级窗口聚合都捉襟见肘。基于Spark的地铁大数据客流分析系统就是用Spark分布式计算来解决这个瓶颈覆盖客流清洗、统计、预测的完整链路Logstash采集、HBase存储、Spark计算附带CSV数据集与HTTP接口文件。适合两类人一类是准备大数据方向毕设的学生缺一个能讲清架构、能真正跑通的参考项目另一类是工作里需要快速上手Spark的从业者想拿真实数据练手而不是反复跑WordCount。这份压缩包拆开是源码和配置合起来就是一条可复现的地铁客流分析Pipeline。2. 系统架构拆解HBase、Logstash与Spark是谁在解决什么问题2.1 为什么不是MySQLHBase加Spark的底层逻辑这个项目选型不是拍脑袋从数据规模与写入模式上都能站住脚。城市地铁的客流数据来源与业务人员手工录入完全不同它由闸机或AFC自动售检票系统持续产生事件流。假设一个城市有200个站每分钟产出上千条进出站事件一天的原始记录就接近百万级。这种高并发写入场景MySQL需要分库分表、读写分离才能撑住而这些方案都会显著增加开发量。HBase做列族存储按RowKey有序排列高吞吐写入是它的主场还天然支持按时间范围做高效Scan。第二数据落库后的分析需求通常是多维聚合按线路、站点、小时、上下行分组统计客流量或者对比工作日与节假日的曲线差异。HBase擅长点查和范围扫但复杂聚合不是它的强项。Spark SQL则专门做这件事——把HBase某个时间段的数据扫出来映射成DataFrame再用GroupBy、Join、窗口函数做分布式聚合计算压力被分散到集群多个executor上。一个扮演存储角色一个扮演计算角色分工明确。第三从毕设答辩视角看HBase加Spark能在架构图上画出完整闭环API采集、Logstash清洗、HBase落地、Spark分析、结果输出。如果选MySQL方案这个闭环会断在“大数据量”这个卖点上评审更容易追问“为什么不用Excel”。选择HBaseSpark至少可以在数据规模和计算模式上给出有说服力的答案。资源包里hbase.command脚本就是数据接入层的操作证明用HBase Shell命令建表、插数、Scan验证。我见过不少毕设项目把精力放在Spark代码上数据链路反而一塌糊涂最后Spark和HBase没连上。这个脚本的存在说明项目作者在数据接入上是有实际操作的。2.2 hbase.command与logstash-nginx.config数据接入层的真实形态hbase.command通常是一组HBase Shell命令用来初始化表结构和验证数据。常见内容与关键参数如下# 创建客流原始数据表列簇为info不保留历史版本 create metro_flow, {NAME info, VERSIONS 1} # 预分区按线路前缀拆4个region避免写入热点 create metro_flow, {NAME info}, {SPLITS [01, 02, 03, 04]} # 查看表是否存在 list metro_flow # 插入一条示例数据RowKey 线路站点编码 时间戳 put metro_flow, 01001_20240101120000, info:station, 站厅A put metro_flow, 01001_20240101120000, info:in_count, 120 # 按时间范围Scan验证数据落库 scan metro_flow, {STARTROW 01001_20240101, ENDROW 01001_20240102}这里三个参数是真正影响性能的地方。第一是RowKey设计。把线路站点编码放前、时间戳放后同一站点的数据在物理存储上相邻Scan就从随机读变成顺序读。如果反过来写成时间戳在前同一时间点所有站点的数据散落在不同区域查询时HBase要做大量随机IO。RowKey设计一旦定下来后期很难改相当于一把后悔药都没有。第二是SPLITS预分区。不预分区的表只有一个region所有写入都压在同一个RegionServer上集群再有5台机器也只有1台在干活。按线路前缀拆成多个region后写入可以分流。分区边界要根据实际RowKey前缀来定不要把四个分区边界设成无意义的数字。第三是VERSIONS参数。客流数据是典型的append-only流式数据旧版本没有保留价值设成1节省存储。代价是无法用旧版本做回滚分析但客流场景不需要。接着是logstash-nginx.config。在真实地铁系统里客流数据通常不是直接写HBase而是由Nginx代理层接收请求后输出access logLogstash再按固定间隔读取增量日志、解析成结构化字段、写出到HBase。配置分input、filter、output三段常见写法如下input { file { path /var/log/nginx/metro-access.log start_position beginning sincedb_path /dev/null codec json } } filter { grok { match { message %{TIMESTAMP_ISO8601:ts} %{DATA:station} %{WORD:direction} %{INT:flow} } } date { match [ts, ISO8601] target timestamp } } output { hbase { table metro_flow rowkey %{station}_%{timestamp} columns [ info:direction, %{direction}, info:flow, %{flow} ] } }这里有一个容易被忽略却影响巨大的配置sincedb_path。它记录Logstash读取文件的offset位置。设置成/dev/null时每次重启Logstash都会从文件开头重新读一遍开发调试期方便但生产环境一旦重启全部历史日志会重复灌入HBase数据直接翻倍。初跑通后务必改成持久化路径比如sincedb_path /var/lib/logstash/sincedb/metro.db。grok插件是Logstash解析文本日志的常用方式用正则按命名捕获字段。如果Nginx日志格式调整过但grok没同步更新字段会解析失败output段引用变量变成空。排查方法是先跑logstash -f 配置文件 --config.test_and_exit验证语法再用stdout输出插件代替hbase输出先确认解析结构无误再接HBase。2.3 szmc.net-metro.csv字段结构与分析切入点szmc.net-metro.csvszmc大概率是深圳地铁的拼音缩写是整个项目的核心数据集。毕设里的CSV规模一般在几十MB到几百MB字段设计参考了真实AFC系统的数据结构。常见字段大致如下字段名类型含义station_codestring站点编码station_namestring站点名称line_noint线路编号directionstring上行/下行ticket_typestring单程票/卡/扫码in_timetimestamp进站时间out_timetimestamp出站时间flow_countint客流量拿到CSV后第一件事不是直接写Spark代码而是先确认文件本身的形态。用命令行看一眼# 看文件编码避免UTF-8 BOM导致表头污染 file szmc.net-metro.csv # 看每行长度排除跨行字段 awk {print NR, length($0)} szmc.net-metro.csv | head -5 # 看表头和前两行内容 head -3 szmc.net-metro.csv如果发现表头第一列有看不见的BOM字符用sed去BOM再继续。字段时间格式不统一也是常见问题有的行是2024-01-01 12:00:00有的行是2024/01/01 12:00:00这种情况Spark的TimestampType解析会部分失败。可以先用sed或Excel批量统一格式比在Spark里写复杂解析逻辑省事得多。分析切入点的选择直接影响毕设功能列表是否饱满我建议覆盖三个方向。一是分时段客流统计找出早高峰和晚高峰的时间窗口用Spark SQL的hour函数加groupBy就能实现。二是站点热度排行算每个站的日进出站总量输出Top N。三是线路断面客流识别最拥挤的区间这个方向需要把站点线间距数据join进来是展示多表join能力的好素材。三个方向代码量都不大但组合起来能覆盖客流分析系统的常见功能答辩时也容易讲清楚设计思路。3. Spark集群搭建与代码落地从local模式到spark-submit的完整链路3.1 Spark的安装与使用集群搭建的两种模式怎么选先明确一个原则这份毕设数据量不大如果本机内存只有8G不要一上来搭三节点集群。先以local模式跑通代码全部功能正常后再考虑Standalone分布式集群既能省去调试集群环境的时间也能在答辩时证明你有能力部署真正的Spark集群。Spark安装的第一步是确认版本匹配关系。毕设代码如果是用Spark 2.x的API写的直接上手3.x可能遇到部分API行为变化。解压后先做三项环境配置export JAVA_HOME/usr/lib/jvm/java-8-openjdk-amd64 export SPARK_HOME/opt/spark-3.2.4-bin-hadoop3.2 export PATH$SPARK_HOME/bin:$PATH然后启动spark-shell做一个最简单的并行计算验证spark-shell --master local[2] sc.parallelize(1 to 100, 4).map(_ * 2).sum()local[2]表示使用本地2个CPU核心。8G内存机器建议local[2]或local[4]不要贪多核心数给多了反而会因为内存不足OOM。如果要展示Spark集群Standalone模式是最轻量的选择不需要HDFS也能跑。主节点和工作节点的启动命令如下# 主节点-h指定Master地址-p指定通信端口 $SPARK_HOME/sbin/start-master.sh -h 192.168.1.10 -p 7077 # 工作节点-c指定CPU核数-m指定可用内存 $SPARK_HOME/sbin/start-worker.sh spark://192.168.1.10:7077 -c 2 -m 2g7077是Master与worker通信的端口8080是Web UI端口前者用于提交作业后者用于浏览器查看集群状态。worker的-m 2g表示最大可分配内存不要超过机器物理内存减去系统预留内存的值否则系统OOM后整个节点都会掉线。提交作业时用spark-submit指定Master地址和主类spark-submit \ --master spark://192.168.1.10:7077 \ --executor-memory 1g \ --total-executor-cores 4 \ --class com.metro.analysis.FlowStat \ metro-analysis.jar \ file:///data/szmc.net-metro.csvexecutor-memory 1g是合理区间。集群经验里常有一个误区内存越大越好。实际上executor堆内存太大JVM GC停顿时间会显著拉长计算反而变慢。毕设数据量下1g到2g的executor内存足够如果发现数据量超过2g就多开executor而不是单开大内存。大数据集群部署策略里还涉及一个细节spark-submit的输入路径。数据放本地就用file://前缀放HDFS则用hdfs://前缀。如果两种路径混用最典型的问题是读取一个不存在的hdfs目录作业启动后卡在等待状态或直接抛FileNotFoundException。用hdfs dfs -ls先确认路径存在是基本操作。3.2 Spark中读取JSON嵌套结构与explode展开“spark中读取json”这个搜索点很多人是在把API返回的JSON变成DataFrame时被卡住。JSON与CSV表格结构不同可能出现嵌套数组。Spark需要用explode把嵌套展开后才能做分析。假设HTTP接口返回的数据格式是{station:科学馆,records:[{hour:8,flow:320},{hour:9,flow:450}]}读取代码import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions._ val spark SparkSession.builder() .appName(MetroJsonReader) .master(local[2]) .getOrCreate() val rawDf spark.read .option(multiLine, true) .json(file:///data/metro_records.json) val flowDf rawDf .withColumn(record, explode(col(records))) .select( col(station), col(record.hour).as(hour), col(record.flow).as(flow) ) flowDf.show(10)explode在这里把一行中的records数组展开为多行每个数组元素生成一行。col(record.hour)用于访问STRUCT子字段。multiLine选项处理JSON跨行的情况一条JSON对象占多行时必须开启否则Spark把每个物理行都当成独立JSON对象解析结果就是大量解析失败。如果文件是每行一个JSON对象而非一个大数组就不要用multiLine直接option(multiLine, false)。两种方式的取舍是文件整体是一个大JSON用true文件每行一个JSON对象用false。3.3 Spark读取CSV显式Schema与类型推断的取舍CSV读取是毕设中最常见的入口代码很简单但坑都在参数上。先看显式schema的写法import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types._ val spark SparkSession.builder() .appName(MetroFlowStat) .master(local[4]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate() val schema StructType(Array( StructField(station_code, StringType, true), StructField(station_name, StringType, true), StructField(line_no, IntegerType, true), StructField(direction, StringType, true), StructField(ticket_type, StringType, true), StructField(in_time, TimestampType, true), StructField(out_time, TimestampType, true), StructField(flow_count, IntegerType, true) )) val df spark.read .option(header, true) .option(dateFormat, yyyy-MM-dd HH:mm:ss) .schema(schema) .csv(file:///data/szmc.net-metro.csv)StructField的第三个参数nullabletrue表示该字段允许null。in_time用TimestampType如果CSV里的时间格式与dateFormat不匹配Spark解析为null而不是报错这是最容易“看起来成功实则失败”的地方。排查方式是用.filter(col(in_time).isNull).count()统计null数量如果数字很大就要检查数据格式。与显式schema相对的是inferSchematrue自动推断类型。它省事但Spark需要额外扫描一遍数据来推断类型数据量大时浪费时间。毕设代码里更推荐显式定义schema一来速度快二来类型错误在读取阶段就暴露出来不用等到聚合时才发现某列是字符串排查成本更低。3.4 客流高峰统计实战从DataFrame到聚合结果现在拿真实CSV做一个完整的客流高峰统计计算每个站点在早高峰(7-9点)和晚高峰(17-19点)的总客流量输出Top20。import org.apache.spark.sql.functions._ val peakStats df .withColumn(hour, hour(col(in_time))) .withColumn( period, when(col(hour).between(7, 9), morning_peak) .when(col(hour).between(17, 19), evening_peak) .otherwise(off_peak) ) .filter(col(period) ! off_peak) .groupBy(station_name, period) .agg( sum(flow_count).as(total_flow), avg(flow_count).as(avg_flow) ) .orderBy(col(total_flow).desc) peakStats.show(20, false)计算逻辑说明如下。hour函数从Timestamp列提取小时整数between包含边界值也就是说7到9包含7点和9点。when分支相当于CASE WHEN用otherwise兜底非高峰时段。groupBy按站点和时段两个维度分组sum和avg分别算出总量与均值orderBy按总量降序排列取Top20。spark.sql.shuffle.partitions这个参数在刚才的session配置里被设置成8用于控制shuffle时数据分成多少个分区。系统默认200是给大数据准备的这份数据量只有几十万行200个task大部分是空的白白占用调度开销。本地或者小数据场景把它降到8到16能明显提速。但如果之后要切换到完整大数据场景记得把这个值改回100以上否则并行度可能成为新的瓶颈。执行后如果结果为空有一个常见排查顺序。第一步df.count()确认CSV读取了多少行。第二步df.select(in_time).filter(col(in_time).isNotNull).limit(5).show()确认时间字段解析成功。第三步把period直接取出打印确认when分支的过滤条件写对没有。这三步配合基本可以定位绝大多数空结果问题。3.5 MLlib客流预测构建训练集的最小实现预测是毕设里的一个闪光点用MLlib做线性回归即可不要再引入深度学习框架。预测思路是利用历史时间序列用前四个小时客流量预测第五个小时客流量构造训练数据时使用lag窗口函数import org.apache.spark.sql.expressions.Window import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.regression.LinearRegression val tsDf df .groupBy(station_code, hour) .agg(sum(flow_count).as(flow)) .orderBy(station_code, hour) val featureDf tsDf .withColumn(lag1, lag(flow, 1).over(Window.partitionBy(station_code).orderBy(hour))) .withColumn(lag2, lag(flow, 2).over(Window.partitionBy(station_code).orderBy(hour))) .withColumn(lag3, lag(flow, 3).over(Window.partitionBy(station_code).orderBy(hour))) .withColumn(lag4, lag(flow, 4).over(Window.partitionBy(station_code).orderBy(hour))) .filter(col(lag4).isNotNull) val assembler new VectorAssembler() .setInputCols(Array(lag1, lag2, lag3, lag4)) .setOutputCol(features) val finalDf assembler.transform(featureDf) val Array(train, test) finalDf.randomSplit(Array(0.8, 0.2), seed 42) val lr new LinearRegression() .setFeaturesCol(features) .setLabelCol(flow) .setMaxIter(50) .setRegParam(0.01) val model lr.fit(train) model.transform(test).select(flow, prediction).show(10)这段代码有几个关键点值得说明。lag函数必须配合Window使用。partitionBy(station_code)意味着每个站点单独计算滞后特征orderBy(hour)确保按时间顺序取前面的值。如果漏掉partitionBy所有站点的数据会被打乱来计算lag预测完全失真。滞后特征生成后前4行必然是null因为历史数据不够。filter(lag4.isNotNull)把不完整记录剔除否则训练时模型遇到null特征会直接报错或丢弃数据。randomSplit切分训练测试集时seed42是故意的为了让每次运行的划分结果一致。如果seed不固定每次运行评估指标不同答辩时无法复现报告中的数字。LinearRegression的RegParam是L2正则化系数。0.01是一个保守值能在客流这种有明显周期性的数据上起到防止过拟合的作用。如果预测结果波动很大可以把RegParam提升到0.1或0.5或者改用GBTRegressor梯度提升树。GBT不用做特征标准化对非线性客流曲线通常表现更好。4. 避坑指南五个常见翻车现场与排查路径4.1 现象Spark作业日志没报错但结果全null现象spark-submit之后日志没有error但输出结果里客流相关字段全是null。原因这个地方容易误导人日志是绿色的看着像成功。时间字段解析失败是首要嫌疑。dateFormat与CSV实际格式不一致时TimestampType解析会返回null而不是抛异常。另外CSV文件如果带UTF-8 BOM表头第一列字段名会带不可见字符导致该列匹配失败所有行这一字段都查不到。排查路径要按顺序走。先df.printSchema()看时间列类型再df.select(in_time).filter(col(in_time).isNotNull).count()看有效值有多少如果为0就是解析问题最后用head -3 szmc.net-metro.csv看原始时间长什么样。BOM处理可以用sed -i s/^\xEF\xBB\xBF//去掉BOM。这个小步骤命令行几秒完成但每年有大量人卡在它上面。还有一个容易被忽略的坑是CSV里同一个字段的时间格式混用比如一半是2024-01-01 12:00:00一半是2024/01/01 12:00:00。Spark内部用SimpleDateFormat解析遇到第二种格式就无法识别。处理方式是用sed做一次文本替换统一格式再重新读取。如果混用属于业务数据特性就要在代码里用when加regexp_replace做额外处理但我建议能在数据清洗时就解决不要拖到Spark代码里增加复杂度。4.2 现象HBase写入慢到无法接受每秒只有几百条现象用代码批量put一百万条记录速度只有每秒几百条作业跑数小时。原因第一个是建表没有预分区。HBase默认单region所有写入热点在同一RegionServer即使集群有5个节点也只有一台在干活。第二个是使用逐条put提交每条put都发起一次网络RPCRTT开销完全掩盖了写入吞吐量。解决分两步。第一步使用SPLITS预分区重建表disable metro_flow drop metro_flow create metro_flow, {NAME info}, {SPLITS [01, 02, 03, 04, 05]}分区数不必贪多按RowKey前缀的实际分布来定。分区边界设在不常出现的值上反而会打散写入热点导致各region数据量不均衡。第二步用BufferedMutator批量写把缓冲区攒满再统一提交BufferedMutatorParams params new BufferedMutatorParams(TableName.valueOf(metro_flow)); params.writeBufferSize(1024 * 1024 * 4); BufferedMutator mutator conn.getBufferedMutator(params); Put put new Put(Bytes.toBytes(rowkey)); put.addColumn(Bytes.toBytes(info), Bytes.toBytes(flow), Bytes.toBytes(120)); mutator.mutate(put); mutator.flush();writeBufferSize设为4MB表示缓冲区攒够4MB批量发送一次相比逐条put性能提升一个数量级。flush时机在循环结束后再做不要每条都flush否则又退化成逐条提交。4.3 现象HBase shell能查到数据Spark读出来却是空表现象HBase命令行执行scan有记录但Spark作业读取同一个表返回的DataFrame行数为0。原因连接器版本与服务端不兼容是首要怀疑对象。HBase 2.x装旧版1.x的连接器Spark执行时出现NoSuchMethodError或者加载不到TableInputFormat。另一个可能性是列族名称不匹配HBase表定义用的是info代码Spark读的时候用的是cf没有实际指定正确列。解决时先做版本核对。打开项目的依赖配置看hbase-client、hbase-server、spark-hbase的版本坐标与目标集群是否匹配。然后写一个最小读取代码不进行任何过滤直接读10行打印结果确认能正常拉数据后再扩大查询范围。还有一个小坑是连接HBase需要的zookeeper.quorum配置代码里没有指定ZooKeeper地址时连接会一直超时表现为“Spark读不到”实际上连HBase都没连上。4.4 现象窗口函数本地能跑上集群却OOM现象本地小数据量测试时lag窗口正常到集群处理完整数据后作业运行时间暴涨甚至OOM。原因窗口函数的shuffle阶段会把partitionBy键相同的所有数据拉到一个executor上计算。如果某个站的客流历史数据特别多比如中心枢纽站数据倾斜就出现了。Web UI上看executor的Shuffle Read Bytes某个任务的数据量远大于其他任务就是倾斜的直接证据。解决方法是让partitionBy的键更具体。原先是partitionBy(station_code)改成partitionBy(station_code, line_no)使每份分区数据量更均衡。再配合spark.sql.shuffle.partitions参数调整shuffle分区数。如果倾斜仍然严重可以对数据做repartition按站点hash重排但这是最后手段会增加额外一轮shuffle。毕设场景下数据量不大通常不需要走到这步但答辩中展示出你知道倾斜怎么排查是一个明显的加分项。4.5 现象spark-submit找不到主类或加载不到依赖现象项目在IDEA里运行顺利打包成jar后提交Spark集群报ClassNotFoundException或“Failed to find main class”。原因IDEA运行时自动打包了依赖classpath但spark-submit不会。要么是主类本身没有被打进jar要么是第三方库没有打包成fat jar。还有一种情况是assembly打包时多个jar的META-INF/services冲突导致运行时服务加载器读不到实现类。解决使用maven-shade-plugin或sbt-assembly插件构建fat jar确保jar内包含全部第三方依赖。主类在插件配置中要写完全限定名。然后先本地验证spark-submit --master local[2] --class com.metro.analysis.FlowStat metro-analysis.jar file:///data/szmc.net-metro.csv本地通过后再换master地址能快速验证打包正确性。如果本地都起不来就没必要上集群排查了。需要特别注意排除Spark自身的jar因为Spark运行环境已经提供它们重复打入会造成class冲突。5. 数据对拍与答辩验证从能跑到能讲的三件事5.1 用awk对拍验证聚合结果Spark跑完后需要对结果做一次独立验证。我用得最多的方法是对拍把同一份CSV用awk做一个轻量汇总与Spark输出对比。由于CSV字段顺序固定awk脚本很短awk -F, NR1 {sum[$2]$8} END {for (s in sum) print s, sum[s]} szmc.net-metro.csv | sort -k2 -rn | head -20这个命令按第二个字段站点名分组把第八个字段客流量累加排序后取前20。如果这20个站点的排名与Spark结果一致主流程基本可靠。有差异时优先排查时间过滤和字段错位问题。Spark的DataFrame计算和awk这类独立工具的结果互相印证是最直接的可信度证明。5.2 HTTP接口与调试文件的使用szt-api.http文件是HTTP接口调试文件。JetBrains系IDE可以直接点击运行VS Code需要安装REST Client插件。演示时建议把接口请求、Spark结果、可视化三块连成一个流程先通过HTTP接口确定数据范围再调用Spark作业产出指标最后把指标输出到图表。这样能同时展示项目的数据采集能力和计算能力。接口验证有一个细节HTTP接口返回JSON时先用格式化工具确认字段名与Spark读取的列名一致。字段名大小写不一致是导致接口与数据链路脱节的常见原因。5.3 答辩演示节奏控制与数据预跑演示前强制做三件事数据对拍通过、计划演示的Spark作业提前跑一遍并保存输出、HTTP接口请求确认可用。答辩时可以快速展示核心代码然后现场提交一次真实运行十几秒内出结果。如果运行超过一分钟不要干等立刻切到架构图或者已有结果继续讲解。从那以后我每次做Spark相关演示都会强制走一遍“对拍、接口验证、提前跑通”这三个动作现场翻车的概率能降到最低。希望帮到你。本文还有配套的精品资源点击获取