シングルノードマルチスレッドでサクッとパイプライン処理できたらいいなと思ったけど，普通にアクターモデルとかに乗っかった方がいい気がするのでボツ

pipeline.py

import queue
import threading
from typing import Callable, Generator, Any

_TERM = threading.Event()


class Worker(threading.Thread):
    def __init__(self, source: queue.Queue, destination: queue.Queue, batch_size: int,
                 callback: Callable[[Generator], Any]):
        self._callback = callback
        self._batch_size = batch_size
        self._source = source
        self._destination = destination
        super().__init__()

    def _batching(self):
        for _ in range(self._batch_size):
            yield self._source.get()

    def run(self):
        while not _TERM.is_set():
            self._destination.put(self._callback(self._batching()))


class Step:
    def __init__(self, source: queue.Queue, batch_size: int, callback: Callable[[Generator], Any]):
        self._source = source
        self._destination = queue.Queue()
        self._batch_size = batch_size
        self._callback = callback
        self._workers = [self._get_worker() for _ in range(4)]
        for w in self._workers:
            w.start()

    @property
    def destination(self):
        return self._destination

    def _get_worker(self) -> Worker:
        return Worker(self._source, self._destination, self._batch_size, self._callback)

    def put(self, row: Any) -> None:
        self._source.put(row)

    def pipe(self, batch_size: int, callback: Callable):
        return Step(self.destination, batch_size, callback)


def to_list(gen):
    l = []
    for row in gen:
        l.append(row)

    return l


def sum_each(gen):
    l = []
    for row in gen:
        l.append(sum(row))

    return l


source = queue.Queue()

for i in range(1000):
    source.put(i)

result = Step(source, 10, to_list).pipe(20, sum_each).destination