实时计算入门：Flume到Kafka再到Spark与Flink的经典链路实战

直接说结论如果只选一条技术主线入门实时计算Flume → Kafka → Spark → Flink 这套链路绝对值得死磕。这四个组件覆盖了数据从产生、采集、传输、缓冲到计算落地的完整闭环是大数据实时处理领域最经典的组合拳也是面试和实际项目中出现频率最高的技术栈。这篇内容我会按自己的实战路径来拆先讲清楚这四个组件在整个实时链路里各自扮演什么角色、为什么是它们组合在一起再分别把 Flume、Kafka、Spark、Flink 的核心机制和上手要点捋一遍附上可以直接抄作业的配置和代码最后给出一套从零搭建实时统计链路的完整案例以及我这些年踩过的坑合集。不管你是刚转行大数据的零基础还是想系统梳理实时计算知识体系的开发这篇都能当一份带注释的实战地图用。1. 整体技术链路拆解四个组件到底在干嘛很多初学者一上来就分别学 Flume、Kafka、Spark、Flink每样都看了一遍文档但还是不知道它们之间怎么配合。我建议反过来先站在一条完整数据流的角度看全局。一条典型的实时数据链路是这样的业务日志/埋点数据 → Flume 采集 → Kafka 消息队列 → Flink/Spark 实时计算 → 结果存储/告警/大屏这个链路里的每一步解决的都是不同的问题。用餐厅出餐来打比方Flume 是后厨门口收菜的伙计负责把原料从各个供应商那里搬进厨房Kafka 是厨房和传菜口之间的备餐台原料先放备餐台上厨师随用随取生意再火爆也不会把后厨挤爆Spark 和 Flink 则是两个风格不同的厨师班组一个擅长把一堆订单攒一波再一起炒微批处理另一个来一个订单炒一个真流式处理。1.1 为什么是这四种技术组合选这四个组件不是偶然它们各自占据了链路中不可替代的位置。Flume 解决的是数据怎么稳定进管道的问题。日志文件、网络端口、Tail 目录这些数据源Flume 天生就支持配置一个 agent 就能持续不断地把增量数据送进 Kafka不需要你自己写轮询脚本。Kafka 解决的是数据怎么缓冲和削峰的问题。实时数据往往有不均匀的洪峰比如电商大促、热搜事件直接打到计算引擎容易引发雪崩。Kafka 把数据持久化到磁盘消费者按自己的节奏拉取天然就是一道缓冲大坝。Spark 和 Flink 解决的是数据怎么算出价值的问题。Spark Streaming 和 Structured Streaming 基于微批micro-batch吞吐量大、生态成熟Flink 则是真正的流式计算引擎毫秒级延迟、状态管理强大、精确一次语义适合对延迟和准确性要求极高的场景。1.2 零基础应该先学哪个我的建议是按数据流动方向学先 Flume再 Kafka然后 Spark最后 Flink。这个顺序最符合认知逻辑——每学一个组件你都能清晰地知道数据从哪来、到哪去、被谁消费。很多教程建议先学 Spark 再学 Flume理由是 Spark 更有名气。但实际效果往往不好因为脱离了数据源Spark 只能拿本地 fake 数据练手理解不深。先搞定 Flume 和 Kafka你就拥有了源源不断的真实数据再上手 Spark 和 Flink 的时候跑的是自己搭的实时管道那种通了的感觉是拿假数据练不出来的。提示学习阶段建议在一台 16G 内存的机器上用虚拟机或 Docker 搭伪分布式环境不需要一上来就上三台服务器。我初学时因为强行搭三节点集群光调试网络就耗了一周性价比极低。2. Flume 实战半小时搞定日志采集Flume 是 Apache 基金会的分布式日志采集系统核心抽象就是 Agent。一个 Agent 由三个部件组成Source数据源、Channel缓冲区、Sink数据出口。2.1 Flume 核心概念与配置结构Source 负责读取数据常见的有spooldir监控整个目录、taildir支持断点续传地 tail 文件生产环境最常用、exec执行命令获取输出比如 tail -F。Channel 是 Source 和 Sink 之间的管道常用的是memory channel快但可能丢数据和file channel慢但更可靠。Sink 决定数据去哪可以写 HDFS、Kafka、Hive、日志文件等。下面是我在项目里最常用的一份配置作用是把某个服务日志目录下的.log文件源源不断地采集到 Kafka# taildir-to-kafka.conf a1.sources r1 a1.channels c1 a1.sinks k1 a1.sources.r1.type taildir a1.sources.r1.positionFile /data/flume/position/taildir_position.json a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /data/logs/order-service/.*log a1.sources.r1.fileHeader true a1.channels.c1.type memory a1.channels.c1.capacity 10000 a1.channels.c1.transactionCapacity 5000 a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers node01:9092,node02:9092,node03:9092 a1.sinks.k1.kafka.topic order-access-log a1.sinks.k1.kafka.flumeBatchSize 100 a1.sources.r1.channels c1 a1.sinks.k1.channel c1启动命令flume-ng agent \ --name a1 \ --conf-file /data/flume/conf/taildir-to-kafka.conf \ --conf /opt/flume/conf \ -Dflume.root.loggerINFO,console这里有几个坑我要重点说。第一positionFile一定要配置否则重启 Flume 后会重复读一遍日志数据就重复了。第二memory channel的capacity和transactionCapacity关系是后者不能超过前者而且要根据日志量估算日志量大但 capacity 设小了会频繁报 Channel full 错误。第三KafkaSink 的flumeBatchSize决定批量发送的条数设太大会增加延迟设太小又浪费网络带宽我一般取 100 到 500 之间。2.2 Flume 生产级调优经验Flume 看起来简单但生产环境里跑起来很容易出幺蛾子最常见的就是采集速度跟不上日志产生速度。这时候优先检查两个地方一是 Channel 类型。memory channel快但容量有限如果日志洪峰特别猛优先考虑file channel它把数据刷到磁盘容量几乎是无限的。二是 Source 的监控粒度。taildir支持多文件组但如果文件组太多Flume 内部会用单线程轮询日志文件多时容易出现延迟。我一般会把文件数控制在一个组不超过 50 个。另外强烈建议给 Flume 配监控。Flume 自带 HTTP 监控端口在配置里加上a1.sources.r1.interceptors i1 a1.sources.r1.interceptors.i1.type org.apache.flume.interceptor.MonitorInterceptor$Builder或者直接用-Dflume.monitoring.typehttp -Dflume.monitoring.port34545启动用 Prometheus 拉指标。没有监控的 Flume 像个黑盒出了问题很难定位是 source 没读到还是 sink 写不出去。3. Kafka 实战消息队列的核心机制与集群部署Kafka 是整个实时链路的心脏几乎所有数据都要流过它。你要是不把 Kafka 的原理吃透后面 Flink/Spark 消费数据遇到 offset、分区、消费组问题时会非常痛苦。3.1 Kafka 中必须搞懂的五个概念我把 Kafka 里最核心的概念浓缩成五个主题Topic、分区Partition、副本Replica、偏移量Offset、消费组Consumer Group。Topic 是数据的逻辑分类。Partition 是物理分片一个 Topic 的数据被拆成多个 Partition 分散在集群多台机器上这是 Kafka 能横向扩展的基础。Replica 是分区的副本每个分区有 leader 和 followerleader 负责读写follower 负责同步leader 挂了会自动切换。Offset 是消息在分区内的位置编号消费者靠它记录读到了哪。Consumer Group 是消费者分组组内每个消费者负责不同的分区组间的消费者互不影响这是 Kafka 实现广播和队列两种模式的关键。用生活类比Topic 是杂志的种类Partition 是每一期的上下册Offset 就是书签Consumer Group 是读杂志的人分的小组小组内每人读不同的册子小组之间各读各的互不干扰。3.2 从零部署一套 Kafka 集群现在主流版本已经推荐使用 KRaft 模式代替 ZooKeeper 模式了。KRaft 模式去掉了独立的 ZooKeeper 集群Kafka 自身就能管理元数据部署和运维都简单很多。以三节点集群为例假设机器 IP 分别是 node01/02/03核心配置如下# config/server.properties process.rolesbroker,controller node.id1 controller.quorum.voters1node01:9093,2node02:9093,3node03:9093 listenersPLAINTEXT://:9092,CONTROLLER://:9093 advertised.listenersPLAINTEXT://node01:9092 log.dirs/data/kafka-logs num.partitions3 default.replication.factor2 offsets.topic.replication.factor2 transaction.state.log.replication.factor2在启动前先格式化存储目录kafka-storage.sh random-uuid kafka-storage.sh format -t uuid -c config/server.properties然后逐台启动kafka-server-start.sh -daemon config/server.properties验证集群状态kafka-metadata.sh quorum-status --bootstrap-server node01:9092 kafka-topics.sh --bootstrap-server node01:9092 --create --topic order-access-log --partitions 3 --replication-factor 2从 ZooKeeper 模式迁移到 KRaft 模式是目前的趋势新项目直接上 KRaft不要再走回头路。如果只是本机学习单节点 KRaft 就够用了把process.roles设为broker,controllernode.id设为 1controller.quorum.voters只写本机一个地址就行。3.3 生产环境 Kafka 参数调优方向Kafka 调优是面试重点我给出最常被问到的几个方向。吞吐量优先调大batch.size默认 16KB、linger.ms默认 0调大后允许攒批、compression.type设为lz4或zstd代价是延迟略微上升。延迟优先调小linger.ms生产端设置acks1或acks0但要注意数据丢失风险。消息大小默认单条消息最大 1MB如果业务需要传输大对象要同时调 broker 的message.max.bytes、topic 的max.message.bytes、consumer 的fetch.max.bytes三处否则消费端会一直拉取失败。注意Kafka 的迁移和升级尽量在低峰期操作。我遇到过一次线上跨大版本升级2.8 升 3.5因为没注意消息格式兼容性消费端反序列化直接报错最后只能回滚。跨大版本前一定先去官方文档查 Upgrade Guide。4. Spark 实战用 Structured Streaming 快速上手实时计算Spark 在实时计算领域的位置很有趣它核心是批处理引擎但通过 Spark Streaming 和 Structured Streaming 也提供了准实时能力。初学者不用纠结这个定位问题直接学 Structured Streaming 就好它是目前 Spark 实时计算的主流 API。4.1 Structured Streaming 的核心思想Structured Streaming 把实时数据流看作一张无限增长的表格。每来一批数据就像往表格里插入一行。你写的 SQL 或 DataFrame 操作就是在这张无限表上做查询Spark 会自动把查询转换成微批任务执行。这种设计的最大价值是你只需要写一份代码就能同时跑批处理和流处理。比如你写了一个统计订单金额的 DataFrame 逻辑跑在静态文件上就是离线报表跑在 Kafka 数据流上就是实时看板代码几乎一模一样。下面是我用 Structured Streaming 消费 Kafka 的一个典型示例统计每 1 分钟的订单金额from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, sum, window from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType spark SparkSession.builder \ .appName(OrderStreaming) \ .master(yarn) \ .config(spark.sql.shuffle.partitions, 50) \ .getOrCreate() schema StructType([ StructField(order_id, StringType()), StructField(user_id, StringType()), StructField(amount, DoubleType()), StructField(ts, LongType()) ]) df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, node01:9092,node02:9092,node03:9092) \ .option(subscribe, order-access-log) \ .option(startingOffsets, latest) \ .load() \ .selectExpr(CAST(value AS STRING) as json) \ .select(from_json(json, schema).alias(data)) \ .select(data.*) result df \ .withWatermark(ts, 1 minutes) \ .groupBy(window(col(ts), 1 minutes)) \ .agg(sum(amount).alias(total_amount)) query result.writeStream \ .outputMode(append) \ .format(console) \ .trigger(processingTime10 seconds) \ .start() query.awaitTermination()这段代码要跑起来需要把spark-sql-kafka-0-10依赖打进 classpath。启动提交命令示例spark-submit \ --master yarn \ --deploy-mode client \ --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 \ order_streaming.py4.2 Spark 实时计算几个容易踩的坑第一个坑是startingOffsets的语义。earliest表示从最早 offset 开始消费latest表示只消费新数据。如果你在测试阶段重启了应用而 offset 没有提交成功用latest会丢掉重启期间的数据用earliest又可能重复消费大量历史数据。生产环境建议自己在外部存储如 Redis维护 offset保证精确消费。第二个坑是水位线和延迟数据。withWatermark必须配合窗口聚合使用它告诉 Spark我最多容忍多少延迟的数据。设置太短会丢数据设置太长窗口不释放导致状态无限膨胀。新手默认 10 分钟起步再根据业务实际数据延迟调整。第三个坑是分区数不匹配。Kafka topic 的分区数和 Spark 的并行度直接相关。如果 topic 有 3 个分区但 Spark 端设了 50 个 shuffle 分区会白白增加网络和内存开销。需要根据实际数据分布和资源情况动态设置spark.sql.shuffle.partitions。5. Flink 实战真正的流式计算窗口、状态与精确一次Flink 在近几年的实时计算场景里几乎成了事实标准。它能做毫秒级延迟的真正流式处理有完整的状态管理和精确一次Exactly-Once语义还支持批流一体。虽然上手门槛比 Spark 高一些但它值得你投入时间。5.1 Flink 的核心抽象与执行模型Flink 的核心抽象是 DataStream API数据在 Flink 中被看作无限的事件流。每个算子Source、Map、KeyBy、Window、Sink都是一个并行任务任务之间通过网络传输数据。Flink 能实现精确一次的关键是三个机制Checkpoint检查点、Barrier屏障和状态后端。Checkpoint 是 Flink 周期性地把算子状态和 source offset 一起做快照Barrier 是插在数据流里的特殊标记它保证在恢复时数据正好在某个时间点对齐状态后端则负责存储这些状态。这个机制有点像录制视频时的存档点。游戏每隔几分钟自动保存一次进度如果中途挂了从最近存档点重新开始不会回到最开始。下面是一个 Flink 消费 Kafka、统计 1 分钟窗口订单金额的 Java 示例StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); env.getCheckpointConfig().setCheckpointStorage(hdfs:///flink/checkpoints); DataStreamString stream env.addSource(new FlinkKafkaConsumer( order-access-log, new SimpleStringSchema(), kafkaProps )); DataStreamOrderInfo orderStream stream .map(json - objectMapper.readValue(json, OrderInfo.class)) .assignTimestampsAndWatermarks( WatermarkStrategy .OrderInfoforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((order, ts) - order.getTs()) ); orderStream .keyBy(OrderInfo::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new TotalAmountAggregate()) .addSink(new FlinkKafkaProducer( order-total-result, new SimpleStringSchema(), kafkaProps ));这个例子里WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))的意思是允许事件时间最多迟到 5 秒超过 5 秒的迟到数据会被丢弃。这里的选择要结合业务容忍度来定不是越大越好。5.2 Flink 窗口和 Watermark 的本质窗口和 Watermark 是 Flink 初学者最容易绕晕的地方。我尽量用一句话讲透。窗口就是把无限流按时间切分成有限段。Flink 支持两类时间Processing Time处理时间按机器当前时间切和Event Time事件时间按数据里携带的时间戳切。做统计业务基本都用 Event Time因为数据可能因为网络延迟乱序到达按处理时间切窗口会算错。Watermark 是 Flink 用来衡量事件时间进度的机制。它表示在这个时间戳之前的乱序数据应该都已经到了。Watermark 当前观察到的最大事件时间 - 允许延迟。Flink 会等 Watermark 越过窗口结束时间时才触发窗口计算。举一个能看到效果的实验手动往 Kafka 里发送几条乱序数据事件时间分别是 10:00:50、10:00:20、10:01:10把允许延迟设为 5 秒观察窗口触发时机。你会清晰看到 Flink 如何等数据、触发窗口、处理迟到数据——这个过程理解了Flink 就算入门一半了。注意allowedLateness和 Watermark 里的乱序容忍是两个概念。前者指窗口触发后最多再等多久接收迟到数据后者决定窗口的触发时机。两者叠加实际窗口关闭时间是watermark 越过窗口结束时间 allowedLateness。5.3 Flink 生产环境必须做的几件事第一件事开启 Checkpoint 并配置状态后端。没有 Checkpoint 的 Flink 任务一挂就全丢基本不能上生产。状态后端大状态用 RocksDB普通场景用 HashMap根据状态规模选。第二件事处理背压。背压是 Flink 下游处理不过来、上游还在使劲发数据的现象。最常见的原因有两个Sink 写入外部系统太慢比如批量插入数据库每次只插一条或者 KeyBy 后某个 key 数据量巨大形成数据倾斜。排查时先看 Web UI 的 BackPressure 面板再定位到具体算子。第三件事配置优雅停机和 savepoint。生产环境升级 Flink 任务时不能直接 kill要执行flink stop -p /data/flink/savepoints jobId下次从 savepoint 恢复flink run -s /data/flink/savepoints/savepoint-xxxx jar包这样能做到无状态丢失的任务升级。6. 完整实战订单日志从日志文件到实时大屏的全链路搭建前面都是单独讲组件这一节我把它们串起来带大家完整搭建一个订单实时统计链路。目标是服务日志每产生一条订单数据大屏上的今日销售额 5 秒内就能更新。这是实时计算领域最经典的入门实战场景。6.1 链路设计与环境准备完整数据流向模拟订单程序写入日志文件 → Flume taildir 采集 → Kafka topic: order-access-log → Flink 消费 窗口聚合 → MySQL/Redis → 大屏展示准备环境最低配置建议组件版本建议说明操作系统CentOS 7.9 / Ubuntu 20.04虚拟机即可JDK1.8 或 11Flink 1.14 需要 JDK 11Flume1.9.0或 1.11Kafka3.5.x使用 KRaft 模式Spark3.5.x配 Hadoop 3.x 客户端Flink1.17.x稳定版即可6.2 逐步操作模拟数据产生到 Flink 消费第一步模拟订单日志。写一个简单的 Shell/Python 脚本每隔 1-2 秒生成一条 JSON 格式订单日志#!/bin/bash while true; do echo {\order_id\:\$(uuidgen)\,\user_id\:\user_$((RANDOM%1000))\,\amount\:$((RANDOM%10001)),\ts\:$(date %s%3N)} /data/logs/order-service/order.log sleep 1 done第二步启动 Flume执行前面给出的 taildir-to-kafka 配置。第三步创建 Kafka topic 并用命令行验证数据是否进来kafka-topics.sh --bootstrap-server node01:9092 --create --topic order-access-log --partitions 3 --replication-factor 1 kafka-console-consumer.sh --bootstrap-server node01:9092 --topic order-access-log --from-beginning --max-messages 5看到 JSON 数据输出说明 Flume → Kafka 链路已经通了。第四步引入 Flink 代码包用前面第 5 节的订单统计代码把 Sink 改成写入 MySQL 或 Redis。以 MySQL 为例Sink 里执行的 SQL 就是INSERT INTO order_stat (window_start, total_amount) VALUES (?, ?) ON DUPLICATE KEY UPDATE total_amount VALUES(total_amount);注意一定要做去重更新否则每个窗口累加会重复累计。第五步写一个轮询查询 MySQL 的脚本模拟大屏展示刷新。这里可以直接用现成的可视化工具如 Grafana配置一个 MySQL 数据源5 秒刷新一次订单金额统计就上屏了。6.3 链路联调中的关键验证点链路搭完后不要急着欢呼至少要做下面三个验证。验证一故障恢复。手动 kill 掉 Flink 任务等 30 秒再启动观察数据是否从最近 checkpoint 恢复、有没有重复统计。如果重复了检查 FlinkKafkaConsumer 的 offset 提交策略和 checkpoint 是否同时开启。验证二数据洪峰。把模拟脚本改成每 0.1 秒写一条数据持续 5 分钟看 Kafka 有没有堆积、Flink 有没有背压、MySQL 写入有没有瓶颈。这个测试能暴露 90% 的隐性性能问题。验证三消息格式变更。改一版日志 JSON 结构比如加一个字段看 Flink 会不会反序列化失败。这提醒我们要在采集端做字段冗余、在计算端做 Schema 兼容处理不然上线后改需求就是事故。7. 常见问题与排查技巧实录最后这部分是硬核经验我按组件分类整理几个高频问题每个都是我在真实项目中处理过的。7.1 Kafka 常见问题Kafka 消息延迟高先看是不是生产端 batch 和 linger 设置过大再看消费端 poll 循环里处理耗时是否过长最后看分区数是否足够。我用kafka-consumer-groups.sh --describe --group group查看消费 lag如果 lag 持续增长先扩容消费者实例数。Kafka OOM多数是消费者拉取的消息太大导致堆溢出。调大fetch.max.bytes是治标根治要控制单条消息大小让生产端把大对象拆小或改为引用存储。Kafka 消息堆积如果消费者处理不过来优先看消费者逻辑里有没有慢操作调外部接口、大事务。我遇到过 Consumer 里同一个事务里查了三次数据库导致吞吐从每秒几千掉到几百。把事务拆小或者用异步批量写问题立刻缓解。7.2 Spark 和 Flink 常见问题Flink 的 JDBC 连接器异常绝大多数是没加目标数据库的驱动 jar 包或者连接池配置太小被打满。先确认ClassNotFoundException再看连接串和目标库网络最后看连接池最大连接数。Flink SQL 中 Watermark 不生效检查 source DDL 里是否声明了WATERMARK FOR ts AS ...而且时间字段类型必须是TIMESTAMP(3)。如果字段是 BIGINT 毫秒要先用TO_TIMESTAMP_LTZ(ts, 3)转换我见过的最多报错来源就是这个。Spark 任务 OOM先调大 executor 内存并配置spark.memory.offHeap.enabledtrue同时检查是不是数据倾斜——某个 key 数据量特别大。用groupBy(col).count().orderBy(desc)看一眼分布对倾斜 key 加随机前缀再做二次聚合基本都能解决。7.3 链路级排查思路当整套链路出问题时别急着翻组件日志。我的排查顺序是从数据源头到下游一段一段验证。第一步验证日志文件有没有新数据产生tail -f。第二步验证 Flume 有没有读到并发出日志看 Flume 日志里的 event 计数。第三步验证 Kafka topic 有没有数据console consumer 消费。第四步验证 Flink/Spark 任务有没有消费到数据看任务 Web UI 的 numRecordsIn。第五步验证结果表有没有更新。每一步输出有/没有就能把问题锁定在某一段。这个方法虽然土但效率比抱着 Flink 日志看半天高得多。我自己有两次线上事故是靠这个笨办法 10 分钟内定位的。8. 个人学习路线建议与避坑总结最后分享一些我这个过来人的体会。入门阶段不要试图一次把四个组件全部精通。先用两周时间把 Flume 和 Kafka 跑通能实现日志文件 → Kafka就算过关再用三周时间上手 Structured Streaming 或 Flink能消费 Kafka 数据做窗口统计同时把事件时间、Watermark 这两个概念搞明白剩下时间做一到两个综合案例把链路串起来。这个节奏比较适合上班族利用业余时间学习我是用了差不多两个月走完的。避坑方面最想提醒大家的是不要过早陷入源码阅读和深度调优。很多新手一上来就研究 Kafka 副本同步原理、Flink 状态后端源码结果基础 API 还没写熟心态直接崩了。先把链路跑通、把 API 用起来原理层面的东西在实战中遇到了再去补效率会高很多。另外环境搭建上我强烈建议用 Docker Desktop 起 Kafka 和 Flink 环境而不是自己在虚拟机里手装。手装能加深理解但极其耗时学习期先用 Docker 把时间花在核心概念上等做了两三个项目后再回过头来手装一遍 Kafka 集群那时你会对底层机制有全新的认识。这套实时计算技术栈的学习曲线确实不友好但只要跟着这条主线一步步走每天能跑通一个实验一个月后你会惊讶于自己已经能独立搭出一条实时数据管道了。数据从日志里产生流经 Kafka被 Flink 实时计算成结果再投递到大屏——这种亲手打通一条数据高速公路的成就感是死记硬背知识点给不了的。