企业官网建设流程全解析

1. 为什么非要用自定义Source：内置组件解决不了的采集场景

先说一个我实际碰到的场景。当时要采集一批注塑机控制器输出的生产日志，格式不是常见的逐行文本，而是“长度字段 + JSON体 + 二进制状态块”拼接的记录结构。每条记录不定长，记录内部还夹带不可见字符，用Flume自带的TaildirSource按行读，读出来全是截断的乱码；用SpoolingDirectorySource做目录监控，它对文件改名有严格要求，而且同样只能逐行解析，完全没办法处理这种“需要先读长度、再按长度切块、还要跳过二进制段”的规则。

那会儿团队里有人提出先写个解析程序把日志转成标准格式，再丢给Flume去采集。这条路当然能走通，但等于在数据链路上多插了一个中转点，程序崩了没人知道、日志多了来不及转、还要单独部署维护。后来我翻了一下Flume的源码结构，发现与其绕路，不如直接在Source层把事干了——这就是写自定义Source的初衷。

所以，什么时候需要自定义Source，其实有一条很清晰的判断标准：当你要采集的数据在“读取方式”“记录切分规则”“编码处理”这三个层面有任何一个无法用内置的TaildirSource、SpoolingDirectorySource、NetcatSource完成时，就可以考虑写了。典型场景包括：

• 日志里有多条记录粘在一起，需要用特定分隔符或长度字段切分
• 源文件被持续覆盖写入，而不是append-only，Taildir的inode+offset续传会失效
• 数据源是二进制格式，比如自定义协议的网络报文、设备串口输出
• 采集前需要根据内容做过滤、解密、解压，而且过滤逻辑依赖记录内部的字段

这些需求如果用Flume的Interceptor在Source后置环节处理也能做，但有些操作必须发生在“读文件之前”。比如按长度偏移量定位记录块，必须让Source在读取时就精确控制字节位置，否则后面无论怎么洗数据都找不回丢失的信息。只有自定义Source能同时把控“读”和“解析”两个环节。

1.1 自定义Source的核心运行机制先搞清楚

在动手写代码之前，得先弄清楚Flume里Source到底是怎么跑起来的。Source在Flume架构里负责从外部系统拉取数据，把它封装成Event，再通过ChannelProcessor写入Channel。ChannelProcessor内部连接着Interceptor链和Channel选择器，这些都不需要你操心，你的代码只需要负责一件事：把外部数据变成Event列表，传给ChannelProcessor。

Flume的Source接口本身定义了start()、stop()、process()三个方法，其中process()是核心循环入口。但实际开发中我们一般不会直接实现Source接口，而是继承AbstractSource类。这个类已经帮你实现好了ChannelProcessor的关联和配置方法，省掉很多样板代码。

整个运行循环是：Flume的SourceRunner线程反复调用process()，每次返回后根据返回值决定是继续调用还是休眠。PollableSource接口规定了三类返回值：

• Status.READY：本次处理完了数据，立刻继续下一轮
• Status.BACKOFF：本次没拿到数据，或者源端暂时不可读，需要按退避算法休息一下再试
• Status.READY后抛出异常：Flume会捕获并打印，自身线程不会挂掉，但日志会持续报错

很多第一次写自定义Source的人容易犯的错，是不管有数据没数据，process()里永远返回READY。结果就是CPU空转，一个采集进程占掉一整颗核。正确做法是识别“暂时没数据”这个状态，按指数退避返回BACKOFF。

1.2 PollableSource与EventDrivenSource的选型差异

自定义Source有两条路线：轮询型和事件驱动型。实际项目里我强烈建议优先走PollableSource，也就是轮询型。原因很现实：你写的Source要应对的是不可控的外部环境，文件可能没写完、网络可能闪断、上游设备可能发送一半就停了。轮询模型天然适合这种“不知道什么时候有数据，但有数据就赶紧拿”的场景，因为你可以在process()里自由控制每次最多读多少、读不到就退避，逻辑简单直接。

EventDrivenSource适合那种“外部系统主动把数据推给你”的协议场景，比如你自己实现了一个TCP服务端，靠Netty接收上游主动上报的数据。这种模型的代码结构完全不同，需要自己管理连接生命周期的回调，复杂度高得多。对于绝大多数“读取特定格式日志文件”的场景，PollableSource就够了。

2. 从零实现：按偏移量精确读取并切分记录的自定义Source

这一节我会把一套完整可用的代码拆开讲。这套代码解决的是这样一个需求：数据源是一个持续增长的二进制日志文件，记录结构为“4字节长度 + 数据体”，数据体内部是UTF-8编码的文本，每条记录被一个0x1F分隔符正着结尾。要求Source从指定偏移量开始读，采集进度实时记录到本地偏移量文件，进程重启后能续传。

2.1 工程结构与依赖准备

先建一个标准的Maven工程，结构如下：

flume-custom-source/ ├── pom.xml └── src/main/java/com/example/flume/source/ ├── LengthPrefixedFileSource.java └── LengthPrefixedFileSourceConfig.java

pom.xml里核心依赖只有两个，以下是关键内容：

<dependencies> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-sdk</artifactId> <version>1.9.0</version> <scope>provided</scope> </dependency> </dependencies>

注意依赖的scope必须是provided。因为这个jar最终要放到Flume的plugins.d目录下由Flume进程加载，如果打成一个包含依赖的大jar包，反而容易和Flume自带的Guava、Avro等库产生冲突。只用provided的话，Maven打包时不会打入这些依赖，天然规避了冲突问题。

实际开发中我还会加一个maven-shade-plugin用于重命名冲突包，不过那是后面踩了坑再补的。第一版工程保持最简，能编译跑通最重要。

2.2 配置项设计与参数透传

自定义Source要能通过Flume的配置文件接收参数，必须实现Configurable接口的configure()方法。我在设计配置项时坚持一个原则：所有和现场环境相关的参数都放在配置里，而不是写死在代码里。文件路径、偏移量保存路径、编码类型、单次最大读取条数，全部做成可配置。

public class LengthPrefixedFileSourceConfig { private String filePath; private String offsetFilePath; private String charset = "UTF-8"; private int batchSize = 100; private long maxBackoffSleepMillis = 2000; private long readTimeoutMillis = 3000; // getters and setters... }

于是configure()里只需要做字符串到配置对象的转换：

@Override public void configure(Context context) { config = new LengthPrefixedFileSourceConfig(); config.setFilePath(context.getString("filePath")); config.setOffsetFilePath(context.getString("offsetFilePath")); config.setCharset(context.getString("charset", "UTF-8")); config.setBatchSize(context.getInteger("batchSize", 100)); config.setMaxBackoffSleepMillis(context.getLong("maxBackoffSleep", 2000L)); }

这段代码有几个关键点：

• context.getString("filePath")取不到值会返回null，所以启动后要在start()里做空值校验，抛异常要带清晰提示，别等到运行时NPE才发现配置写错了
• batchSize是单轮process()里最多写入Channel的Event条数，这个值直接决定吞吐量，后面第五节细说
• maxBackoffSleep是没数据时的最大休眠间隔，本质是给Flume的重试机制踩刹车

配置参数名要和Flume配置文件中a1.sources.s1.xxx的xxx保持一致。很多人在这里忽略了一个细节：Flume的Context里除了你自己的自定义参数，还会传入type这个参数，所以configure()里第一件事应该是确认type和当前类名匹配，防止整个agent文件里多个Source共用一个配置段时发生串参数。

2.3 核心采集逻辑：读长度、切记录、产Event

接下来是核心的process()实现。整体流程是：打开文件句柄、跳到上次记录的偏移量、读取长度字段、按长度截取数据体、切掉结尾分隔符、构造成Event、批量提交、更新偏移量并保存。代码结构并不复杂，但每一步都必须考虑边界情况。

@Override public Status process() throws BackPressureException { if (raf == null) { openFile(); } ChannelProcessor channelProcessor = getChannelProcessor(); List<Event> events = new ArrayList<>(config.getBatchSize()); try { for (int i = 0; i < config.getBatchSize(); i++) { byte[] lengthBytes = readFully(4); if (lengthBytes == null) { break; // 文件暂时没有更多数据 } int recordLength = ByteBuffer.wrap(lengthBytes).getInt(); if (recordLength <= 0 || recordLength > 16 * 1024 * 1024) { throw new IOException("Invalid record length: " + recordLength); } byte[] recordBytes = readFully(recordLength); if (recordBytes == null) { seekBack(lengthBytes.length); // 长度字段读到了但数据体不完整，回退 break; } byte[] body = stripDelimiter(recordBytes); Event event = EventBuilder.withBody(body, Charset.forName(config.getCharset())); Map<String, String> headers = new HashMap<>(); headers.put("sourceFile", config.getFilePath()); headers.put("recordOffset", String.valueOf(offset)); event.setHeaders(headers); events.add(event); offset += 4 + recordLength; } if (events.isEmpty()) { Thread.sleep(computeBackoff()); return Status.BACKOFF; } channelProcessor.processEventBatch(events); persistOffset(offset); return Status.READY; } catch (IOException e) { logger.error("Read file failed", e); return Status.BACKOFF; } }

这里有几个非常容易写错的点：

readFully不是简简单单read一次就够了。Java的InputStream.read()不保证一次读满指定长度的字节，必须用循环保证实际读到的字节数等于请求的数，否则长度字段读一半就去解析，得到的是错误数据。所以readFully要自己实现：

private byte[] readFully(int length) throws IOException { byte[] buffer = new byte[length]; int read = 0; while (read < length) { int result = raf.read(buffer, read, length - read); if (result == -1) { return null; // 文件结束但没有读到完整数据 } read += result; } return buffer; }

“读到了长度字段但数据体不完整”这个场景是最隐蔽的坑。日志文件在写入过程中，进程可能在写长度字段和数据体之间被暂停，你恰好在这个间隙发起读取，就会出现长度字段有值、后面内容不够的情况。如果此时直接把读到的残缺数据当作一条Event发出去，就是脏数据。我的处理方式是把文件指针回退到读取之前的位置，下轮循环再试。

回退操作在RandomAccessFile里就是raf.seek(offset)，但要注意必须在更新offset变量之前做，别把内存里的偏移量也改了。这类边界逻辑建议配合单元测试去验证：预先构造一个尾部只有长度字段没有数据体的文件，确认返回的是BACKOFF且文件指针没有前进。

2.4 状态保存与断点续传设计

采集系统一旦重启，最怕的不是丢数据，而是重复读数据。对日志类数据来说，重复往往比丢失更隐蔽，因为下游的统计任务会把同一条记录算两次，导致指标失真。

我的做法是维护一份独立的偏移量文件，每次成功写入一批Event后就把最新的offset刷进去。这样进程崩溃最多丢最后一次成功提交以后的那批数据，但不会出现大范围重复采集。一个微妙问题是，Flume本身的可靠性模型是“At least once”，也就是说Channel到Sink的传递可能发生重复，但Source端的重复要尽量减少。

private void persistOffset(long offset) throws IOException { Path path = Paths.get(config.getOffsetFilePath()); Path tmp = Paths.get(config.getOffsetFilePath() + ".tmp"); Files.write(tmp, String.valueOf(offset).getBytes(StandardCharsets.UTF_8), StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING); Files.move(tmp, path, StandardCopyOption.ATOMIC_MOVE); logger.debug("Persisted offset: {}", offset); }

先写临时文件再原子改名，是为了防止进程在写偏移量文件写到一半时崩溃，留下一个半截内容，下次启动解析时直接NumberFormatException。虽然这个概率不高，但采集进程长年累月跑着，偶发性故障累积起来就很可观，能用一个原子操作规避的事不值得省。

启动时读取偏移量的逻辑也要做容错：偏移量文件不存在时从头开始读；存在但内容非法时打印告警并从0开始；文件指向的位置已经超出当前目标文件长度时，说明日志文件被重置过，同样从0开始。

3. 部署接入：从打包到Flume agent配置的完整链路

代码写完之后，部署方式是否规范，决定了后面踩坑的难易程度。Flume从1.7开始全面支持plugins.d目录模式，自定义Source的jar应该放在这里而不是直接扔进lib下。正确目录结构是这样的：

/opt/flume/plugins.d/custom-source/ ├── lib/ │ └── flume-custom-source-1.0.jar └── libext/ └── (可选的其他依赖jar)

lib放你自己的工程jar，libext放这个Source额外依赖的第三方库。如果你的工程用到了某个特定版本的JSON库、序列化库，而Flume核心lib里也有这个库的不同版本，那就该把依赖放libext里，利用Flume的隔离类加载机制，避免和核心库互相踩。

打包时我一般直接执行：

mvn clean package

然后把target/flume-custom-source-1.0.jar复制到上面的目录下，重启Flume agent。注意这里不需要使用mvn package后那个带依赖的jar包，因为它会把flume-ng-core也打进去，造成类重复加载。

3.1 agent配置文件里的Source段写法

接着配置flume.conf里的Source段：

a1.sources = src1 a1.sources.src1.type = com.example.flume.source.LengthPrefixedFileSource a1.sources.src1.filePath = /data/logs/machine_ctl.dat a1.sources.src1.offsetFilePath = /data/flume/offsets/machine_ctl.offset a1.sources.src1.charset = UTF-8 a1.sources.src1.batchSize = 200 a1.sources.src1.maxBackoffSleep = 3000 a1.channels = ch1 a1.channels.ch1.type = file a1.channels.ch1.checkpointDir = /data/flume/checkpoint a1.channels.ch1.dataDirs = /data/flume/data a1.sinks = sk1 a1.sinks.sk1.type = hdfs a1.sinks.sk1.hdfs.path = /data/flume/events/%Y%m%d a1.sinks.sk1.hdfs.fileType = DataStream a1.sinks.sk1.hdfs.rollInterval = 600 a1.sources.src1.channels = ch1 a1.sinks.sk1.channel = ch1

这里有两个容易混淆的地方。

type参数的值填的是自定义Source的全限定类名，而不是什么简短别名。Flume加载时会根据这个字符串去plugins.d目录下找对应类，找不到就启动失败。

Channel我选了file类型。这是数据可靠性要求较高时的常见选择：Source写入Channel成功后再更新偏移量，至少保证数据进了持久化队列；进程崩溃恢复时，Flume会从Channel checkpoint恢复未发送的数据，Source则从偏移量文件续传。两者配合，数据丢失的概率被压到非常低。

3.2 启动验证与日志定位的三板斧

部署完成后，验证方式很有讲究。我每次都会按这个顺序排查：

第一，看Flume进程端的日志有没有“Source started”这种信息，确认类加载成功。如果有ClassNotFoundException，先检查jar是否真的在plugins.d/custom-source/lib下，再检查jar内部META-INF/MANIFEST.MF里的Class-Path有没有写错。

第二，手动往目标文件里追加几条构造好的记录，观察Channel的容量监控或者Sink写入的文件内容。这一步建议用bin/flume-ng agent -n a1 -c conf -f ./conf/flume.conf -Dflume.root.logger=DEBUG,console以前台模式跑，直接看控制台日志，比翻日志文件快得多。

第三，检查偏移量文件是否按预期更新。这是判断Source是否真正工作的核心指标——偏移量往前走，说明解析逻辑通了；偏移量不动，说明数据没被识别为完整记录，多半卡在长度解析或边界判断上。

我第一次部署时就在第三步卡了很久：偏移量文件一直是0，但用手动构造的数据测试时明明能采集。后来发现是测试文件本身的问题——用手写文本编辑器生成的文件末尾没有正常换行，而我的解析逻辑要求完整记录长度字段，测试数据的长度根本没有写进文件头。这个场景和“日志写入一半”是同一个问题，提醒我测试数据必须模拟真实生产过程，不能拿手工文件将就。

4. 线上运行半年后：四个高频故障的完整排查链路

自定义Source真正上线以后，故障排查的挑战就不在“写代码”而在“定位问题”。以下四个问题是运行期最常碰到的，我把完整的排查链路写出来，照着这个思路走能少走很多弯路。

4.1 采集进程CPU占用100%：退避策略失效

现象：机器监控面板显示Flume进程CPU持续跑满，但采集量并不大。

排查链路：第一步用top -Hp <pid>找到具体线程，看到进程里有一个线程的CPU时间累积特别高。第二步用jstack导线程栈，发现栈顶停在自定义Source的process()里，进一步往上翻是raf.read()调用。第三步看代码，发现process()里文件暂时没有数据时直接return Status.READY。

根因：Flume的PollableSourceRunner收到READY后立即再次调用process()，如果readFully在文件末尾返回了null但被处理为“继续循环”，线程就在文件句柄和空转之间高速切换，CPU自然被打满。

修复：区分“文件暂时无数据”和“读取成功”两种状态。无数据时获取当前时间，按指数退避算法计算休眠时长，最小200ms，最大不超过配置的maxBackoffSleep，然后Thread.sleep再返回BACKOFF。

我当时还踩了一个隐藏的连环坑：Thread.sleep在process()里引发InterruptedException，直接往外抛会导致Flume把这次调用标记为异常，触发更激进的重试。所以正确写法是捕获中断异常后恢复中断标志，再正常返回BACKOFF。

4.2 采集到的日志乱码和行数对不上：编码处理与记录切分

现象：HDFS上的输出文件打开后中文字符变成乱码，且总行数比源端业务系统记录的日志条数少。

排查链路：第一步对比源文件字节数和已采集字节数的大小，确认不是“读少了”。第二步取出其中一条Event的body，用十六进制格式打印前100字节。第三步比对原始文件的对应区域，发现每条记录的末尾多了一个字节的0x1F，而业务系统统计条数时把0x1F当作分隔符，不会把它算进正文。

根因：我在设计时保留了原始记录末尾的分隔符，想着下游处理时统一切分。但实际下游是按\n去split的，0x1F混在正文里，导致每行末尾多一个不可见字符，部分数据清洗组件选择直接解码显示成乱码。

修复：在stripDelimiter里显式去除记录末端的0x1F，并把分条逻辑处理干净。这个案例给了一个通用教训：Source层最好只负责“把数据装进Event”，不要在Body里保留协议层面的元信息，除非下游明确约定要使用它。

4.3 偏移量与实际文件长度不一致导致重复采集

现象：某次日志文件在凌晨被运维脚本truncate -s 0清空，第二天发现同一台设备的数据重复采集了一批。

排查链路：第一步检查偏移量文件内容，发现记录了很大的offset。第二步检查设备日志文件，发现文件被清空后重新开始写入，文件长度从头增长。第三步回头看源码里的启动逻辑：启动时判断offset > 文件长度时没有做任何处理，直接seek到那个偏移量，结果文件读取位置超出文件末尾，readFully永远返回null，数据一直采不进来。运维重启Flume后发现文件里已经有新数据，但偏移量文件还是旧位置，于是从头开始读，导致重复。

根因：偏移量语义和文件生命周期没有挂钩。对持续追加型文件来说，offset只对同一个文件实例有效。文件一旦被truncate或重建，旧offset就失去了参照意义。

修复：启动时增加“offset合法性校验”，比较目标和偏移量的大小关系。如果偏移量超过文件当前长度，说明文件被重置过，记录一条WARN日志后从文件头重新采集。这个方法不完美，但对绝大多数“日志文件被清空重建”的场景已经足够。如果要更精确，还可以记录文件inode或文件名+大小快照，但那引入了额外状态，复杂度会明显上升，除非业务非常依赖精确断点，否则不建议一开始就做成重状态。

4.4 数据进入到了Channel但Sink没写出去：与Channel容量的拉锯战

现象：Source日志显示每轮都成功写入了一批Event，但HDFS上的最新文件长时间没有更新。

排查链路：第一步检查Sink日志，发现大量Channel is full之类的警告。第二步看Channel监控指标，发现ChannelCapacity已满，而Sink写入速度远低于Source产出速度。

根因：自定义Source把batchSize设得过大，一次性把大量Event灌进Channel，而下游HDFS Sink在滚动文件时写IO遇到瓶颈，吞吐跟不上，直接把Channel塞满了。Channel满了之后Source再processEventBatch就会阻塞或异常，Source线程卡在Channel写入上，整体链路瘫痪。

修复：把batchSize从1000降到200，同时给Channel配置合理的容量与事务容量，再对比Source与Sink两侧的吞吐指标做调优。这条经验很重要：自定义Source的产出速度是可以非常快的，因为它在进程内直接向Channel写数据，没有任何网络开销，但这不意味着就该让它以极限速度产出。数据链路的整体吞吐取决于最慢的那个环节，Source的速度必须和Channel、Sink匹配，不然制造问题再解决问题，纯属给自己添堵。

5. 进阶优化：把自定义Source推向生产级水准的思路

基础的Source跑通后，生产环境还有几个方向值得再打磨。这些内容不是“必须做”，而是“根据场景取舍”，我按实用性排序介绍一下。

5.1 引入背压机制，别让Source无脑塞

Flume 1.9版本后支持背压，当Channel达到容量上限时，processEventBatch会抛出BackPressureException。自定义Source的process()方法签名里已经带了throws BackPressureException，但很多人的实现里根本没有捕获或感知它。

正确做法是：当捕获到BackPressureException时，立即停止继续读取新数据，更新偏移量持久化，返回BACKOFF。这比依赖Flume默认的重试策略更主动——你说到底，Source是你写的，只有你最清楚哪些状态可以安全回退、哪些状态需要优先保数据。

5.2 用SourceCounter做全链路指标监控

AbstractSource里没有直接提供计数器，但你可以在start()里实例化一个SourceCounter，在关键节点埋点：读了多少字节、构造了多少Event、处理失败多少条、事务提交多少次。这样在Ganglia或Prometheus的监控大盘上，你能直接看到采集吞吐的波动曲线，而不是等数据丢了之后才靠人工翻日志。

private SourceCounter sourceCounter; @Override protected void start() { sourceCounter = new SourceCounter("custom-source"); sourceCounter.start(); // ... } // process() 循环中 sourceCounter.incrementEventReceivedCount(events.size()); sourceCounter.addToEventReceivedByteCount(totalBytes);

5.3 多文件轮询与线程模型设计

如果不止一个日志文件需要采集，最简单的扩展是同一个Source实例里维护一个文件列表，轮流检查每个文件是否有新数据。这里要特别注意：多个文件的偏移量状态要独立保存，不能所有文件用一个偏移量。我推荐把偏移量文件路径设计成和目标文件一一对应，比如目标文件名为a.dat，偏移文件就叫a.dat.offset，用这种约定式命名减少配置管理的负担。

真的要做高吞吐采集时再考虑多线程模型，但那已经不是Source内部能解决的事，需要同时调整Channel的事务容量和SourceRunner的线程数。实际操作中，一个JVM里跑几个独立的agent进程往往比单个agent内加线程更可控，至少故障隔离范围更清晰。

5.4 记录级别过滤和改造：把Interceptor的活前置到Source里

常规Flume链路中Interceptor可以做数据清洗，但每个Interceptor处理的是已经成形的Event。某些格式下，是否丢弃一条记录取决于源文件里前后的关联信息，比如记录A和记录B成对出现，A的头部包含B的校验值。如果等Interceptor处理，你需要缓存跨Event的状态，实现起来很别扭。

这时可以在自定义Source内部先做一次“内容级过滤”，把满足条件的记录构造成Event，其余直接跳过，同时更新偏移量。这么做等于把过滤逻辑下沉到采集层，下游链路更干净，调试思路也更清晰。但要注意区分职责：涉及跨Event内容的复杂状态匹配逻辑，还是留到专门的流处理框架里做比较合适，Source里做太多运算会影响采集性能。

6. 写在最后：自定义Source的正确打开方式

回头再看这个完整项目，我的体会很直接：自定义Source不是银弹，它只解决“Flume预置能力覆盖不到的那部分数据接入需求”，但它也绝不是洪水猛兽，整个实现过程比我预想的要短得多。关键在于抓住三件事：第一，好好设计配置项，把现场环境的不确定性都挡在代码外面；第二，严格处理边界状态，文件被截断、读到半个记录、offset文件损坏这类情况都要有明确的对策；第三，监控优先于优化，先把SourceCounter埋进去，把采集行为实时可视化，再谈压不压性能的问题。

最后再分享一个我在运维中习惯的做法：每次升级Source代码，都先用一段带标记的测试数据验证“新旧偏移量兼容性”。具体来说，就是先跑旧版本采集完一部分数据，记录偏移量；然后再跑新版本，观察它能否从同一个偏移量继续，而不是重读一遍。这种回归测试不花多少时间，远远好过上线后半夜被告警叫醒、才发现重复数据把整个数仓的指标都污染了再回头清理。自定义Source好不好用，只有长期运行过才知道，而长期运行的前提，就是一开始就把边边角角都想清楚。