Dart: 项目优化与发布 — DataPipeline 从开发到生产

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

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

import 'dart:isolate';

// Bad: Static chunking - uneven work
List<List``<T>``> staticChunk``<T>``(List``<T>`` data, int chunks) {
  final size = (data.length / chunks).ceil();
  return [
    for (var i = 0; i < chunks; i++)
      data.sublist(i * size, ((i + 1) * size).clamp(0, data.length))
  ];
}

// Good: Dynamic work stealing with SendPort
class DynamicIsolatePool {
  final int _poolSize;
  var _pendingChunks = <List<Map<String, dynamic>>>[];
  final _results = ``<AnalysisResult>``[];
  var _activeWorkers = 0;
  late Completer<List``<AnalysisResult>``> _completer;

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

  Future<List``<AnalysisResult>``> process(
    List<Map<String, dynamic>> allOrders, {
    int chunkSize = 50000,
  }) async {
    _pendingChunks = [
      for (var i = 0; i < allOrders.length; i += chunkSize)
        allOrders.sublist(i, (i + chunkSize).clamp(0, allOrders.length))
    ];

    _completer = Completer<List``<AnalysisResult>``>();
    _activeWorkers = 0;

    final workers = <_Worker>[];
    for (var i = 0; i < _poolSize && _pendingChunks.isNotEmpty; i++) {
      final chunk = _pendingChunks.removeAt(0);
      workers.add(await _spawnWorker(chunk));
      _activeWorkers++;
    }

    return _completer.future;
  }

  Future<_Worker> _spawnWorker(List<Map<String, dynamic>> initialChunk) async {
    final receivePort = ReceivePort();
    final isolate = await Isolate.spawn(_workerEntry, receivePort.sendPort);
    final sendPort = await receivePort.first as SendPort;

    final responsePort = ReceivePort();
    sendPort.send(_WorkMessage(data: initialChunk, responsePort: responsePort.sendPort));

    responsePort.listen((result) {
      _results.add(result as AnalysisResult);
      _activeWorkers--;

      if (_pendingChunks.isNotEmpty) {
        final nextChunk = _pendingChunks.removeAt(0);
        sendPort.send(_WorkMessage(data: nextChunk, responsePort: responsePort.sendPort));
        _activeWorkers++;
      } else if (_activeWorkers == 0) {
        isolate.kill(priority: Isolate.immediate);
        receivePort.close();
        responsePort.close();
        _completer.complete(_results);
      }
    });

    return _Worker(isolate: isolate, sendPort: sendPort);
  }

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

    receivePort.listen((message) {
      if (message is _WorkMessage) {
        final orders = message.data.map((m) => Order(
          id: m['id'] as String,
          amount: (m['amount'] as num).toDouble(),
          status: m['status'] as String,
          category: m['category'] as String,
        )).toList();

        final result = const OrderAnalyzer().analyze(orders);
        message.responsePort.send(result);
      }
    });
  }
}

class _Worker {
  final Isolate isolate;
  final SendPort sendPort;
  const _Worker({required this.isolate, required this.sendPort});
}

class _WorkMessage {
  final List<Map<String, dynamic>> data;
  final SendPort responsePort;
  const _WorkMessage({required this.data, required this.responsePort});
}

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

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

import 'dart:async';

class BackpressurePipeline {
  final DataSource _source;
  final OrderParser _parser;
  final int _maxConcurrent;

  BackpressurePipeline({
    required DataSource source,
    required OrderParser parser,
    int maxConcurrent = 1000,
  })  : _source = source,
        _parser = parser,
        _maxConcurrent = maxConcurrent;

  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    var pending = 0;
    final completer = Completer``<void>``();

    final subscription = _source.readLines().listen(
      (line) {
        try {
          orders.add(_parser.parse(line));
        } on FormatException {
          return;
        }

        pending++;
        if (pending >= _maxConcurrent) {
          subscription.pause();
        }
      },
      onDone: () {
        completer.complete();
      },
      onError: (e) {
        completer.completeError(e);
      },
    );

    await completer.future;

    return const OrderAnalyzer().analyze(orders);
  }

  // Pausable stream with buffer control
  Stream``<Order>`` streamWithBackpressure() async* {
    var buffer = ``<Order>``[];

    await for (final line in _source.readLines()) {
      try {
        buffer.add(_parser.parse(line));
      } on FormatException {
        continue;
      }

      if (buffer.length >= _maxConcurrent) {
        for (final order in buffer) {
          yield order;
        }
        buffer = [];
      }
    }

    for (final order in buffer) {
      yield order;
    }
  }
}

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

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

import 'dart:math';

class MemoryOptimizer {
  // Use iterables instead of lists where possible
  static Iterable``<Order>`` lazyParse(Iterable``<String>`` lines, OrderParser parser) sync* {
    for (final line in lines) {
      try {
        yield parser.parse(line);
      } on FormatException {
        continue;
      }
    }
  }

  // Chunk processing to limit memory
  static Future``<AnalysisResult>`` processInChunks(
    Stream``<String>`` lines,
    OrderParser parser, {
    int chunkSize = 100000,
  }) async {
    var chunk = ``<Order>``[];
    var totalResult = _EmptyAnalysisResult();

    await for (final line in lines) {
      try {
        chunk.add(parser.parse(line));
      } on FormatException {
        continue;
      }

      if (chunk.length >= chunkSize) {
        final partial = const OrderAnalyzer().analyze(chunk);
        totalResult = totalResult.merge(partial);
        chunk = [];
      }
    }

    if (chunk.isNotEmpty) {
      final partial = const OrderAnalyzer().analyze(chunk);
      totalResult = totalResult.merge(partial);
    }

    return totalResult.toAnalysisResult();
  }
}

class _EmptyAnalysisResult {
  int totalOrders = 0;
  int completedOrders = 0;
  double revenue = 0;
  double tax = 0;
  final Map<String, double> revenueByCategory = {};

  _EmptyAnalysisResult merge(AnalysisResult other) {
    totalOrders += other.totalOrders;
    completedOrders += other.completedOrders;
    revenue += other.revenue;
    tax += other.tax;
    for (final entry in other.revenueByCategory.entries) {
      revenueByCategory.update(entry.key, (v) => v + entry.value, ifAbsent: () => entry.value);
    }
    return this;
  }

  AnalysisResult toAnalysisResult() => AnalysisResult(
    totalOrders: totalOrders,
    completedOrders: completedOrders,
    revenue: revenue,
    tax: tax,
    total: revenue + tax,
    revenueByCategory: Map.unmodifiable(revenueByCategory),
  );
}

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

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

# Debug build (large, includes debug info)
dart compile exe bin/data_pipeline.dart -o build/data_pipeline_debug

# Release build (optimized, smaller)
dart compile exe bin/data_pipeline.dart -o build/data_pipeline \
  --define=MODE=release

# Check binary size
ls -lh build/

# Strip debug symbols (additional size reduction)
# On Linux/macOS:
strip build/data_pipeline

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

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

// Conditional imports for tree shaking
const _isRelease = bool.fromEnvironment('MODE.release');

void log(String message) {
  if (!_isRelease) {
    print('[DEBUG] $message');
  }
}

// This entire class will be tree-shaken in release mode
class DebugLogger {
  void log(String message) {
    if (!_isRelease) {
      print('[DEBUG] ${DateTime.now()}: $message');
    }
  }
}

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

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

# .github/workflows/ci.yml
name: DataPipeline CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  analyze-and-test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        sdk: [stable, beta]

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1
        with:
          sdk: ${{ matrix.sdk }}

      - name: Install dependencies
        run: dart pub get

      - name: Verify formatting
        run: dart format --output=none --set-exit-if-changed .

      - name: Analyze code
        run: dart analyze --fatal-infos

      - name: Run tests
        run: dart test --coverage=coverage

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: coverage/lcov.info

  build-and-release:
    needs: analyze-and-test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Install dependencies
        run: dart pub get

      - name: Compile AOT binary
        run: |
          dart compile exe bin/data_pipeline.dart -o build/data_pipeline

      - name: Run integration test
        run: |
          ./build/data_pipeline analyze --source csv --input test/fixtures/sample.csv

      - name: Upload binary artifact
        uses: actions/upload-artifact@v4
        with:
          name: data_pipeline-linux
          path: build/data_pipeline

  publish:
    needs: build-and-release
    runs-on: ubuntu-latest
    if: startsWith(github.ref, 'refs/tags/v')

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Install dependencies
        run: dart pub get

      - name: Publish to pub.dev
        run: dart pub publish --force
        env:
          PUB_CREDENTIALS: ${{ secrets.PUB_CREDENTIALS }}

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

name: data_pipeline
description: E-commerce data analytics CLI tool - process million-level orders with Stream + Isolate
version: 1.0.0

homepage: https://github.com/bob/datapipeline
repository: https://github.com/bob/datapipeline
issue_tracker: https://github.com/bob/datapipeline/issues

environment:
  sdk: ^3.0.0

dependencies:
  args: ^2.4.2
  http: ^1.1.0
  csv: ^6.0.0

dev_dependencies:
  test: ^1.24.0
  build_runner: ^2.4.0
  json_serializable: ^6.7.0

executables:
  data_pipeline: data_pipeline

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

# CHANGELOG


---

## 7. Release Notes

### (1) 1.0.0 - 2025-01-15

#### Features
- CSV and JSON data source support
- Stream-based pipeline with backpressure
- Isolate pool for parallel processing
- CLI with analyze/export/validate subcommands
- Pattern Matching + Sealed class architecture

### (2) Performance
- 1M orders processed in ~8 seconds (4-core)
- Memory usage under 200MB for 1M records

### (3) Testing
- 95% code coverage
- Unit + integration + E2E tests

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

# Pre-publish checklist

# 1. Dry run publish
dart pub publish --dry-run

# 2. Check score on pub.dev
dart pub publish --dry-run 2>&1 | grep -E "score|warning|error"

# 3. Generate API docs
dart doc .

# 4. Verify package contents
dart pub pack  # creates .tar.gz for inspection

# 5. Final publish
dart pub publish

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

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

/// Analyzes e-commerce order data and produces statistical results.
///
/// Use [OrderAnalyzer] as the primary entry point for data analysis.
/// Configure the [taxRate] based on the target region.
///
/// Example:
/// ```dart
/// final analyzer = OrderAnalyzer(taxRate: 0.08);
/// final result = analyzer.analyze(orders);
/// print('Revenue: \$${result.revenue} USD');
/// ```
///
/// See also:
/// - [AnalysisResult] for the output structure
/// - [Order] for the input model
class OrderAnalyzer {
  /// Creates an analyzer with the given [taxRate].
  ///
  /// Defaults to 0.08 (8% US tax rate).
  /// Use region-specific rates: US=0.08, EU=0.20, UK=0.15.
  const OrderAnalyzer({this.taxRate = 0.08});

  /// The tax rate applied to completed orders.
  final double taxRate;

  /// Analyzes a list of [orders] and returns aggregated statistics.
  ///
  /// Only orders with status 'completed' are included in revenue
  /// calculations. Orders with other statuses are counted in
  /// [AnalysisResult.totalOrders] but excluded from revenue.
  AnalysisResult analyze(List``<Order>`` orders) {
    // ... implementation
  }
}

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

// ============================================
// DataPipeline Performance Benchmark
// Alice's End-to-End Stress Test
// ============================================

import 'dart:async';
import 'dart:io';

class Benchmark {
  final String name;
  final Stopwatch _stopwatch;

  Benchmark(this.name) : _stopwatch = Stopwatch();

  Future``<T>`` run``<T>``(Future``<T>`` Function() action) async {
    _stopwatch.reset();
    _stopwatch.start();
    final result = await action();
    _stopwatch.stop();
    print('$name: ${_stopwatch.elapsedMilliseconds}ms');
    return result;
  }
}

// Generate test data
List``<String>`` generateOrders(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(7, '0')},$amount,$status,$category,$region';
  });
}

void main() async {
  print('=== DataPipeline Performance Benchmark ===\n');

  final sizes = [10000, 100000, 1000000];

  for (final size in sizes) {
    final label = size >= 1000000 ? '${size ~/ 1000000}M' : '${size ~/ 1000}K';
    print('--- $label orders ---');

    // Generate and save to temp file
    final data = generateOrders(size);
    final tempFile = File('${Directory.systemTemp.path}/bench_${label}.csv');
    await tempFile.writeAsString(data.join('\n'));

    // Benchmark 1: Stream processing
    final streamResult = await Benchmark('Stream ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      final parser = const CsvOrderParser();
      final pipeline = Pipeline(source: source, parser: parser);
      return pipeline.execute();
    });

    // Benchmark 2: Memory-efficient chunked processing
    final chunkedResult = await Benchmark('Chunked ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      return MemoryOptimizer.processInChunks(
        source.readLines(),
        const CsvOrderParser(),
        chunkSize: 50000,
      );
    });

    // Verify results match
    final streamRev = streamResult.revenue.toStringAsFixed(2);
    final chunkedRev = chunkedResult.revenue.toStringAsFixed(2);
    print('Revenue match: ${streamRev == chunkedRev ? "PASS" : "FAIL"}');
    print('Revenue: \$${streamRev} USD\n');

    await tempFile.delete();
  }

  // Benchmark 3: AOT binary size check
  print('--- Build Artifacts ---');
  final exe = File('build/data_pipeline');
  if (await exe.exists()) {
    final sizeKB = await exe.length() ~/ 1024;
    print('AOT binary: ${sizeKB}KB');
  } else {
    print('AOT binary: not built (run: dart compile exe bin/data_pipeline.dart)');
  }

  print('\n=== Benchmark Complete ===');
  print('Alice验收标准: 1M orders < 10s, memory < 300MB, binary < 20MB');
}

// ---- Required stubs (imported from previous lessons) ----

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());
}

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');
    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 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 * 0.08;
}

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});
}

class OrderAnalyzer {
  const OrderAnalyzer();
  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 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: revenue + tax, revenueByCategory: byCategory);
  }
}

class Pipeline {
  final DataSource source;
  final OrderParser parser;
  const Pipeline({required this.source, required this.parser});
  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    await for (final line in source.readLines()) {
      try { orders.add(parser.parse(line)); } on FormatException { continue; }
    }
    return const OrderAnalyzer().analyze(orders);
  }
}

class MemoryOptimizer {
  static Future``<AnalysisResult>`` processInChunks(Stream``<String>`` lines, OrderParser parser, {int chunkSize = 50000}) async {
    var chunk = ``<Order>``[];
    int totalOrders = 0, completedOrders = 0;
    double revenue = 0, tax = 0;
    final byCategory = <String, double>{};

    await for (final line in lines) {
      try { chunk.add(parser.parse(line)); } on FormatException { continue; }
      if (chunk.length >= chunkSize) {
        final r = const OrderAnalyzer().analyze(chunk);
        totalOrders += r.totalOrders;
        completedOrders += r.completedOrders;
        revenue += r.revenue;
        tax += r.tax;
        for (final e in r.revenueByCategory.entries) {
          byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
        }
        chunk = [];
      }
    }
    if (chunk.isNotEmpty) {
      final r = const OrderAnalyzer().analyze(chunk);
      totalOrders += r.totalOrders;
      completedOrders += r.completedOrders;
      revenue += r.revenue;
      tax += r.tax;
      for (final e in r.revenueByCategory.entries) {
        byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
      }
    }
    return AnalysisResult(totalOrders: totalOrders, completedOrders: completedOrders, revenue: revenue, tax: tax, total: revenue + tax, revenueByCategory: byCategory);
  }
}

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