diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index 58e37ce5e..012478406 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -117,7 +117,7 @@ class Scraper:
         ]:
             self._check_deprecated_itemproc_method(method)
 
-        self.concurrent_items: int = crawler.settings.getint("CONCURRENT_ITEMS")
+        self.concurrent_items: int = max(1, crawler.settings.getint("CONCURRENT_ITEMS"))
         self.crawler: Crawler = crawler
         self.signals: SignalManager = crawler.signals
         assert crawler.logformatter
diff --git a/scrapy/utils/asyncio.py b/scrapy/utils/asyncio.py
index 7c7697f56..7ff0bf44d 100644
--- a/scrapy/utils/asyncio.py
+++ b/scrapy/utils/asyncio.py
@@ -102,32 +102,30 @@ async def _parallel_asyncio(
     """Execute a callable over the objects in the given iterable, in parallel,
     using no more than ``count`` concurrent calls.
 
+    Tasks are created on demand, one per item, so that a *count* much larger
+    than the amount of work does not cost anything.
+
     This function is only used in
     :meth:`scrapy.core.scraper.Scraper.handle_spider_output_async` and so it
     assumes that neither *callable* nor iterating *iterable* will raise an
     exception.
     """
-    queue: asyncio.Queue[_T | None] = asyncio.Queue(count * 2)
+    semaphore = asyncio.Semaphore(count)
+    tasks: set[asyncio.Task[None]] = set()
 
-    async def worker() -> None:
-        while True:
-            item = await queue.get()
-            if item is None:
-                break
-            try:
-                await callable_(item, *args, **kwargs)
-            finally:
-                queue.task_done()
+    async def work(item: _T) -> None:
+        try:
+            await callable_(item, *args, **kwargs)
+        finally:
+            semaphore.release()
 
-    async def fill_queue() -> None:
-        async for item in as_async_generator(iterable):
-            await queue.put(item)
-        for _ in range(count):
-            await queue.put(None)
-
-    fill_task = asyncio.create_task(fill_queue())
-    work_tasks = [asyncio.create_task(worker()) for _ in range(count)]
-    await asyncio.wait([fill_task, *work_tasks])
+    async for item in as_async_generator(iterable):
+        await semaphore.acquire()
+        task = asyncio.create_task(work(item))
+        tasks.add(task)
+        task.add_done_callback(tasks.discard)
+    if tasks:
+        await asyncio.wait(tasks)
 
 
 class AsyncioLoopingCall:
diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py
index 7c6235f29..f3ec03e85 100644
--- a/scrapy/utils/defer.py
+++ b/scrapy/utils/defer.py
@@ -149,21 +149,56 @@ def mustbe_deferred(
     return defer_result(result)
 
 
+class _ParallelTasks:
+    """Consumes *work* with no more than *count* concurrent cooperative tasks.
+
+    Tasks are started on demand: :meth:`start` must be called every time a new
+    item becomes available, so that a *count* much larger than the amount of
+    work does not cost anything.
+
+    :attr:`finished` fires once *work* is exhausted and every task is done.
+    """
+
+    def __init__(self, work: Iterator[Any], count: int):
+        self._coop = Cooperator()
+        self._work = work
+        self._count = count
+        self._running = 0
+        self.finished: Deferred[None] = Deferred()
+
+    def start(self) -> None:
+        if self._running >= self._count:
+            return
+        self._running += 1
+        self._coop.coiterate(self._work).addBoth(self._task_done)
+
+    def _task_done(self, _: Any) -> None:
+        self._running -= 1
+        # Only a running task can start another one, so once none is left no
+        # more work can come.
+        if not self._running:
+            self.finished.callback(None)
+
+
 def parallel(
     iterable: Iterable[_T],
     count: int,
     callable: Callable[Concatenate[_T, _P], _T2],  # noqa: A002
     *args: _P.args,
     **named: _P.kwargs,
-) -> Deferred[list[tuple[bool, Iterator[_T2]]]]:
+) -> Deferred[None]:
     """Execute a callable over the objects in the given iterable, in parallel,
     using no more than ``count`` concurrent calls.
-
-    Taken from: https://jcalderone.livejournal.com/24285.html
     """
-    coop = Cooperator()
-    work: Iterator[_T2] = (callable(elem, *args, **named) for elem in iterable)
-    return DeferredList([coop.coiterate(work) for _ in range(count)])
+
+    def work() -> Iterator[_T2]:
+        for elem in iterable:
+            tasks.start()
+            yield callable(elem, *args, **named)
+
+    tasks = _ParallelTasks(work(), count)
+    tasks.start()
+    return tasks.finished
 
 
 class _AsyncCooperatorAdapter(Iterator[Deferred[Any]], Generic[_T]):
@@ -226,12 +261,16 @@ class _AsyncCooperatorAdapter(Iterator[Deferred[Any]], Generic[_T]):
         self.finished: bool = False
         self.waiting_deferreds: deque[Deferred[Any]] = deque()
         self.anext_deferred: Deferred[_T] | None = None
+        # Called whenever aiterator produces a value, so that parallel_async()
+        # can start a task for it.
+        self.on_value: Callable[[], None] = lambda: None
 
     def _callback(self, result: _T) -> None:
         # This gets called when the result from aiterator.__anext__() is available.
         # It calls the callable on it and sends the result to the oldest waiting Deferred
         # (by chaining if the result is a Deferred too or by firing if not).
         self.anext_deferred = None
+        self.on_value()
         callable_result = self.callable(
             result, *self.callable_args, **self.callable_kwargs
         )
@@ -276,16 +315,13 @@ def parallel_async(
     callable: Callable[Concatenate[_T, _P], Deferred[Any] | None],  # noqa: A002
     *args: _P.args,
     **named: _P.kwargs,
-) -> Deferred[list[tuple[bool, Iterator[Deferred[Any]]]]]:
+) -> Deferred[None]:
     """Like ``parallel`` but for async iterators"""
-    coop = Cooperator()
-    work: Iterator[Deferred[Any]] = _AsyncCooperatorAdapter(
-        async_iterable, callable, *args, **named
-    )
-    dl: Deferred[list[tuple[bool, Iterator[Deferred[Any]]]]] = DeferredList(
-        [coop.coiterate(work) for _ in range(count)]
-    )
-    return dl
+    work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named)
+    tasks = _ParallelTasks(work, count)
+    work.on_value = tasks.start
+    tasks.start()
+    return tasks.finished
 
 
 def process_chain(
diff --git a/tests/test_utils_asyncio.py b/tests/test_utils_asyncio.py
index 4bd54acd4..59fa17594 100644
--- a/tests/test_utils_asyncio.py
+++ b/tests/test_utils_asyncio.py
@@ -115,6 +115,19 @@ class TestParallelAsyncio:
             assert parallel_count[0] == 0
             assert max_parallel_count[0] <= self.CONCURRENT_ITEMS
 
+    @coroutine_test
+    async def test_count_higher_than_work(self):
+        results: list[int] = []
+        task_counts: list[int] = []
+
+        async def callable_(o: int) -> None:
+            task_counts.append(len(asyncio.all_tasks()))
+            results.append(o)
+
+        await _parallel_asyncio(range(3), 1_000_000, callable_)
+        assert results == [0, 1, 2]
+        assert max(task_counts) < 100
+
 
 @pytest.mark.only_asyncio
 class TestAsyncioLoopingCall:
diff --git a/tests/test_utils_defer.py b/tests/test_utils_defer.py
index 175a4fe03..bb174e66f 100644
--- a/tests/test_utils_defer.py
+++ b/tests/test_utils_defer.py
@@ -4,9 +4,11 @@ import asyncio
 import random
 from asyncio import Future
 from typing import TYPE_CHECKING, Any
+from unittest import mock
 
 import pytest
 from twisted.internet.defer import Deferred, inlineCallbacks
+from twisted.internet.task import Cooperator
 
 from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.defer import (
@@ -17,6 +19,7 @@ from scrapy.utils.defer import (
     iter_errback,
     maybe_deferred_to_future,
     mustbe_deferred,
+    parallel,
     parallel_async,
 )
 from tests.utils.decorators import coroutine_test, inline_callbacks_test
@@ -129,6 +132,20 @@ class TestAsyncDefTestsuite:
         raise RuntimeError("This is expected to be raised")
 
 
+@pytest.mark.requires_reactor  # parallel() requires a reactor
+class TestParallel:
+    @inline_callbacks_test
+    def test_count_higher_than_work(self) -> Generator[Deferred[Any], Any, None]:
+        results: list[int] = []
+        with mock.patch.object(
+            Cooperator, "coiterate", autospec=True, side_effect=Cooperator.coiterate
+        ) as coiterate:
+            yield parallel(range(3), 1_000_000, results.append)
+        assert results == [0, 1, 2]
+        # One task per item, plus the one that finds no more work.
+        assert coiterate.call_count <= 4
+
+
 @pytest.mark.requires_reactor  # parallel_async() requires a reactor
 class TestParallelAsync:
     """This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
@@ -239,6 +256,17 @@ class TestParallelAsync:
             assert parallel_count[0] == 0
             assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0]
 
+    @inline_callbacks_test
+    def test_count_higher_than_work(self) -> Generator[Deferred[Any], Any, None]:
+        results: list[int] = []
+        with mock.patch.object(
+            Cooperator, "coiterate", autospec=True, side_effect=Cooperator.coiterate
+        ) as coiterate:
+            yield parallel_async(self.get_async_iterable(3), 1_000_000, results.append)
+        assert sorted(results) == [0, 1, 2]
+        # One task per item, plus the one that finds no more work.
+        assert coiterate.call_count <= 4
+
 
 class TestDeferredFromCoro:
     def test_deferred(self):