Python 并行批处理：限制待执行任务，并汇总每个文件的失败

python3 --version
mkdir parallel-batch-demo
cd parallel-batch-demo
printf 'alpha\nbeta\n' > a.txt
printf 'gamma' > 'b space.txt'
printf 'a.txt\nb space.txt\n' > good.txt

"""Count UTF-8 text lines with bounded parallel submissions. Python 3.9+."""
import argparse
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
import json
from pathlib import Path
import sys


def worker_count(value):
    number = int(value)
    if not 1 <= number <= 16:
        raise argparse.ArgumentTypeError("workers must be 1..16")
    return number


def count_lines(path):
    with path.open(encoding="utf-8", errors="strict") as source:
        return sum(1 for _ in source)


def collect(pending):
    done, _ = wait(pending, return_when=FIRST_COMPLETED)
    failed = 0
    for future in done:
        path = pending.pop(future)
        record = {"type": "file", "path": str(path)}
        try:
            record.update(ok=True, lines=future.result())
        except (OSError, UnicodeError, ValueError) as exc:
            failed += 1
            record.update(ok=False, error=f"{type(exc).__name__}: {exc}")
        print(json.dumps(record), flush=True)
    return len(done), failed


def run(manifest, workers):
    manifest = manifest.absolute()
    pending = {}
    processed = failed = peak_pending = 0
    with manifest.open(encoding="utf-8", errors="strict") as source:
        with ThreadPoolExecutor(max_workers=workers) as pool:
            for line in source:
                name = line.rstrip("\r\n")
                if not name:
                    continue
                path = manifest.parent / name
                pending[pool.submit(count_lines, path)] = path
                peak_pending = max(peak_pending, len(pending))
                if len(pending) >= 2 * workers:
                    count, errors = collect(pending)
                    processed += count
                    failed += errors
            while pending:
                count, errors = collect(pending)
                processed += count
                failed += errors
    print(json.dumps({
        "type": "summary", "processed": processed, "failed": failed,
        "workers": workers, "peak_pending": peak_pending,
    }), flush=True)
    return 1 if failed else 0


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("manifest", type=Path, help="UTF-8 path list, one per line")
    parser.add_argument("--workers", type=worker_count, default=4)
    args = parser.parse_args()
    try:
        return run(args.manifest, args.workers)
    except (OSError, UnicodeError, ValueError) as exc:
        print(f"FATAL: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

python3 count_files.py --help
python3 count_files.py good.txt --workers 2 > good-report.jsonl
cat good-report.jsonl

{"type": "summary", "processed": 2, "failed": 0, "workers": 2, "peak_pending": 2}

python3 - <<'PY_CHECK'
import json
from pathlib import Path
rows = [json.loads(line) for line in
        Path("good-report.jsonl").read_text(encoding="utf-8").splitlines()]
files = rows[:-1]
assert {Path(row["path"]).name: row["lines"] for row in files} == {
    "a.txt": 2, "b space.txt": 1,
}
assert all(row["ok"] for row in files)
assert rows[-1] == {
    "type": "summary", "processed": 2, "failed": 0,
    "workers": 2, "peak_pending": 2,
}
print("PASS: both files counted; pending bound respected")
PY_CHECK

python3 - <<'PY_DATA'
from pathlib import Path
Path("bad.bin").write_bytes(b"\xff\n")
Path("mixed.txt").write_text(
    "a.txt\nmissing.txt\nbad.bin\nb space.txt\n", encoding="utf-8"
)
PY_DATA
if python3 count_files.py mixed.txt --workers 2 > mixed-report.jsonl; then
  printf 'UNEXPECTED: all inputs accepted\n'
else
  batch_status=$?
  printf 'exit=%s\n' "$batch_status"
fi
python3 - <<'PY_CHECK'
import json
from pathlib import Path
rows = [json.loads(line) for line in
        Path("mixed-report.jsonl").read_text(encoding="utf-8").splitlines()]
assert rows[-1]["processed"] == 4
assert rows[-1]["failed"] == 2
assert rows[-1]["peak_pending"] <= 4
bad = {Path(row["path"]).name: row["error"]
       for row in rows[:-1] if not row["ok"]}
assert bad["missing.txt"].startswith("FileNotFoundError:")
assert bad["bad.bin"].startswith("UnicodeDecodeError:")
print("PASS: two failures recorded; healthy files still completed")
PY_CHECK

if python3 count_files.py good.txt --workers 2 > next-report.jsonl; then
  python3 -c 'print("NEXT: every listed file passed")'
else
  printf 'STOP: inspect the report and stderr\n' >&2
fi
python3 count_files.py good.txt --workers 0
printf 'exit=%s\n' "$?"