English 中文 العربية 日本語 Português

Dart: 项目设计与开发 — DataPipeline CLI 工具实战

真正的工程师不是写代码，而是设计能让代码生长的架构。 — Bob

1. 你将学到

2. 一个开发者的真实故事

(1) 痛点：脚本演化到工具的阵痛

Bob 最初写了几个独立 Dart 脚本处理电商数据：parse_csv.dart、calc_stats.dart、gen_report.dart。随着需求增长，脚本间复制粘贴越来越多，改一个字段名要改 5 个文件。Alice 问"能否加个 JSON 数据源？"，Bob 发现所有解析逻辑和 CSV 耦合，改不动。

(2) 重新设计的解法

Bob 决定从头设计 DataPipeline，采用分层架构、Sealed class 抽象数据源、Stream 管道异步处理。Charlie 做代码评审，确保质量。

100% 
graph TD
  subgraph 架构分层
    A[CLI Layer<br/>args 解析]
    B[Service Layer<br/>Pipeline 调度]
    C[Data Layer<br/>Source + Parser]
    D[Core Layer<br/>Models + Utils]
  end
  A --> B
  B --> C
  C --> D
  B --> E[Isolate Pool]
  B --> F[Stream Pipeline]
  C --> G["Sealed DataSource"]
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

(3) 收益

3. 项目架构设计

(1) 分层架构

层级 目录 职责 依赖
CLI Layer bin/ 参数解析、命令路由 Service Layer
Service Layer lib/src/services/ Pipeline 调度、Isolate 管理 Data + Core Layer
Data Layer lib/src/data/ 数据源抽象、解析器 Core Layer
Core Layer lib/src/core/ 模型、工具、常量 无外部依赖

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：项目目录结构

TEXT 📖 仅展示 
data_pipeline/
  bin/
    data_pipeline.dart        # CLI entry point
  lib/
    src/
      core/
        models/
          order.dart
          product.dart
          customer.dart
          analysis_result.dart
        utils/
          formatters.dart
          validators.dart
          constants.dart
      data/
        sources/
          data_source.dart     # Sealed class
          csv_source.dart
          json_source.dart
          api_source.dart
        parsers/
          order_parser.dart
          product_parser.dart
      services/
        pipeline.dart
        analyzer.dart
        isolate_pool.dart
        report_generator.dart
    data_pipeline.dart         # Barrel export
  test/
    core/
      models_test.dart
      utils_test.dart
    data/
      parsers_test.dart
    services/
      pipeline_test.dart
      analyzer_test.dart
  pubspec.yaml
```text


```text
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

4. CLI 参数解析与子命令

(1) args 包子命令设计

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：CLI 入口与子命令

DART 
import 'package:args/args.dart';
import 'package:args/command_runner.dart';

// Sub-command: analyze
class AnalyzeCommand extends Command {
  @override
  final name = 'analyze';
  @override
  final description = 'Analyze order data and generate statistics';

  AnalyzeCommand() {
    argParser
      ..addOption('source', abbr: 's', defaultsTo: 'csv', allowed: ['csv', 'json', 'api'])
      ..addOption('input', abbr: 'i', mandatory: true)
      ..addOption('output', abbr: 'o', defaultsTo: 'stdout')
      ..addFlag('parallel', abbr: 'p', defaultsTo: false)
      ..addOption('isolate-count', defaultsTo: '4');
  }

  @override
  Future``<void>`` run() async {
    final source = argResults!['source'] as String;
    final input = argResults!['input'] as String;
    final output = argResults!['output'] as String;
    final parallel = argResults!['parallel'] as bool;
    final isolateCount = int.parse(argResults!['isolate-count'] as String);

    print('Source: $source | Input: $input | Parallel: $parallel');
  }
}

// Sub-command: export
class ExportCommand extends Command {
  @override
  final name = 'export';
  @override
  final description = 'Export analysis results to file';

  ExportCommand() {
    argParser
      ..addOption('format', defaultsTo: 'json', allowed: ['json', 'csv', 'markdown'])
      ..addOption('input', abbr: 'i', mandatory: true)
      ..addOption('output', abbr: 'o', mandatory: true);
  }

  @override
  Future``<void>`` run() async {
    final format = argResults!['format'] as String;
    final input = argResults!['input'] as String;
    final output = argResults!['output'] as String;

    print('Export: $format | $input -> $output');
  }
}

void main(List``<String>`` args) async {
  final runner = CommandRunner('data_pipeline', 'DataPipeline - E-commerce data analytics CLI')
    ..addCommand(AnalyzeCommand())
    ..addCommand(ExportCommand());

  try {
    await runner.run(args);
  } on UsageException catch (e) {
    print(e);
  }
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。
子命令 功能 关键参数
analyze 数据分析与统计 --source, --input, --parallel
export 导出报告 --format, --input, --output
validate 数据校验 --input, --strict

5. 数据源抽象 — Sealed Class

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Sealed DataSource

DART 
sealed class DataSource {
  const DataSource();

  String get displayName;

  Stream``<String>`` readLines();
}

class CsvSource extends DataSource {
  final String path;
  final String delimiter;

  const CsvSource({required this.path, this.delimiter = ','});

  @override
  String get displayName => 'CSV: $path';

  @override
  Stream``<String>`` readLines() => File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

class JsonSource extends DataSource {
  final String path;

  const JsonSource({required this.path});

  @override
  String get displayName => 'JSON: $path';

  @override
  Stream``<String>`` readLines() async* {
    final content = await File(path).readAsString();
    final jsonList = jsonDecode(content) as List;
    for (final item in jsonList) {
      yield jsonEncode(item);
    }
  }
}

class ApiSource extends DataSource {
  final String endpoint;
  final Map<String, String> headers;

  const ApiSource({required this.endpoint, this.headers = const {}});

  @override
  String get displayName => 'API: $endpoint';

  @override
  Stream``<String>`` readLines() async* {
    final client = Client();
    final response = await client.get(Uri.parse(endpoint), headers: headers);
    final jsonList = jsonDecode(response.body) as List;
    for (final item in jsonList) {
      yield jsonEncode(item);
    }
    client.close();
  }
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Pattern Matching 消费 DataSource

DART 
DataSource createSource(String type, String input) => switch (type) {
  'csv' => CsvSource(path: input),
  'json' => JsonSource(path: input),
  'api' => ApiSource(endpoint: input),
  _ => throw ArgumentError('Unknown source type: $type'),
};

String sourceIcon(DataSource source) => switch (source) {
  CsvSource() => '📄',
  JsonSource() => '📋',
  ApiSource() => '🌐',
};
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。
模式 用途 优势
Sealed class 数据源类型 exhaustive switch 编译保证
Pattern Matching 消费数据源 无需 if-else 链
工厂函数 创建数据源 统一入口

6. 多格式解析器

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Order 解析器

DART 
abstract class OrderParser {
  const OrderParser();

  Order parse(String raw);
  List``<Order>`` parseBatch(List``<String>`` raws) => raws.map(parse).toList();
}

class CsvOrderParser extends OrderParser {
  final String delimiter;

  const CsvOrderParser({this.delimiter = ','});

  @override
  Order parse(String raw) {
    final parts = raw.split(delimiter);
    if (parts.length < 4) {
      throw FormatException('Invalid CSV row: $raw');
    }
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
      region: parts.length > 4 ? parts[4].trim() : 'US',
    );
  }
}

class JsonOrderParser extends OrderParser {
  const JsonOrderParser();

  @override
  Order parse(String raw) {
    final json = jsonDecode(raw) as Map<String, dynamic>;
    return Order(
      id: json['id'] as String,
      amount: (json['amount'] as num).toDouble(),
      status: json['status'] as String,
      category: json['category'] as String,
      region: (json['region'] as String?) ?? 'US',
    );
  }
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

7. Stream 异步管道

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Pipeline 流处理

DART 
import 'dart:async';
import 'dart:convert';
import 'dart:io';

class Pipeline {
  final DataSource _source;
  final OrderParser _parser;
  final int _chunkSize;

  Pipeline({
    required DataSource source,
    required OrderParser parser,
    int chunkSize = 1000,
  })  : _source = source,
        _parser = parser,
        _chunkSize = chunkSize;

  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    var processed = 0;
    var skipped = 0;

    await for (final line in _source.readLines()) {
      try {
        final order = _parser.parse(line);
        orders.add(order);
        processed++;
      } on FormatException {
        skipped++;
      }

      if (processed % _chunkSize == 0) {
        stdout.writeln('Progress: $processed orders processed, $skipped skipped');
      }
    }

    stdout.writeln('Total: $processed processed, $skipped skipped');

    final analyzer = OrderAnalyzer();
    return analyzer.analyze(orders);
  }

  Stream``<Order>`` streamOrders() async* {
    await for (final line in _source.readLines()) {
      try {
        yield _parser.parse(line);
      } on FormatException {
        continue;
      }
    }
  }

  Stream``<AnalysisResult>`` streamByCategory() {
    final controller = StreamController``<AnalysisResult>``();

    streamOrders().fold<Map<String, List``<Order>``>>(
      {},
      (groups, order) {
        groups.update(order.category, (v) => v..add(order), ifAbsent: () => [order]);
        return groups;
      },
    ).then((groups) {
      final analyzer = OrderAnalyzer();
      for (final entry in groups.entries) {
        controller.add(analyzer.analyze(entry.value));
      }
      controller.close();
    });

    return controller.stream;
  }
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

8. Isolate 并行处理

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Isolate Pool

DART 
import 'dart:isolate';

class IsolatePool {
  final int _poolSize;
  final List``<Isolate>`` _isolates = [];
  final List``<SendPort>`` _sendPorts = [];

  IsolatePool({int poolSize = 4}) : _poolSize = poolSize;

  Future``<void>`` initialize() async {
    for (var i = 0; i < _poolSize; i++) {
      final receivePort = ReceivePort();
      final isolate = await Isolate.spawn(
        _isolateEntryPoint,
        receivePort.sendPort,
      );
      final sendPort = await receivePort.first as SendPort;
      _isolates.add(isolate);
      _sendPorts.add(sendPort);
    }
  }

  static void _isolateEntryPoint(SendPort mainSendPort) {
    final receivePort = ReceivePort();
    mainSendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message is _IsolateTask) {
        final result = _processChunk(message.orders, message.taxRate);
        message.responsePort.send(result);
      }
    });
  }

  static AnalysisResult _processChunk(List<Map<String, dynamic>> rawOrders, double taxRate) {
    final orders = rawOrders.map((o) => Order(
      id: o['id'] as String,
      amount: (o['amount'] as num).toDouble(),
      status: o['status'] as String,
      category: o['category'] as String,
      region: (o['region'] as String?) ?? 'US',
    )).toList();

    final analyzer = OrderAnalyzer(taxRate: taxRate);
    return analyzer.analyze(orders);
  }

  Future<List``<AnalysisResult>``> processInParallel(
    List<Map<String, dynamic>> allOrders,
    double taxRate,
  ) async {
    final chunkSize = (allOrders.length / _poolSize).ceil();
    final results = ``<AnalysisResult>``[];
    final completers = <Completer``<AnalysisResult>``>[];

    for (var i = 0; i < _poolSize; i++) {
      final start = i * chunkSize;
      final end = (start + chunkSize).clamp(0, allOrders.length);
      if (start >= allOrders.length) break;

      final chunk = allOrders.sublist(start, end);
      final completer = Completer``<AnalysisResult>``();
      completers.add(completer);

      _sendPorts[i].send(_IsolateTask(
        orders: chunk,
        taxRate: taxRate,
        responsePort: completer.future as dynamic,
      ));
    }

    for (final completer in completers) {
      results.add(await completer.future);
    }

    return results;
  }

  void dispose() {
    for (final isolate in _isolates) {
      isolate.kill(priority: Isolate.immediate);
    }
  }
}

class _IsolateTask {
  final List<Map<String, dynamic>> orders;
  final double taxRate;
  final SendPort responsePort;

  const _IsolateTask({
    required this.orders,
    required this.taxRate,
    required this.responsePort,
  });
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。
并行模式 适用场景 优势
Isolate.run 一次性计算 简单 API
Isolate.spawn 长驻 worker 可复用
Isolate Pool 均匀分片 充分利用多核
compute (Flutter) UI 不卡顿 Flutter 专用

9. 依赖注入

▶ 示例

TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Service Locator 模式

DART 
class ServiceLocator {
  static final _instances = <Type, dynamic>{};

  static void register``<T>``(T instance) {
    _instances[T] = instance;
  }

  static T get``<T>``() {
    final instance = _instances[T];
    if (instance == null) {
      throw StateError('Service not registered: $T');
    }
    return instance as T;
  }

  static void reset() {
    _instances.clear();
  }
}

// Registration at startup
void setupServices({required String sourceType, required String inputPath}) {
  final source = createSource(sourceType, inputPath);

  ServiceLocator.register``<DataSource>``(source);

  final parser = switch (source) {
    CsvSource() => const CsvOrderParser() as OrderParser,
    JsonSource() => const JsonOrderParser(),
    ApiSource() => const JsonOrderParser(),
  };

  ServiceLocator.register``<OrderParser>``(parser);
  ServiceLocator.register``<OrderAnalyzer>``(const OrderAnalyzer());
  ServiceLocator.register``<Pipeline>``(Pipeline(source: source, parser: parser));
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

10. 完整示例：DataPipeline CLI 工具

DART 
// ============================================
// DataPipeline CLI Tool - Complete Implementation
// Bob's e-commerce data analytics tool
// ============================================

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';

// ---- Core Models ----

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;
  final String region;

  const Order({
    required this.id,
    required this.amount,
    required this.status,
    required this.category,
    this.region = 'US',
  });

  double get tax => amount * _taxRate(region);
  double get total => amount + tax;

  static double _taxRate(String region) => switch (region) {
    'US' => 0.08,
    'EU' => 0.20,
    'UK' => 0.15,
    'JP' => 0.10,
    _ => 0.10,
  };

  Map<String, dynamic> toJson() => {
    'id': id,
    'amount': amount,
    'status': status,
    'category': category,
    'region': region,
  };
}

class AnalysisResult {
  final int totalOrders;
  final int completedOrders;
  final double revenue;
  final double tax;
  final double total;
  final Map<String, double> revenueByCategory;

  const AnalysisResult({
    required this.totalOrders,
    required this.completedOrders,
    required this.revenue,
    required this.tax,
    required this.total,
    required this.revenueByCategory,
  });

  double get averageOrderValue => completedOrders > 0 ? revenue / completedOrders : 0;

  @override
  String toString() => '''
=== DataPipeline Analytics ===
Orders:    $completedOrders/$totalOrders completed
Revenue:   \$${revenue.toStringAsFixed(2)} USD
Tax:       \$${tax.toStringAsFixed(2)} USD
Total:     \$${total.toStringAsFixed(2)} USD
Average:   \$${averageOrderValue.toStringAsFixed(2)} USD
Categories: ${revenueByCategory.keys.join(', ')}
''';
}

// ---- Core Services ----

class OrderAnalyzer {
  final double taxRate;

  const OrderAnalyzer({this.taxRate = 0.08});

  AnalysisResult analyze(List``<Order>`` orders) {
    final completed = orders.where((o) => o.status == 'completed').toList();
    final revenue = completed.fold``<double>``(0, (s, o) => s + o.amount);
    final tax = completed.fold``<double>``(0, (s, o) => s + o.tax);
    final total = revenue + tax;

    final byCategory = <String, double>{};
    for (final o in completed) {
      byCategory.update(o.category, (v) => v + o.amount, ifAbsent: () => o.amount);
    }

    return AnalysisResult(
      totalOrders: orders.length,
      completedOrders: completed.length,
      revenue: revenue,
      tax: tax,
      total: total,
      revenueByCategory: byCategory,
    );
  }
}

// ---- Data Source (Sealed) ----

sealed class DataSource {
  const DataSource();
  String get displayName;
  Stream``<String>`` readLines();
}

class CsvSource extends DataSource {
  final String path;
  const CsvSource({required this.path});

  @override
  String get displayName => 'CSV: $path';

  @override
  Stream``<String>`` readLines() =>
    File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

class JsonSource extends DataSource {
  final String path;
  const JsonSource({required this.path});

  @override
  String get displayName => 'JSON: $path';

  @override
  Stream``<String>`` readLines() async* {
    final content = await File(path).readAsString();
    final list = jsonDecode(content) as List;
    for (final item in list) {
      yield jsonEncode(item);
    }
  }
}

// ---- Parser ----

abstract class OrderParser {
  const OrderParser();
  Order parse(String raw);
}

class CsvOrderParser extends OrderParser {
  const CsvOrderParser();

  @override
  Order parse(String raw) {
    final parts = raw.split(',');
    if (parts.length < 4) throw FormatException('Invalid row: $raw');
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
    );
  }
}

class JsonOrderParser extends OrderParser {
  const JsonOrderParser();

  @override
  Order parse(String raw) {
    final json = jsonDecode(raw) as Map<String, dynamic>;
    return Order(
      id: json['id'] as String,
      amount: (json['amount'] as num).toDouble(),
      status: json['status'] as String,
      category: json['category'] as String,
    );
  }
}

// ---- Pipeline ----

class Pipeline {
  final DataSource source;
  final OrderParser parser;

  const Pipeline({required this.source, required this.parser});

  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    var count = 0;

    await for (final line in source.readLines()) {
      try {
        orders.add(parser.parse(line));
        count++;
        if (count % 100000 == 0) {
          stdout.writeln('Progress: ${(count / 1000).toStringAsFixed(0)}K orders');
        }
      } on FormatException {
        continue;
      }
    }

    stdout.writeln('Loaded: ${(count / 1000).toStringAsFixed(0)}K orders from ${source.displayName}');
    return const OrderAnalyzer().analyze(orders);
  }
}

// ---- Demo Data Generator ----

List``<String>`` generateDemoOrders(int count) {
  final categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final regions = ['US', 'EU', 'UK', 'JP'];

  return List.generate(count, (i) {
    final category = categories[i % categories.length];
    final status = statuses[i % statuses.length];
    final region = regions[i % regions.length];
    final amount = (100 + (i * 37) % 5000).toDouble();
    return 'ORD-${(i + 1).toString().padLeft(5, '0')},$amount,$status,$category,$region';
  });
}

// ---- Main ----

void main() async {
  print('=== DataPipeline CLI v1.0 ===\n');

  final demoData = generateDemoOrders(1000);
  final tempFile = File('${Directory.systemTemp.path}/demo_orders.csv');
  await tempFile.writeAsString(demoData.join('\n'));

  final source = CsvSource(path: tempFile.path);
  final parser = const CsvOrderParser();
  final pipeline = Pipeline(source: source, parser: parser);

  print('Source: ${source.displayName}');
  print('Processing...\n');

  final result = await pipeline.execute();
  print(result);

  final sorted = result.revenueByCategory.entries.toList()
    ..sort((a, b) => b.value.compareTo(a.value));

  print('Revenue by Category:');
  for (final entry in sorted) {
    print('  ${entry.key.padRight(12)}: \$${entry.value.toStringAsFixed(2)} USD');
  }

  await tempFile.delete();
  print('\nDone. Temporary file cleaned up.');
}
TEXT 📖 仅展示 
> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

11. Charlie 的代码评审

(1) ▶ 评审要点

维度 评审标准 DataPipeline 检查项
类型安全 无 dynamic，exhaustive switch Sealed DataSource + Pattern Matching
可测试性 核心逻辑不依赖 I/O OrderParser 纯函数，Pipeline 接受接口
错误处理 不吞异常，有降级 FormatException skip + 计数
性能 内存可控，异步不阻塞 Stream 逐行 + Isolate 分片
可扩展性 新功能不改旧代码 新 DataSource 子类零修改

Charlie: "Sealed class + Pattern Matching 让编译器帮我们检查遗漏，比 if-else 安全得多。OrderParser 是纯函数，测试无需 mock 文件系统。"

❓ 常见问题

Q：为什么要用 Sealed class 而不是枚举来表示数据源？ A：枚举不能携带数据（如 path、endpoint），Sealed class 可以有字段和方法，同时保持 exhaustive switch 的编译保证。

Q：Pipeline 中 Stream 和 Future 该选哪个？ A：数据逐条到达且需实时处理用 Stream；一次性加载完再处理用 Future。百万级数据推荐 Stream，避免内存溢出。

Q：Isolate 通信只能传基本类型吗？ A：SendPort 传递的数据必须可序列化。自定义类需转成 Map 或用 jsonEncode/Decode。Dart 3 的 Records 也支持传递。

Q：args 包和 dcli 包哪个好？ A：args 是官方包，适合标准子命令模式；dcli 提供更丰富的 CLI 工具（文件操作、进程管理等），但非官方。本教程选用 args。

Q：依赖注入一定要用框架吗？ A：不一定。小项目用 Service Locator 或构造函数注入即可。flutter_bloc、get_it 等框架适合大型项目。DataPipeline 用简单的 ServiceLocator。

Q：如何测试 Pipeline 的端到端流程？ A：用临时文件（Directory.systemTemp）作为测试数据源，验证输出 AnalysisResult 的各字段值。见 L20 测试课的集成测试示例。

📖 小节

📝 作业

  1. 基础题（难度⭐）：为 DataPipeline 添加 validate 子命令，接收 --input 参数，统计 CSV 文件中格式错误行数并报告。
  2. 进阶题（难度⭐⭐）：实现 JsonSource + JsonOrderParser，用一段 JSON 数组作为输入，运行 Pipeline 输出分析结果。验证 Sealed class 的 exhaustive switch 是否覆盖所有子类。
  3. 挑战题（难度⭐⭐⭐）：实现 IsolatePool，将 1000 条 demo 数据分成 4 片，用 4 个 Isolate 并行分析，最后合并 4 个 AnalysisResult（收入和订单数分别求和）。测量并行 vs 串行的时间差异。

← 上一课 | 下一课 →

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏