Hadoop分布式计算下的朴素贝叶斯文本分类：从MapReduce原理到实现与调优

简介：这是基于Hadoop的朴素贝叶斯文本分类项目，利用MapReduce实现训练与评估，适用于Hadoop课程设计、毕业设计或文本分类入门实践。项目包含词频统计、文档计数、模型构建与测试集分类等核心步骤，并输出精确率、召回率和F1评估结果；实验数据选用NBCorpus中的CHINA与CANA两类，按70%与30%划分训练集和测试集，可复现完整的分类实验流程。压缩包内共552个文件，以txt语料为主，辅以java源码、png流程示意图、md说明、docx报告及xml/properties配置文件，整体大小3.75MB，目录结构清晰。已有230人学习，适合计算机相关专业学生对照源码理解分布式朴素贝叶斯实现，也可作为课程作业或毕业设计的高质量参考。

1. 从课程设计到生产脚本：Hadoop 朴素贝叶斯文本分类器到底值不值得做

拿到“基于Hadoop开发实现的朴素贝叶斯文本分类器+源代码+文档说明”这个选题，多半是课程设计或毕业设计的场景：MapReduce 入门实验已经把 WordCount 跑吐了，想找一个看起来有用、工作量又可控的方向。朴素贝叶斯文本分类正好卡在中间——算法简单，几页纸能讲清楚；可一旦往 Hadoop 上搬，就要处理序列化、Shuffle、模型下发这些真实问题。所以这个方向表面上是作业，实际上是理解 Hadoop 计算模型的好样本。

这套代码的核心价值不在“准确率能到多少”，而在帮你建立训练和预测两套 MapReduce 数据流。真正把环境装好、把数据喂进去、把模型文件看懂的人，不会觉得这份作业白做。下文按模型推导、训练代码、预测代码、运行避坑、调优技巧五个环节展开，每一步都能直接抄，但更重要的是知道为什么这么写。

2. 朴素贝叶斯的 MapReduce 建模：先验、条件概率与 log 加法为什么缺一不可

2.1 多项式模型下的文本表示：词袋与条件独立

朴素贝叶斯处理文本时，默认用词袋模型。文档被切成本文分词后的词序列，不考虑词序和语法，只统计每个词在文档里出现了几次。这个“丢失顺序”的假设在 NLP 里看起来很粗糙，但它在 MapReduce 下的语义非常干净：一个词出现的次数就是一次累加，两个词之间没有依赖关系，天然适合 Mapper 独立计数、Reducer 汇总的结构。

这里要分清两种朴素贝叶斯变体。伯努利模型只看词是否出现，值为 0/1；多项式模型统计词频，词出现 3 次和出现 1 次对后验概率的贡献不同。文本分类场景里多项式模型几乎总是更稳，因为它保留了“财经类文章会反复出现‘股价’，体育类文章会反复出现‘比分’”这种信息。本项目的实现也是多项式模型，训练阶段计数的单位是“词次”而不是“文档中有没有这个词”。

条件独立假设是整套算法的地基：给定文档类别后，任意两个词的出现相互独立。这个假设在真实文本里肯定不成立——“利率”和“银行”明显相关——但它让联合概率拆成了单个词概率的连乘，MapReduce 才能把统计任务打散到每个词上。实际工程里，这个不成立的假设带来的误差远小于训练数据不干净带来的误差，所以不用太纠结。

2.2 训练阶段需要统计的三组数：先验、条件概率与词汇表

训练一个多项式朴素贝叶斯分类器，最终只需要一组数字：每个类别 c 的先验概率 P(c)，以及每个词 w 在类别 c 下的条件概率 P(w|c)。先验用文档数算：P(c) = c 类文档数 / 总文档数。条件概率用词频算：

P(w|c) = (词 w 在 c 类文档中出现的总次数 + α) / (c 类所有词出现总次数 + α × V)

其中 V 是训练语料去重后的词汇表大小，α 是拉普拉斯平滑系数。α 的经典取值是 1，含义是“即使某个词在某个类别里一次都没出现过，也给它一个微小的概率，避免预测时直接撞上 0”。V 取全局词汇表，而不是每个类各自的词汇表，这样所有类的条件概率分母尺寸一致，概率之间才可比较。

把这三组数和 MapReduce 对应起来就清楚了：

这三组数不需要分开跑三个 Job，一个 Mapper 里同时输出两类 key，一个 Reducer 里根据 key 前缀分别处理就行。但注意：如果想让 Reducer 在结束时统一计算概率，必须保证所有统计量进入同一个 Reducer。这直接决定了后面你运行时要给 Job 设置mapreduce.job.reduces=1。

2.3 预测公式与 log 转换：为什么新手实现总是一篇文档分到全是 0

训练完成后，预测文档 d 的类别就是找后验概率最大的 c：

score(c) = log P(c) + Σ log P(w|c)

注意是 log 概率加法，不是概率连乘。原因很直接：文本分类的词汇表动辄几万、几十万项，一篇文档里出现的词也有几十上百个，每个条件概率通常都远小于 1，连乘起来立刻下溢成 0.0。你会在预测结果里看到所有文档的后验概率全是 0，最后取最大值时取到的往往是先验最大的那个类——整个分类器等于退化成“猜频率最高的类别”。

log 变换把连乘变成加法，既避免了浮点下溢，也把概率值拉到了可比较的尺度。模型文件里直接存 log P(c) 和 log P(w|c)，预测时做加法就好，不用每次读模型都现算一遍对数。

还有一个边界问题：测试文档里会出现训练语料里没见过的词，这些词没有对应的条件概率。常见做法是直接跳过，不让未登录词影响得分。如果你强行让每个未登录词都参与计算，就得给它一个概率值，给大了会稀释真实特征，给小了会变成负无穷。我的建议是跳过，第 4 章的代码也是这么写的。

3. 训练阶段代码实现：从 TextInputFormat 到一份可追加的模型文件

3.1 先约定三件事：输入格式、输出格式和项目结构

训练代码动手前，先把数据格式钉死。我用的训练集格式是每行一条样本，类别标签和正文用 TAB 分隔。TextInputFormat 天然按行切分，TAB 在中文文本里几乎不会出现，做 split 时比空格和逗号都安全。格式示意：

财经 央行今日开展5000亿元逆回购操作 体育 湖人客场逆转掘金取得系列赛领先

注意不要在正文里再混入 TAB，预处理阶段要做一次清洗。文档 ID 在训练阶段不需要，预测阶段才需要，所以训练集不用带 ID。这里推荐一份最基本的项目结构：

README 这个东西很多人拖到最后才补，但我建议一开始就写。里面至少要包含 Hadoop 版本、JDK 版本、打包命令、训练与预测的完整命令、输入数据格式说明。这份文档说明才是别人愿意复现你项目的底气，比把代码写漂亮重要得多。

3.2 TrainMapper：一份输入数据同时产出两类计数

Mapper 的核心逻辑是把“类别出现一次”和“词在类别下出现一次”都变成<Text, IntWritable>输出。下面这段代码可以直接用：

public class TrainMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private static final Pattern WORD_PATTERN = Pattern.compile("[\\u4e00-\\u9fa5]{2,}|[a-zA-Z0-9]+"); private static final IntWritable ONE = new IntWritable(1); private Text outKey = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); int sep = line.indexOf('\t'); if (sep < 0) { return; } String label = line.substring(0, sep).trim(); String body = line.substring(sep + 1); if (label.isEmpty() || body.isEmpty()) { return; } // 输出先验计数：__PRIOR__ + 类别 outKey.set("__PRIOR__" + label); context.write(outKey, ONE); // 输出词频计数：词__类别 for (String word : tokenize(body)) { outKey.set(word + "__" + label); context.write(outKey, ONE); } } private List<String> tokenize(String text) { List<String> words = new ArrayList<>(); Matcher matcher = WORD_PATTERN.matcher(text); while (matcher.find()) { words.add(matcher.group().toLowerCase()); } return words; } }

这段代码有两点要说清楚。第一，__PRIOR__前缀是用来区分先验计数和条件概率计数的，Reducer 看到这个前缀就知道这条记录属于文档级统计，而不是词级统计。第二，tokenize()用正则连切中文和英文，两个连续汉字以上切成一个词，英文单词和数字保留，所有词转小写。这是课程设计里能最快跑通又不引入第三方依赖的分词方案；实际产品里中文分词会用 IK 或 jieba，后面避坑章节会讲替换分词器时踩过的坑。

Mapper 的输出 key 设计成“词__类别”而不是“类别__词”，是为了让同一类别的词在 Shuffle 后尽量聚集在一起。不过由于 Reducer 设置成 1，这个 key 顺序在这里影响不大，真正影响大的是 Combiner 能不能正确合并。

3.3 Combiner 与 Reducer：为什么需要-D mapreduce.job.reduces=1

Combiner 的作用是在 Mapper 端先做一次局部合并，减少 Shuffle 的数据量。对这份代码来说，Combiner 的 reduce 逻辑和 Reducer 完全一样，都是对同一个 key 的 value 求和，唯一区别是 Reducer 还需要在 cleanup 阶段算概率。所以 Combiner 可以单独写一个类继承 Reducer 只做求和，也可以直接在 Driver 里把 Reducer 类同时设为 Combiner。我通常单独写一个TrainCombiner，因为后续如果要加停用词过滤或特征剪枝，Combiner 和 Reducer 的职责会快速分离。

Reducer 的写法放在训练阶段最核心的位置：

public class TrainReducer extends Reducer<Text, IntWritable, Text, Text> { private Map<String, Long> classDocCount = new HashMap<>(); private Map<String, Long> wordCount = new HashMap<>(); private Map<String, Long> classWordTotal = new HashMap<>(); private Set<String> vocabulary = new HashSet<>(); private static final double ALPHA = 1.0; @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) { long sum = 0; for (IntWritable val : values) { sum += val.get(); } String k = key.toString(); if (k.startsWith("__PRIOR__")) { classDocCount.put(k.substring("__PRIOR__".length()), sum); } else { int sep = k.lastIndexOf("__"); if (sep < 0) { return; } String word = k.substring(0, sep); String label = k.substring(sep + 2); wordCount.put(k, sum); classWordTotal.merge(label, sum, Long::sum); vocabulary.add(word); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { long totalDocs = classDocCount.values().stream().mapToLong(Long::longValue).sum(); int vocabSize = vocabulary.size(); for (Map.Entry<String, Long> entry : classDocCount.entrySet()) { String label = entry.getKey(); double prior = Math.log((double) entry.getValue() / totalDocs); context.write(new Text("__PRIOR__" + label), new Text(String.valueOf(prior))); } for (Map.Entry<String, Long> entry : wordCount.entrySet()) { String fullKey = entry.getKey(); int sep = fullKey.lastIndexOf("__"); String word = fullKey.substring(0, sep); String label = fullKey.substring(sep + 2); long count = entry.getValue(); double prob = (count + ALPHA) / (classWordTotal.get(label) + ALPHA * vocabSize); context.write(new Text(fullKey), new Text(String.valueOf(Math.log(prob)))); } } }

注意最后两段循环都写在 cleanup 方法里。cleanup 会在所有 reduce 调用结束后执行，此时classDocCount、wordCount、vocabulary三个集合已经装满了全量统计，才能算出先验、词汇表大小和条件概率。这要求整个 Job 的所有统计结果落在同一个 Reducer 里，所以提交命令必须带-D mapreduce.job.reduces=1。如果设置成多个 Reducer，每份统计数据被 hash 到不同的节点，cleanup 里算出来的模型就是残缺的，准确率会莫名其妙地崩掉。

wordCount这个 Map 在数据量大时会占不少内存。一个典型课程设计语料几十万词条，几百 MB 堆内存够用；如果语料上千万词条，就要考虑不缓存全量数据，改成两个 Job 分阶段计算，或者用 HBase 这类外部存储在 cleanup 里查数。伪分布式环境下遇到内存不足，先看第 5 章的内存参数调整，不要上来就加机器。

训练 Job 跑完后，HDFS 的模型目录下会生成part-r-00000，每一行是key\tlog概率的格式。这个文件就是预测阶段的全部依赖，后续做特征剪枝时压缩的也是这个文件。

4. 预测阶段代码实现：一个 Mapper 完成全部分类的设计与模型分发

4.1 模型文件如何进入每个 Mapper：-files 参数比 DistributedCache 更省心

预测阶段每个 Mapper 都需要加载完整的模型文件，也就是第 3 章那份part-r-00000。最直接的方式是在 setup 里用 FileSystem 打开 HDFS 上的模型路径去读，但这样每个容器都要远程读一次 HDFS 文件，模型稍大时 NameNode 和 DataNode 的压力明显上升。

常见的做法是用-files参数把模型文件随任务分发到每个容器的本地工作目录。提交命令长这样：

hadoop jar nb.jar nb.PredictDriver \ -files hdfs:///user/hadoop/nb/model/part-r-00000#model.txt \ /user/hadoop/nb/test.txt /user/hadoop/nb/result

#后面的model.txt是容器内可见的文件名。分发成功后，PredictMapper 的 setup 里直接以本地文件方式读取model.txt即可，不需要再操作 HDFS 客户端。这个写法的好处是对 Hadoop 2.x 和 3.x 都有效，而且绕开了 DistributedCache 在版本升级后 API 变化的问题。

提示：-files后面指定的是单个文件。如果训练时 reducers 数不为 1，模型会拆成多个 part 文件，这时先用hadoop fs -getmerge把模型合并成一个文件再分发，比在 setup 里循环读 HDFS 目录省事得多。

4.2 PredictMapper 的 setup 与 map：读模型、算得分、取最大

测试集的格式和训练集不同，每行是三列：文档ID\t真实类别\t正文。真实类别这列只用于事后验证，分类时完全不读它。如果做纯预测，可以省略第二列，但带上它对后面算混淆矩阵非常方便。

public class PredictMapper extends Mapper<LongWritable, Text, Text, Text> { private Map<String, Map<String, Double>> condProbs = new HashMap<>(); private Map<String, Double> priorProbs = new HashMap<>(); @Override protected void setup(Context context) throws IOException { try (BufferedReader br = new BufferedReader(new FileReader("model.txt"))) { String line; while ((line = br.readLine()) != null) { int idx = line.indexOf('\t'); if (idx < 0) { continue; } String key = line.substring(0, idx); double value = Double.parseDouble(line.substring(idx + 1)); if (key.startsWith("__PRIOR__")) { priorProbs.put(key.substring("__PRIOR__".length()), value); } else { int sep = key.lastIndexOf("__"); if (sep < 0) { continue; } String word = key.substring(0, sep); String label = key.substring(sep + 2); condProbs.computeIfAbsent(word, k -> new HashMap<>()) .put(label, value); } } } } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] parts = value.toString().split("\t", 3); if (parts.length < 3) { return; } String docId = parts[0]; String realLabel = parts[1]; String body = parts[2]; List<String> words = tokenize(body); String bestLabel = null; double bestScore = Double.NEGATIVE_INFINITY; for (String label : priorProbs.keySet()) { double score = priorProbs.get(label); for (String word : words) { Map<String, Double> wordProbs = condProbs.get(word); if (wordProbs != null) { Double logProb = wordProbs.get(label); if (logProb != null) { score += logProb; } } } if (score > bestScore) { bestScore = score; bestLabel = label; } } context.write(new Text(docId), new Text(realLabel + "\t" + bestLabel + "\t" + bestScore)); } }

setup 里把模型文件逐行解析成两个 Map。condProbs的 key 是词，value 是“类别到 log 概率”的映射；priorProbs是“类别到 log 先验”。map 里对每个类别都累加一次得分，未登录词直接跳过，最终取最高分的类别输出。这里的tokenize()必须和训练阶段用完全一样的逻辑，否则模型里查不到训练时的词，特征全部失效，分类结果会退化到只比先验。这也是整个项目里最容易出的隐形错误之一。

因为预测只需要计算和比较，不需要跨文档聚合，所以 Driver 里设置job.setNumReduceTasks(0)。最终结果直接写到part-m-00000文件里。新手经常在这里翻车：习惯性去找part-r-00000，发现目录是空的，其实结果就在part-m-00000中。

4.3 预测输出的稳定格式：让你的结果可验证、可对比、可排错

输出格式定为文档ID\t真实类别\t预测类别\t得分四列，其中得分是 log 后验概率的绝对值，只用于排序对比，没有业务含义。真实类别保留在结果里，是为了后面做混淆矩阵时不用再拿文档 ID 去关联原测试集。这个四列格式从第一次运行就固定下来，后续所有调优都以它为依据，不要跑着跑着改列顺序。

复杂度上，每个 Mapper 对每篇文档都要遍历所有类别，整体是 O(文档词数 × 类别数)。类别在十几个以内时完全没有性能压力；如果类别上百，建议先给每个类做一次特征词剪枝，减小模型的遍历规模，这正好接到第 6 章的内容上。

测试集如果很大，一个 Mapper 处理一个分片，模型文件会被复制到每个节点上，再加上每篇文档都在循环里查 HashMap，实际瓶颈通常不在 CPU，而在模型加载和内存。模型文件压缩到几 MB 级别，预测跑起来会轻松很多。

5. 运行避坑：伪分布式内存、分词包与模型分发的 5 个经典问题

5.1 伪分布式搭建常见失败：NameNode 起不来和 YARN 内存配置不合理

现象：按教程装完 Hadoop，执行jps发现少了 NameNode，日志里报 Java heap space；或者 NameNode 起来了，提交训练 Job 后 ApplicationMaster 反复重启。原因分两种：一是hadoop-env.sh里没显式设置堆内存，默认值在你的笔记本上可能过大；二是/etc/hosts里主机名解析到了127.0.1.1，NameNode 和 DataNode 进程互相找不到对方。

解决时先改内存，在hadoop-env.sh里加一行HADOOP_HEAPSIZE=512，把 NameNode 和 DataNode 的堆压到 512MB，伪分布式完全够用。然后确认etc/hadoop/core-site.xml里fs.defaultFS用的是hdfs://localhost:9000，hdfs-site.xml里副本数设为 1。最后记得hdfs namenode -format只执行一次，反复格式化会导致 DataNode 的 namespaceID 对不上，启动后 DataNode 会一直报错退出。

伪分布式环境跑这个项目，本身就是单机跑多个 JVM 进程，内存是最大的敌人。训练阶段 Reducer 要把词表缓存到内存，预测阶段每个容器又要加载一遍模型文件。我先调好 YARN 内存再调 JVM 堆，顺序不要反过来。

5.2 中文分词包缺失：NoClassDefFoundError 的根源是 classpath

现象：把第 3 章的正则分词换成 IKAnalyzer 后，提交 Job 运行，Mapper 阶段立刻报ClassNotFoundException: org.wltea.analyzer...。原因很直接：你本地 IDE 能跑是因为 classpath 里有这个 jar，但hadoop jar提交的是你自己的业务 jar，第三方依赖不会自动跟着进集群。解决有两个方向，选一个就行。

一是打包时用 Maven Shade 插件把依赖打成一个 fat jar，业务 jar 里直接内嵌 IKAnalyzer 的 class。二是提交命令加-libjars参数，把 IK 的 jar 放到客户端的 classpath 中再随任务分发。我倾向于 fat jar，因为课程设计项目本身不大，fat jar 多几百 KB 无感知，而且-libjars在不同 Hadoop 版本里的路径规则偶尔有差异，排查起来费时间。

这里不要忘了事情的另一面：分词器替换后，训练和预测的切词结果必须完全一致。IKAnalyzer 的版本不同，切词结果都可能有差异，更别提正则分词和 IK 混用。我在实际项目里吃过这个亏，训练用 IK、预测忘了换，结果模型词表和预测词表对不上，准确率直接从 0.85 掉到 0.5 左右。

5.3 概率计算出现 NaN 与负无穷：log 世界的边界行为

现象：模型文件里出现-Infinity，或者预测结果里所有文档的得分都一样，分到同一个类别。原因：模型输出时如果某个类别没有任何词条计数，classWordTotal为 0，分母变成alpha * vocabSize，理论上不会负无穷；真正出问题的是预测端，某个词在 condProbs 中存在，但遍历某个类别时取不到对应的 log 概率，如果你写的代码用Math.log(0)兜底，就会得到负无穷。

一个类别只要有一个词的概率是负无穷，整个类别的得分就变成负无穷，最后所有类别得分全是负无穷，max取到的永远是先验最大的类。解决方式是第 4 章代码里那种“找到才累加、找不到就跳过”的逻辑：wordProbs.get(label)返回空时，不要用 0 概率参与计算，直接 continue。这里不是数学问题，是对稀疏数据的容错问题。

5.4 YARN 容器内存超限：任务被反复 kill 而不是报错

现象：日志里出现Container killed by the ApplicationMaster或Physical memory limit exceeded，Job 在 map 阶段反复失败重试。原因：每个 Mapper 的 setup 要把整个模型文件加载进 HashMap，默认容器内存只有 1GB，而 JVM 的堆外内存、Metaspace、Hadoop 自身开销都没算进去。你的模型文件即使只有 10MB，转成 Java 对象后的内存占用也可能膨胀好几倍。

解决方式是给 map 和 reduce 分别调大容器内存，同时把 JVM 堆设置成容器内存的 60% 到 70%。我常用的配置组合：

注意只改.java.opts而不改.memory.mb会直接把容器内存撑爆，这是一个反向翻车的典型操作。提交命令里用-D传参最省事，改配置文件容易影响集群里其他任务。

5.5 模型文件“进不去”Mapper：DistributedCache 的版本差异

现象：setup 里调用context.getLocalCacheFiles()返回null，或者代码直接编译不通过。原因：getLocalCacheFiles()在 Hadoop 2.x 后期就被标记废弃，3.x 版本彻底移除，很多老博客里的代码照抄必然翻车。

我现在的写法统一是第 4 章那套：提交命令用-files hdfs:///path/model.txt#model.txt，setup 里直接用new FileReader("model.txt")读本地文件。这个方案绕开了 DistributedCache 的 API 差异，只依赖-files参数本身的行为，而它从 Hadoop 1.x 到 3.x 语义一直稳定。如果模型由多个 part 文件组成，先把它们getmerge成一个文件再分发，不要试图用-files传目录，这个参数不支持目录递归。

6. 提升分类效果的小技巧：特征词剪枝、增量更新与五分钟混淆矩阵

6.1 特征词剪枝：把模型文件压缩一个数量级的正确姿势

模型文件里存着全部词汇表中每个词在每个类别的概率，行数等于词汇表大小乘以类别数。大量低频词只在训练集里出现一两次，对预测几乎没有区分能力，却占满了内存和模型文件体积。剪枝的做法是在第 3 章 Reducer 的 cleanup 里加一步：遍历wordCount，只保留每个类别下词频最高的前 N 个词，N 通常在 2000 到 10000 之间。剪掉的长尾词不再输出到模型，预测端查询时自然跳过。注意这个剪枝必须以类别为单位做，而不是全局 top N，否则样本量大的类别会占满所有名额，小类别直接被删空。

6.2 增量更新：新数据来了不用每次全量重训

朴素贝叶斯的本质是计数，所以增量训练理论上可以做成“旧统计 + 新统计”的合并。最简单的落地方式是把第 3 章的 Mapper 改成每批新数据单独产出一份计数目录，然后写一个脚本把旧模型的计数文件和新计数文件相加，重新计算概率。前提是分词逻辑和类别集合完全不变，否则计数对不上。实际做的时候，我发现维护增量逻辑的复杂度比全量重训高不少，课程设计阶段还是全量重训更稳。

6.3 用 awk 一句话生成混淆矩阵

预测结果四列格式里，第二列是真实类别，第三列是预测类别。用一个 awk 命令就能统计混淆矩阵：

hadoop fs -cat /user/hadoop/nb/result/part-m-00000 | \ awk -F'\t' '{matrix[$2","$3]++; total[$2]++} END { for (real in total) { printf "%s\t", real; for (pred in total) printf "%d\t", matrix[real","pred]+0; printf "\n"; } }'

输出每行是一个真实类别，每列是一个预测类别，对角线是正确分类的数量。我每次调完参数都先跑这个命令，一眼就能看出哪两个类别在互相打架。比如“财经”和“互联网”容易混，说明这两个类的训练语料有大量公共词，需要回头看看是不是停用词没过滤干净。这个方法比看准确率数字直观得多。

我现在的习惯是拿到任何一份 Hadoop 分类项目，第一件事先看两处：训练和预测的输入分隔符是不是统一的 TAB，模型文件在预测 Job 里是怎么被读取的。这两个地方定了，后面的问题基本都是看日志就能解决。希望帮到你。

本文还有配套的精品资源，点击获取