diff --git a/CHANGELOG.md b/CHANGELOG.md
index 33371e7..44846b5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,23 @@
 
 ### Added
 
+- **`--threads M` on `run_database_tasks`.** The worker runs M tasks at
+  once in executor threads while staying one process with one polling
+  thread, which still does the fetching, the broker receives and the signal
+  handling. The polling load stays at one fetch per `--interval`, a broker
+  receive asks for no more messages than there are idle threads, and the
+  claim is made by the polling thread with the executor's worker id, so the
+  next fetch never hands out the same row. Each thread records its own id
+  (`<worker id>-t1` to `-tM`) on the tasks it runs, has a database
+  connection of its own, and calls `close_old_connections()` after every
+  task. `--max-tasks` counts the tasks handed out, failures are counted
+  across the threads, and on `SIGTERM` the tasks in the threads finish
+  before the worker exits. The task code must be thread-safe, and the option
+  only helps I/O-bound tasks; the README's *`--threads M`* and *Trade-offs*
+  sections say what it saves (memory, not connections) and what it costs.
+  Several threads are refused on SQLite, as several workers are. The default
+  is 1, with the loop unchanged, so existing projects need no changes.
+  ([#25](https://github.com/tokibito/django-database-task/issues/25))
 - **`claim_task()` and `run_claimed_task()` on the backend.** The two steps
   of `DatabaseTaskBackend.run_task()`, which now calls one and then the
   other, so that a worker can claim a task in the thread that fetched it and
diff --git a/README.md b/README.md
index 41654b7..7ba4c96 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,7 @@ A database-backed task queue backend for Django's built-in task framework.
 - **Django Admin integration** - View and manage tasks from the admin interface
 - **Async support** - Supports async task functions
 - **Graceful shutdown** - Workers finish the running task before exiting on `SIGTERM`
-- **Parallel workers** - `--workers N` keeps N worker processes running from one command, or run one per systemd unit / Kubernetes replica as before
+- **Parallel workers** - `--workers N` keeps N worker processes running from one command, `--threads M` runs M tasks at once inside a worker, or run one process per systemd unit / Kubernetes replica as before
 - **Crash recovery** - Tasks stranded in `RUNNING` by a killed worker are found and requeued
 - **Monitoring** - Counts per status and the age of the oldest pending task, from Python or `/tasks/status/`
 - **Job scheduler friendly** - Opt-in exit codes that tell an idle run from a failed one, plus structured log fields for JP1 / Hinemos / cron / systemd timers
@@ -260,6 +260,7 @@ python manage.py run_database_tasks [options]
 | `--wait-time` | Seconds to wait for a broker message before looking again (default: 20) |
 | `--max-messages` | Maximum number of broker messages to receive at a time (default: 1) |
 | `--workers` | Number of worker processes to run from this command (default: 1). See [Running several workers](#running-several-workers) |
+| `--threads` | Number of threads running tasks inside the worker (default: 1). The task code must be thread-safe. See [`--threads M`](#--threads-m) |
 | `--shutdown-timeout` | Maximum seconds to wait for the running task after `SIGTERM`/`SIGINT` before forcing exit (0=wait indefinitely, default: 0) |
 | `--no-graceful-shutdown` | Do not install signal handlers (terminate immediately, even while a task is running) |
 | `--empty-exit-code` | Exit with this code when no task was processed (0=exit normally, default: 0) |
@@ -699,12 +700,13 @@ with GracefulShutdown(timeout=50) as shutdown:
 `run_database_tasks` is one process running one task at a time. To run more
 tasks at once, run more of them: every worker claims its own tasks with
 `SELECT FOR UPDATE SKIP LOCKED`, so workers on one host or on many never run
-the same task. There are two ways to keep N of them running.
+the same task. There are three tiers, and they combine.
 
-| | The processes are kept running by | Reach for it when |
+| | Unit of parallelism | Reach for it when |
 |---|---|---|
-| A process manager | systemd template units, Kubernetes replicas, supervisord `numprocs` | You have one. It restarts, logs and health-checks the workers, and the command's default (one process, with the signals and exit codes described above) is what it expects |
-| `--workers N` | The command itself | One host without a process manager, or one unit or container that should hold N workers: a small supervisor inside the command keeps N worker processes running |
+| A process manager | Processes, kept running by systemd template units, Kubernetes replicas, supervisord `numprocs` | You have one. It restarts, logs and health-checks the workers, and the command's default (one process, with the signals and exit codes described above) is what it expects |
+| `--workers N` | Processes, kept running by the command itself | One host without a process manager, or one unit or container that should hold N workers: a small supervisor inside the command keeps N worker processes running |
+| `--threads M` | Threads inside one worker process | Memory is tight and the tasks are I/O-bound and written thread-safe: M tasks in progress for the footprint of one process, with one poller |
 
 ### `--workers N`
 
@@ -767,11 +769,52 @@ Shutdown complete: every worker exited.
   [requeue_stale_database_tasks](#recovering-tasks-left-in-running-status)
   is for.
 
+### `--threads M`
+
+```bash
+python manage.py run_database_tasks --continuous --threads 8
+```
+
+The worker stays one process with one polling thread, which is still the
+one that fetches, receives from the broker and handles the signals. What
+changes is that the task it found is handed to one of M executor threads
+instead of being run inline, so M tasks are in progress at once. The polling
+load stays at one fetch per `--interval`; a broker receive asks for no more
+messages than there are idle threads, up to `--max-messages`.
+
+- **Thread safety is on you.** The M tasks share one interpreter, one set of
+  module globals and whatever your task code caches. A task that is not
+  thread-safe, that calls `os._exit()`, or that crashes the interpreter takes
+  the other M-1 with it. `--workers` asks nothing of the task code; this
+  option does, and that is the trade it makes explicit.
+- **I/O-bound only.** The GIL serialises Python bytecode across the threads,
+  so CPU-bound tasks gain nothing from M; HTTP calls, database-heavy tasks
+  and anything that waits do. For CPU-bound work use `--workers`.
+- **Memory, not connections.** Threads share the process, so M tasks cost
+  roughly one worker's footprint rather than M. They do not save database
+  connections: Django keeps one per thread, so a worker with M threads holds
+  M connections, plus the polling thread's, plus the `LISTEN` connection with
+  the PostgreSQL broker. Each executor calls `close_old_connections()` after
+  every task, the way a request does, so `CONN_MAX_AGE` applies.
+- **Worker ids.** Each thread records its own id on the tasks it runs,
+  `<worker id>-t1` to `<worker id>-tM`, so `worker_ids` and the log fields
+  say which thread ran what. The claim itself is made by the polling thread,
+  with the executor's id, so the next fetch never hands out the same row.
+- **Shutdown and `--max-tasks`** work as with one thread: on `SIGTERM` no
+  more tasks are handed out and the ones in the threads finish, within
+  `--shutdown-timeout` if set; `--max-tasks` counts the tasks handed out, so
+  no more than that many are ever started.
+- **Combining.** `--workers N --threads M` is N processes with M threads
+  each: N pollers, N×M tasks in progress, process isolation between the N.
+- **SQLite** is refused above one thread, as with `--workers`: without row
+  locks, claims and finishing writes from several threads collide on the
+  database lock.
+
 ### Trade-offs
 
 Processes are the simple, robust unit of parallelism, and they are not free.
 What `--workers N` buys and what it costs, so you can pick N with your eyes
-open:
+open, and where `--threads M` stands against it:
 
 - **Isolation.** Every worker is a process of its own. A task that crashes the
   interpreter, leaks until the OOM killer steps in, or blocks forever takes
@@ -818,13 +861,12 @@ open:
   running `--workers N`, is a reasonable middle: the manager handles the
   machine-level concerns and the supervisor the fan-out.
 - **SQLite** has no row locks, so the command refuses `--workers` above 1
-  there. Use PostgreSQL, MySQL or MariaDB.
-
-Threads inside a worker, the memory-lean option for I/O-bound tasks that are
-written thread-safe, are not part of this release; the design is tracked in
-[#25](https://github.com/tokibito/django-database-task/issues/25). On this
-backend they would save memory but not database connections, since each
-thread needs a connection of its own.
+  there, and `--threads` above 1 too. Use PostgreSQL, MySQL or MariaDB.
+- **Threads instead.** [`--threads M`](#--threads-m) trades the isolation
+  and the CPU scaling above for memory: M tasks in progress for one
+  process's footprint, as long as the tasks are I/O-bound and thread-safe.
+  It does not reduce database connections, and one bad task can take the
+  whole worker down. When memory is not the constraint, prefer processes.
 
 ## Running from a job scheduler
 
diff --git a/django_database_task/locale/es/LC_MESSAGES/django.mo b/django_database_task/locale/es/LC_MESSAGES/django.mo
index 4198add..c19ee00 100644
Binary files a/django_database_task/locale/es/LC_MESSAGES/django.mo and b/django_database_task/locale/es/LC_MESSAGES/django.mo differ
diff --git a/django_database_task/locale/es/LC_MESSAGES/django.po b/django_database_task/locale/es/LC_MESSAGES/django.po
index f78b41c..c61c524 100644
--- a/django_database_task/locale/es/LC_MESSAGES/django.po
+++ b/django_database_task/locale/es/LC_MESSAGES/django.po
@@ -6,7 +6,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: django-database-task 0.1.0\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-09-30 17:27+0000\n"
+"POT-Creation-Date: 2026-10-02 05:07+0000\n"
 "PO-Revision-Date: 2026-09-30 00:00+0000\n"
 "Last-Translator: Shinya Okano <tokibito@gmail.com>\n"
 "Language-Team: Spanish\n"
@@ -135,31 +135,31 @@ msgstr "Número de tareas a procesar de una vez (predeterminado: 1000)"
 msgid "Show what would happen without changing anything"
 msgstr "Muestra lo que ocurriría sin cambiar nada"
 
-#: management/commands/run_database_tasks.py:55
+#: management/commands/run_database_tasks.py:63
 msgid "Execute tasks queued in the database"
 msgstr "Ejecuta las tareas encoladas en la base de datos"
 
-#: management/commands/run_database_tasks.py:62
+#: management/commands/run_database_tasks.py:70
 msgid "Queue name to process (all queues if not specified)"
 msgstr "Nombre de la cola a procesar (todas las colas si no se indica)"
 
-#: management/commands/run_database_tasks.py:68
+#: management/commands/run_database_tasks.py:76
 msgid "Backend name (default: default)"
 msgstr "Nombre del backend (predeterminado: default)"
 
-#: management/commands/run_database_tasks.py:73
+#: management/commands/run_database_tasks.py:81
 msgid "Continuous mode (keep polling even when no tasks)"
 msgstr "Modo continuo (sigue sondeando aunque no haya tareas)"
 
-#: management/commands/run_database_tasks.py:79
+#: management/commands/run_database_tasks.py:87
 msgid "Polling interval in seconds for continuous mode (default: 5)"
 msgstr "Intervalo de sondeo en segundos en modo continuo (predeterminado: 5)"
 
-#: management/commands/run_database_tasks.py:85
+#: management/commands/run_database_tasks.py:93
 msgid "Maximum number of tasks to process (0=unlimited, default: 0)"
 msgstr "Número máximo de tareas a procesar (0=ilimitado, predeterminado: 0)"
 
-#: management/commands/run_database_tasks.py:92
+#: management/commands/run_database_tasks.py:100
 msgid ""
 "Where to look for tasks: 'db' polls the database, 'broker' receives from the "
 "backend's broker, 'both' does the broker first and falls back to the "
@@ -171,7 +171,7 @@ msgstr ""
 "datos, y 'auto' (predeterminado) significa 'both' cuando el backend tiene un "
 "broker del que recibir y 'db' en caso contrario"
 
-#: management/commands/run_database_tasks.py:104
+#: management/commands/run_database_tasks.py:112
 msgid ""
 "Seconds to wait for a broker message before looking again. Replaces --"
 "interval as the idle wait when receiving from a broker (default: 20)"
@@ -180,12 +180,12 @@ msgstr ""
 "Sustituye a --interval como espera inactiva al recibir de un broker "
 "(predeterminado: 20)"
 
-#: management/commands/run_database_tasks.py:114
+#: management/commands/run_database_tasks.py:122
 msgid "Maximum number of broker messages to receive at a time (default: 1)"
 msgstr ""
 "Número máximo de mensajes del broker a recibir de una vez (predeterminado: 1)"
 
-#: management/commands/run_database_tasks.py:123
+#: management/commands/run_database_tasks.py:131
 msgid ""
 "Run this many worker processes from one command, each a copy of this command "
 "without --workers. With --continuous a worker that exits is replaced; "
@@ -196,7 +196,18 @@ msgstr ""
 "se reemplaza; sin él, el comando termina cuando todos los workers lo hayan "
 "hecho (predeterminado: 1)"
 
-#: management/commands/run_database_tasks.py:134
+#: management/commands/run_database_tasks.py:143
+msgid ""
+"Run tasks in this many threads inside the worker. The task code must be "
+"thread-safe. The worker still polls once per interval; each thread has a "
+"database connection of its own (default: 1)"
+msgstr ""
+"Ejecuta las tareas en esta cantidad de hilos dentro del worker. El código "
+"de las tareas debe ser seguro para hilos. El worker sigue consultando una "
+"vez por intervalo; cada hilo tiene su propia conexión a la base de datos "
+"(predeterminado: 1)"
+
+#: management/commands/run_database_tasks.py:154
 msgid ""
 "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT "
 "before forcing exit (0=wait indefinitely, default: 0)"
@@ -205,7 +216,7 @@ msgstr ""
 "SIGINT antes de forzar la salida (0=esperar indefinidamente, predeterminado: "
 "0)"
 
-#: management/commands/run_database_tasks.py:143
+#: management/commands/run_database_tasks.py:163
 msgid ""
 "Do not install SIGTERM/SIGINT handlers; the process is terminated "
 "immediately, even while a task is running"
@@ -213,7 +224,7 @@ msgstr ""
 "No instala manejadores de SIGTERM/SIGINT; el proceso termina de inmediato, "
 "incluso con una tarea en ejecución"
 
-#: management/commands/run_database_tasks.py:153
+#: management/commands/run_database_tasks.py:173
 msgid ""
 "Exit with this code when no task was processed, so a job scheduler can tell "
 "an idle run from a real one (0=exit normally, default: 0)"
@@ -222,7 +233,7 @@ msgstr ""
 "planificador de trabajos distinga una ejecución vacía de una real "
 "(0=terminar normalmente, predeterminado: 0)"
 
-#: management/commands/run_database_tasks.py:164
+#: management/commands/run_database_tasks.py:184
 msgid ""
 "Exit with this code when at least one task failed or could not be run. Takes "
 "precedence over --empty-exit-code (0=exit normally, default: 0)"
diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.mo b/django_database_task/locale/ja/LC_MESSAGES/django.mo
index 130d80b..532ff55 100644
Binary files a/django_database_task/locale/ja/LC_MESSAGES/django.mo and b/django_database_task/locale/ja/LC_MESSAGES/django.mo differ
diff --git a/django_database_task/locale/ja/LC_MESSAGES/django.po b/django_database_task/locale/ja/LC_MESSAGES/django.po
index bb4bade..007e2a2 100644
--- a/django_database_task/locale/ja/LC_MESSAGES/django.po
+++ b/django_database_task/locale/ja/LC_MESSAGES/django.po
@@ -7,7 +7,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: django-database-task 0.1.0\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-09-30 17:27+0000\n"
+"POT-Creation-Date: 2026-10-02 05:07+0000\n"
 "PO-Revision-Date: 2025-12-27 00:00+0000\n"
 "Last-Translator: Shinya Okano <tokibito@gmail.com>\n"
 "Language-Team: Japanese\n"
@@ -129,31 +129,31 @@ msgstr "一度に処理するタスク数 (デフォルト: 1000)"
 msgid "Show what would happen without changing anything"
 msgstr "何も変更せずに、何が行われるかを表示"
 
-#: management/commands/run_database_tasks.py:55
+#: management/commands/run_database_tasks.py:63
 msgid "Execute tasks queued in the database"
 msgstr "データベースのキューに登録されたタスクを実行"
 
-#: management/commands/run_database_tasks.py:62
+#: management/commands/run_database_tasks.py:70
 msgid "Queue name to process (all queues if not specified)"
 msgstr "処理するキュー名 (指定しない場合はすべてのキュー)"
 
-#: management/commands/run_database_tasks.py:68
+#: management/commands/run_database_tasks.py:76
 msgid "Backend name (default: default)"
 msgstr "バックエンド名 (デフォルト: default)"
 
-#: management/commands/run_database_tasks.py:73
+#: management/commands/run_database_tasks.py:81
 msgid "Continuous mode (keep polling even when no tasks)"
 msgstr "継続モード (タスクがなくてもポーリングを続ける)"
 
-#: management/commands/run_database_tasks.py:79
+#: management/commands/run_database_tasks.py:87
 msgid "Polling interval in seconds for continuous mode (default: 5)"
 msgstr "継続モードのポーリング間隔（秒）(デフォルト: 5)"
 
-#: management/commands/run_database_tasks.py:85
+#: management/commands/run_database_tasks.py:93
 msgid "Maximum number of tasks to process (0=unlimited, default: 0)"
 msgstr "処理する最大タスク数 (0=無制限、デフォルト: 0)"
 
-#: management/commands/run_database_tasks.py:92
+#: management/commands/run_database_tasks.py:100
 msgid ""
 "Where to look for tasks: 'db' polls the database, 'broker' receives from the "
 "backend's broker, 'both' does the broker first and falls back to the "
@@ -165,7 +165,7 @@ msgstr ""
 "バックする。'auto' (デフォルト) は、受信できるブローカーがバックエンドにあれ"
 "ば 'both'、なければ 'db' になる"
 
-#: management/commands/run_database_tasks.py:104
+#: management/commands/run_database_tasks.py:112
 msgid ""
 "Seconds to wait for a broker message before looking again. Replaces --"
 "interval as the idle wait when receiving from a broker (default: 20)"
@@ -173,21 +173,32 @@ msgstr ""
 "再確認するまでブローカーのメッセージを待つ秒数。ブローカーから受信する場合、"
 "アイドル時の待機として --interval の代わりに使われる (デフォルト: 20)"
 
-#: management/commands/run_database_tasks.py:114
+#: management/commands/run_database_tasks.py:122
 msgid "Maximum number of broker messages to receive at a time (default: 1)"
 msgstr "一度に受信するブローカーメッセージの最大数 (デフォルト: 1)"
 
-#: management/commands/run_database_tasks.py:123
+#: management/commands/run_database_tasks.py:131
 msgid ""
 "Run this many worker processes from one command, each a copy of this command "
 "without --workers. With --continuous a worker that exits is replaced; "
 "without it the command exits when every worker has (default: 1)"
 msgstr ""
-"このコマンドから起動するワーカープロセスの数。各ワーカーは --workers を除いたこのコマンドの"
-"コピー。--continuous 指定時は終了したワーカーを再起動し、未指定時は全ワーカーが終了する"
-"とコマンドも終了する (デフォルト: 1)"
+"このコマンドから起動するワーカープロセスの数。各ワーカーは --workers を除いた"
+"このコマンドのコピー。--continuous 指定時は終了したワーカーを再起動し、未指定"
+"時は全ワーカーが終了するとコマンドも終了する (デフォルト: 1)"
 
-#: management/commands/run_database_tasks.py:134
+#: management/commands/run_database_tasks.py:143
+msgid ""
+"Run tasks in this many threads inside the worker. The task code must be "
+"thread-safe. The worker still polls once per interval; each thread has a "
+"database connection of its own (default: 1)"
+msgstr ""
+"ワーカー内でこの数のスレッドでタスクを実行する。タスクのコードはスレッドセーフである必要がある。ポーリングは引き続き "
+"interval ごとに 1 "
+"回で、スレッドごとにデータベース接続を 1 本持つ "
+"(デフォルト: 1)"
+
+#: management/commands/run_database_tasks.py:154
 msgid ""
 "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT "
 "before forcing exit (0=wait indefinitely, default: 0)"
@@ -195,7 +206,7 @@ msgstr ""
 "SIGTERM/SIGINTを受信してから強制終了するまで、実行中のタスクを待つ最大秒数 "
 "(0=無期限に待つ、デフォルト: 0)"
 
-#: management/commands/run_database_tasks.py:143
+#: management/commands/run_database_tasks.py:163
 msgid ""
 "Do not install SIGTERM/SIGINT handlers; the process is terminated "
 "immediately, even while a task is running"
@@ -203,7 +214,7 @@ msgstr ""
 "SIGTERM/SIGINTハンドラーを設定しない。タスクの実行中でもプロセスは即座に終了"
 "する"
 
-#: management/commands/run_database_tasks.py:153
+#: management/commands/run_database_tasks.py:173
 msgid ""
 "Exit with this code when no task was processed, so a job scheduler can tell "
 "an idle run from a real one (0=exit normally, default: 0)"
@@ -211,7 +222,7 @@ msgstr ""
 "タスクを1件も処理しなかった場合にこの終了コードで終了する。ジョブスケジュー"
 "ラーが空振りの実行を区別できるようにする (0=通常どおり終了、デフォルト: 0)"
 
-#: management/commands/run_database_tasks.py:164
+#: management/commands/run_database_tasks.py:184
 msgid ""
 "Exit with this code when at least one task failed or could not be run. Takes "
 "precedence over --empty-exit-code (0=exit normally, default: 0)"
diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo b/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo
index 7dd4daa..275e84e 100644
Binary files a/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo and b/django_database_task/locale/pt_BR/LC_MESSAGES/django.mo differ
diff --git a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po
index 69700ab..8ca5a27 100644
--- a/django_database_task/locale/pt_BR/LC_MESSAGES/django.po
+++ b/django_database_task/locale/pt_BR/LC_MESSAGES/django.po
@@ -6,7 +6,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: django-database-task 0.1.0\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-09-30 17:27+0000\n"
+"POT-Creation-Date: 2026-10-02 05:07+0000\n"
 "PO-Revision-Date: 2026-09-30 00:00+0000\n"
 "Last-Translator: Shinya Okano <tokibito@gmail.com>\n"
 "Language-Team: Brazilian Portuguese\n"
@@ -132,31 +132,31 @@ msgstr "Número de tarefas a processar de uma vez (padrão: 1000)"
 msgid "Show what would happen without changing anything"
 msgstr "Mostra o que aconteceria sem alterar nada"
 
-#: management/commands/run_database_tasks.py:55
+#: management/commands/run_database_tasks.py:63
 msgid "Execute tasks queued in the database"
 msgstr "Executa as tarefas enfileiradas no banco de dados"
 
-#: management/commands/run_database_tasks.py:62
+#: management/commands/run_database_tasks.py:70
 msgid "Queue name to process (all queues if not specified)"
 msgstr "Nome da fila a processar (todas as filas se não for informado)"
 
-#: management/commands/run_database_tasks.py:68
+#: management/commands/run_database_tasks.py:76
 msgid "Backend name (default: default)"
 msgstr "Nome do backend (padrão: default)"
 
-#: management/commands/run_database_tasks.py:73
+#: management/commands/run_database_tasks.py:81
 msgid "Continuous mode (keep polling even when no tasks)"
 msgstr "Modo contínuo (continua consultando mesmo sem tarefas)"
 
-#: management/commands/run_database_tasks.py:79
+#: management/commands/run_database_tasks.py:87
 msgid "Polling interval in seconds for continuous mode (default: 5)"
 msgstr "Intervalo de consulta em segundos no modo contínuo (padrão: 5)"
 
-#: management/commands/run_database_tasks.py:85
+#: management/commands/run_database_tasks.py:93
 msgid "Maximum number of tasks to process (0=unlimited, default: 0)"
 msgstr "Número máximo de tarefas a processar (0=ilimitado, padrão: 0)"
 
-#: management/commands/run_database_tasks.py:92
+#: management/commands/run_database_tasks.py:100
 msgid ""
 "Where to look for tasks: 'db' polls the database, 'broker' receives from the "
 "backend's broker, 'both' does the broker first and falls back to the "
@@ -168,7 +168,7 @@ msgstr ""
 "e 'auto' (padrão) significa 'both' quando o backend tem um broker do qual "
 "receber e 'db' caso contrário"
 
-#: management/commands/run_database_tasks.py:104
+#: management/commands/run_database_tasks.py:112
 msgid ""
 "Seconds to wait for a broker message before looking again. Replaces --"
 "interval as the idle wait when receiving from a broker (default: 20)"
@@ -176,22 +176,33 @@ msgstr ""
 "Segundos de espera por uma mensagem do broker antes de verificar novamente. "
 "Substitui --interval como espera ociosa ao receber de um broker (padrão: 20)"
 
-#: management/commands/run_database_tasks.py:114
+#: management/commands/run_database_tasks.py:122
 msgid "Maximum number of broker messages to receive at a time (default: 1)"
 msgstr "Número máximo de mensagens do broker a receber de uma vez (padrão: 1)"
 
-#: management/commands/run_database_tasks.py:123
+#: management/commands/run_database_tasks.py:131
 msgid ""
 "Run this many worker processes from one command, each a copy of this command "
 "without --workers. With --continuous a worker that exits is replaced; "
 "without it the command exits when every worker has (default: 1)"
 msgstr ""
-"Executa esta quantidade de processos worker a partir de um único comando, cada "
-"um uma cópia deste comando sem --workers. Com --continuous, um worker que sai "
-"é substituído; sem ele, o comando termina quando todos os workers terminarem "
+"Executa esta quantidade de processos worker a partir de um único comando, "
+"cada um uma cópia deste comando sem --workers. Com --continuous, um worker "
+"que sai é substituído; sem ele, o comando termina quando todos os workers "
+"terminarem (padrão: 1)"
+
+#: management/commands/run_database_tasks.py:143
+msgid ""
+"Run tasks in this many threads inside the worker. The task code must be "
+"thread-safe. The worker still polls once per interval; each thread has a "
+"database connection of its own (default: 1)"
+msgstr ""
+"Executa as tarefas em esta quantidade de threads dentro do worker. O código "
+"das tarefas precisa ser thread-safe. O worker continua consultando uma vez "
+"por intervalo; cada thread tem a sua própria conexão com o banco de dados "
 "(padrão: 1)"
 
-#: management/commands/run_database_tasks.py:134
+#: management/commands/run_database_tasks.py:154
 msgid ""
 "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT "
 "before forcing exit (0=wait indefinitely, default: 0)"
@@ -199,7 +210,7 @@ msgstr ""
 "Tempo máximo, em segundos, de espera pela tarefa em execução após receber "
 "SIGTERM/SIGINT antes de forçar a saída (0=esperar indefinidamente, padrão: 0)"
 
-#: management/commands/run_database_tasks.py:143
+#: management/commands/run_database_tasks.py:163
 msgid ""
 "Do not install SIGTERM/SIGINT handlers; the process is terminated "
 "immediately, even while a task is running"
@@ -207,7 +218,7 @@ msgstr ""
 "Não instala manipuladores de SIGTERM/SIGINT; o processo é encerrado "
 "imediatamente, mesmo com uma tarefa em execução"
 
-#: management/commands/run_database_tasks.py:153
+#: management/commands/run_database_tasks.py:173
 msgid ""
 "Exit with this code when no task was processed, so a job scheduler can tell "
 "an idle run from a real one (0=exit normally, default: 0)"
@@ -216,7 +227,7 @@ msgstr ""
 "agendador de jobs distinga uma execução ociosa de uma real (0=sair "
 "normalmente, padrão: 0)"
 
-#: management/commands/run_database_tasks.py:164
+#: management/commands/run_database_tasks.py:184
 msgid ""
 "Exit with this code when at least one task failed or could not be run. Takes "
 "precedence over --empty-exit-code (0=exit normally, default: 0)"
diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo
index f77b2be..1920335 100644
Binary files a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo and b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.mo differ
diff --git a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po
index adff526..0035b39 100644
--- a/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po
+++ b/django_database_task/locale/zh_Hans/LC_MESSAGES/django.po
@@ -6,7 +6,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: django-database-task 0.1.0\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-09-30 17:27+0000\n"
+"POT-Creation-Date: 2026-10-02 05:07+0000\n"
 "PO-Revision-Date: 2026-09-30 00:00+0000\n"
 "Last-Translator: Shinya Okano <tokibito@gmail.com>\n"
 "Language-Team: Simplified Chinese\n"
@@ -123,31 +123,31 @@ msgstr "每次处理的任务数（默认：1000）"
 msgid "Show what would happen without changing anything"
 msgstr "显示将会执行的操作，但不做任何更改"
 
-#: management/commands/run_database_tasks.py:55
+#: management/commands/run_database_tasks.py:63
 msgid "Execute tasks queued in the database"
 msgstr "执行在数据库中排队的任务"
 
-#: management/commands/run_database_tasks.py:62
+#: management/commands/run_database_tasks.py:70
 msgid "Queue name to process (all queues if not specified)"
 msgstr "要处理的队列名称（未指定时为所有队列）"
 
-#: management/commands/run_database_tasks.py:68
+#: management/commands/run_database_tasks.py:76
 msgid "Backend name (default: default)"
 msgstr "后端名称（默认：default）"
 
-#: management/commands/run_database_tasks.py:73
+#: management/commands/run_database_tasks.py:81
 msgid "Continuous mode (keep polling even when no tasks)"
 msgstr "持续模式（没有任务时也继续轮询）"
 
-#: management/commands/run_database_tasks.py:79
+#: management/commands/run_database_tasks.py:87
 msgid "Polling interval in seconds for continuous mode (default: 5)"
 msgstr "持续模式的轮询间隔秒数（默认：5）"
 
-#: management/commands/run_database_tasks.py:85
+#: management/commands/run_database_tasks.py:93
 msgid "Maximum number of tasks to process (0=unlimited, default: 0)"
 msgstr "要处理的最大任务数（0=不限，默认：0）"
 
-#: management/commands/run_database_tasks.py:92
+#: management/commands/run_database_tasks.py:100
 msgid ""
 "Where to look for tasks: 'db' polls the database, 'broker' receives from the "
 "backend's broker, 'both' does the broker first and falls back to the "
@@ -158,7 +158,7 @@ msgstr ""
 "消息代理再回退到数据库，'auto'（默认）在后端有可接收的消息代理时为 'both'，否"
 "则为 'db'"
 
-#: management/commands/run_database_tasks.py:104
+#: management/commands/run_database_tasks.py:112
 msgid ""
 "Seconds to wait for a broker message before looking again. Replaces --"
 "interval as the idle wait when receiving from a broker (default: 20)"
@@ -166,20 +166,29 @@ msgstr ""
 "再次查看前等待消息代理消息的秒数。从消息代理接收时，取代 --interval 作为空闲"
 "时的等待（默认：20）"
 
-#: management/commands/run_database_tasks.py:114
+#: management/commands/run_database_tasks.py:122
 msgid "Maximum number of broker messages to receive at a time (default: 1)"
 msgstr "每次接收的消息代理消息的最大数量（默认：1）"
 
-#: management/commands/run_database_tasks.py:123
+#: management/commands/run_database_tasks.py:131
 msgid ""
 "Run this many worker processes from one command, each a copy of this command "
 "without --workers. With --continuous a worker that exits is replaced; "
 "without it the command exits when every worker has (default: 1)"
 msgstr ""
-"从一条命令启动这么多个工作进程，每个都是不带 --workers 的本命令副本。指定 --continuous 时，"
-"退出的工作进程会被替换；未指定时，所有工作进程退出后命令即退出（默认：1）"
+"从一条命令启动这么多个工作进程，每个都是不带 --workers 的本命令副本。指定 --"
+"continuous 时，退出的工作进程会被替换；未指定时，所有工作进程退出后命令即退出"
+"（默认：1）"
 
-#: management/commands/run_database_tasks.py:134
+#: management/commands/run_database_tasks.py:143
+msgid ""
+"Run tasks in this many threads inside the worker. The task code must be "
+"thread-safe. The worker still polls once per interval; each thread has a "
+"database connection of its own (default: 1)"
+msgstr ""
+"在工作进程内用这么多个线程运行任务。任务代码必须是线程安全的。工作进程仍每个间隔轮询一次；每个线程有自己的数据库连接（默认：1）"
+
+#: management/commands/run_database_tasks.py:154
 msgid ""
 "Maximum seconds to wait for the running task after receiving SIGTERM/SIGINT "
 "before forcing exit (0=wait indefinitely, default: 0)"
@@ -187,13 +196,13 @@ msgstr ""
 "收到 SIGTERM/SIGINT 后、强制退出前等待正在运行的任务的最长秒数（0=无限期等"
 "待，默认：0）"
 
-#: management/commands/run_database_tasks.py:143
+#: management/commands/run_database_tasks.py:163
 msgid ""
 "Do not install SIGTERM/SIGINT handlers; the process is terminated "
 "immediately, even while a task is running"
 msgstr "不安装 SIGTERM/SIGINT 处理程序；即使有任务正在运行，进程也会立即终止"
 
-#: management/commands/run_database_tasks.py:153
+#: management/commands/run_database_tasks.py:173
 msgid ""
 "Exit with this code when no task was processed, so a job scheduler can tell "
 "an idle run from a real one (0=exit normally, default: 0)"
@@ -201,7 +210,7 @@ msgstr ""
 "未处理任何任务时以此代码退出，以便作业调度器区分空转的运行与实际的运行（0=正"
 "常退出，默认：0）"
 
-#: management/commands/run_database_tasks.py:164
+#: management/commands/run_database_tasks.py:184
 msgid ""
 "Exit with this code when at least one task failed or could not be run. Takes "
 "precedence over --empty-exit-code (0=exit normally, default: 0)"
diff --git a/django_database_task/management/commands/run_database_tasks.py b/django_database_task/management/commands/run_database_tasks.py
index 19f8f3d..524e2c6 100644
--- a/django_database_task/management/commands/run_database_tasks.py
+++ b/django_database_task/management/commands/run_database_tasks.py
@@ -1,7 +1,9 @@
 import argparse
+import functools
 import logging
 import socket
 import sys
+import threading
 import uuid
 from contextlib import ExitStack
 
@@ -17,6 +19,7 @@
 from django_database_task.models import DatabaseTask
 from django_database_task.shutdown import GracefulShutdown, signal_name
 from django_database_task.supervisor import WorkerSupervisor, worker_arguments
+from django_database_task.threads import ExecutorThreads
 
 #: Where the worker looks for tasks to run.
 SOURCE_AUTO = "auto"
@@ -27,6 +30,11 @@
 
 logger = logging.getLogger("django_database_task")
 
+#: How long the main thread waits for an executor thread to report back
+#: before looking at the shutdown flag again. A result wakes it at once;
+#: this only bounds the wait.
+COLLECT_WAIT = 1.0
+
 
 def _exit_code_argument(value):
     """Parse an exit code option, rejecting what a shell cannot report."""
@@ -126,6 +134,18 @@ def add_arguments(self, parser):
                 "exits when every worker has (default: 1)"
             ),
         )
+        parser.add_argument(
+            "--threads",
+            type=int,
+            default=1,
+            metavar="M",
+            help=_(
+                "Run tasks in this many threads inside the worker. The task "
+                "code must be thread-safe. The worker still polls once per "
+                "interval; each thread has a database connection of its own "
+                "(default: 1)"
+            ),
+        )
         parser.add_argument(
             "--shutdown-timeout",
             type=float,
@@ -185,6 +205,9 @@ def handle(self, *args, **options):
             raise CommandError("--max-messages must be at least 1")
         if options["workers"] < 1:
             raise CommandError("--workers must be at least 1")
+        threads = options["threads"]
+        if threads < 1:
+            raise CommandError("--threads must be at least 1")
 
         backend = task_backends[backend_name]
         source = self._resolve_source(backend, options["source"])
@@ -200,6 +223,8 @@ def handle(self, *args, **options):
             self.stdout.write(f"Worker ID: {worker_id}")
             self.stdout.write(f"Backend: {backend_name}")
             self.stdout.write(f"Source: {source}")
+            if threads > 1:
+                self.stdout.write(f"Threads: {threads}")
             if queue_name:
                 self.stdout.write(f"Queue: {queue_name}")
             if continuous:
@@ -224,8 +249,10 @@ def handle(self, *args, **options):
         self.verbosity = verbosity
         # Counted here rather than returned from the loop because a task can
         # fail at several depths (the run itself, the broker message that
-        # named it) and every one of them feeds the same exit code.
+        # named it) and every one of them feeds the same exit code. With
+        # --threads it is counted from several threads, hence the lock.
         self.tasks_failed = 0
+        self._failures_lock = threading.Lock()
 
         logger.info(
             "Worker started: id=%s backend=%s source=%s",
@@ -238,6 +265,7 @@ def handle(self, *args, **options):
                 "source": source,
                 "queue_name": queue_name,
                 "continuous": continuous,
+                "threads": threads,
             },
         )
 
@@ -255,6 +283,11 @@ def handle(self, *args, **options):
                 # A broker a worker receives from may hold a connection
                 # open, so release it however the loop ends.
                 stack.callback(self._close_broker, backend.broker)
+            executors = None
+            if threads > 1:
+                self._check_database_supports_concurrency("--threads", threads)
+                executors = ExecutorThreads(threads, worker_id).start()
+                stack.callback(executors.stop)
             tasks_processed = self._process_tasks(
                 shutdown=shutdown,
                 backend=backend,
@@ -268,6 +301,7 @@ def handle(self, *args, **options):
                 wait_time=wait_time,
                 max_messages=max_messages,
                 verbosity=verbosity,
+                executors=executors,
             )
 
         if shutdown.is_set() and verbosity >= 1:
@@ -315,7 +349,7 @@ def _supervise(self, options):
         workers = options["workers"]
         verbosity = options["verbosity"]
 
-        self._check_database_supports_workers(workers)
+        self._check_database_supports_concurrency("--workers", workers)
 
         supervisor = WorkerSupervisor(
             worker_arguments(),
@@ -333,23 +367,25 @@ def _supervise(self, options):
         if exit_code:
             sys.exit(exit_code)
 
-    def _check_database_supports_workers(self, workers):
+    def _check_database_supports_concurrency(self, option, value):
         """
-        Refuse several workers on SQLite.
+        Refuse several workers, or several threads, on SQLite.
 
         Nothing runs twice there, since a task is claimed with a conditional
-        UPDATE, but SQLite has no row locks: two workers claiming at once
-        both hold the read lock and one is refused the write with "database
-        is locked", which the worker counts as a failed task. The README
-        says SQLite is for development and single-worker deployments; this
-        is where that stops being a note.
+        UPDATE, but SQLite has no row locks: two claims at once both hold the
+        read lock and one is refused the write with "database is locked",
+        which the worker counts as a failed task. Threads claim from one
+        thread when polling the database, but not for broker messages, and
+        their finishing writes collide the same way. The README says SQLite
+        is for development and single-worker deployments; this is where
+        that stops being a note.
         """
         alias = router.db_for_write(DatabaseTask)
         if connections[alias].vendor == "sqlite":
             raise CommandError(
-                f"--workers {workers} needs a database with row locks, and "
-                f"the {alias!r} database is SQLite. Run one worker, or use "
-                "PostgreSQL, MySQL or MariaDB."
+                f"{option} {value} needs a database with row locks, and the "
+                f"{alias!r} database is SQLite. Run one worker with one "
+                "thread, or use PostgreSQL, MySQL or MariaDB."
             )
 
     def _exit_code(self, tasks_processed, empty_exit_code, failed_exit_code):
@@ -407,19 +443,62 @@ def _process_tasks(
         wait_time=20.0,
         max_messages=1,
         verbosity=1,
+        executors=None,
     ):
+        """
+        The worker loop: find tasks and run them until there is reason to stop.
+
+        Without ``executors`` every task is run right here, in the thread
+        that found it. With them (``--threads``) this thread only fetches,
+        receives and claims, and hands each task to an idle executor; the
+        executors report back what they ran, and that is added up here.
+        """
         use_broker = source in (SOURCE_BROKER, SOURCE_BOTH)
         use_database = source in (SOURCE_DATABASE, SOURCE_BOTH)
         broker = backend.broker if use_broker else None
         tasks_processed = 0
+        exhausted = False
+
+        def in_flight():
+            return executors.busy if executors is not None else 0
+
+        def free_slots():
+            return executors.idle_count() if executors is not None else 1
+
+        def collect(wait=False):
+            """Add up what the executors have finished since last time."""
+            nonlocal tasks_processed
+            if executors is None:
+                return
+            outcomes = executors.collect(timeout=COLLECT_WAIT if wait else None)
+            tasks_processed += sum(1 for processed in outcomes if processed)
 
         def remaining():
-            """How many more tasks may be run, capped by --max-messages."""
+            """How many more tasks may be started right now."""
+            limit = max_messages
+            if executors is not None:
+                limit = min(limit, free_slots())
             if not max_tasks:
-                return max_messages
-            return min(max_messages, max_tasks - tasks_processed)
+                return limit
+            return min(limit, max_tasks - tasks_processed - in_flight())
+
+        def reached_max():
+            # Tasks handed to an executor count towards the cap, so no more
+            # than --max-tasks are ever started. One that turns out not to
+            # run (a broker message for a task that is gone) can leave the
+            # run short of the cap by that many.
+            return self._reached_max_tasks(
+                tasks_processed + in_flight(), max_tasks, verbosity
+            )
 
         while not shutdown.is_set():
+            collect()
+            if executors is not None and free_slots() == 0:
+                # Every executor is busy: wait for one to come back, not
+                # for the poll interval.
+                collect(wait=True)
+                continue
+
             worked = False
 
             if broker is not None:
@@ -435,10 +514,12 @@ def remaining():
                     wait,
                     remaining(),
                     verbosity,
+                    executors,
                 )
-                tasks_processed += count
+                if executors is None:
+                    tasks_processed += count
                 worked = count > 0
-                if self._reached_max_tasks(tasks_processed, max_tasks, verbosity):
+                if reached_max():
                     break
                 if shutdown.is_set():
                     break
@@ -447,19 +528,24 @@ def remaining():
                 task = fetch_task(queue_name=queue_name, backend_name=backend_name)
                 if task is not None:
                     worked = True
-                    if self._run_database_task(backend, task, worker_id, verbosity):
+                    if executors is None:
+                        counted = self._run_database_task(
+                            backend, task, worker_id, verbosity
+                        )
+                    else:
+                        counted = self._dispatch_database_task(
+                            executors, backend, task, verbosity
+                        )
+                    if counted:
                         tasks_processed += 1
-                        if self._reached_max_tasks(
-                            tasks_processed, max_tasks, verbosity
-                        ):
-                            break
+                    if reached_max():
+                        break
 
             if worked:
                 continue
 
             if not continuous:
-                if verbosity >= 1:
-                    self.stdout.write("No more tasks to process.")
+                exhausted = True
                 break
 
             if verbosity >= 2:
@@ -471,7 +557,7 @@ def remaining():
             if source == SOURCE_BOTH and wait_time > 0:
                 # The broker's own wait doubles as the idle interval, so a
                 # message wakes the worker up straight away.
-                tasks_processed += self._receive_and_run(
+                count = self._receive_and_run(
                     broker,
                     backend_name,
                     queue_name,
@@ -479,8 +565,11 @@ def remaining():
                     wait_time,
                     remaining(),
                     verbosity,
+                    executors,
                 )
-                if self._reached_max_tasks(tasks_processed, max_tasks, verbosity):
+                if executors is None:
+                    tasks_processed += count
+                if reached_max():
                     break
             elif source == SOURCE_BROKER and wait_time > 0:
                 # receive() has already waited.
@@ -490,6 +579,16 @@ def remaining():
                 # requested instead of waiting out the interval.
                 break
 
+        while in_flight():
+            # The tasks already handed out finish first. A shutdown that
+            # runs past --shutdown-timeout ends the process from the
+            # GracefulShutdown timer, as it does for a task run inline.
+            collect(wait=True)
+
+        if exhausted and verbosity >= 1:
+            # Reported after the executors are done, so it reads in order.
+            self.stdout.write("No more tasks to process.")
+
         return tasks_processed
 
     def _reached_max_tasks(self, tasks_processed, max_tasks, verbosity):
@@ -501,7 +600,7 @@ def _reached_max_tasks(self, tasks_processed, max_tasks, verbosity):
 
     def _run_database_task(self, backend, task, worker_id, verbosity):
         """
-        Run a task fetched straight from the database.
+        Run a task fetched straight from the database, here and now.
 
         Returns:
             True if the task was run here, False if another worker claimed
@@ -513,14 +612,7 @@ def _run_database_task(self, backend, task, worker_id, verbosity):
         try:
             result = backend.run_task(task, worker_id=worker_id)
         except Exception as e:
-            logger.exception(
-                "Worker could not run task: id=%s path=%s",
-                task.id,
-                task.task_path,
-                extra=task_log_fields(task, worker_id),
-            )
-            self.stdout.write(self.style.ERROR(f"  Error running task: {e}"))
-            self.tasks_failed += 1
+            self._report_run_error(task, worker_id, e)
             return True
 
         if result is None:
@@ -528,10 +620,78 @@ def _run_database_task(self, backend, task, worker_id, verbosity):
                 self.stdout.write("  Task is not ready to run; nothing to do")
             return False
 
+        self._record_result(result, verbosity)
+        return True
+
+    def _dispatch_database_task(self, executors, backend, task, verbosity):
+        """
+        Claim a task fetched from the database and hand it to an executor.
+
+        The claim happens here, in the fetching thread, so that the next
+        fetch does not return the same READY row; it is made with the
+        executor's worker id, which is what the task records. The executor
+        reports the run back through the loop's ``collect()``.
+
+        Returns:
+            True if the task is to be counted now (the claim itself failed,
+            which is a failed task that reached no executor), False if it
+            was handed out or was no longer READY.
+        """
+        index = executors.acquire()
+        thread_worker_id = executors.worker_ids[index]
+        if verbosity >= 1:
+            self.stdout.write(f"\nProcessing task: {task.id} ({task.task_path})")
+
+        try:
+            claimed = backend.claim_task(task, worker_id=thread_worker_id)
+        except Exception as e:
+            executors.release(index)
+            self._report_run_error(task, thread_worker_id, e)
+            return True
+
+        if not claimed:
+            executors.release(index)
+            if verbosity >= 1:
+                self.stdout.write("  Task is not ready to run; nothing to do")
+            return False
+
+        executors.submit(
+            index,
+            functools.partial(
+                self._run_claimed_task, backend, task, thread_worker_id, verbosity
+            ),
+        )
+        return False
+
+    def _run_claimed_task(self, backend, task, worker_id, verbosity):
+        """Run a task :meth:`_dispatch_database_task` claimed, in an executor."""
+        try:
+            result = backend.run_claimed_task(task, worker_id=worker_id)
+        except Exception as e:
+            self._report_run_error(task, worker_id, e)
+            return True
+        self._record_result(result, verbosity)
+        return True
+
+    def _report_run_error(self, task, worker_id, error):
+        """Record that this worker could not run ``task`` at all."""
+        logger.exception(
+            "Worker could not run task: id=%s path=%s",
+            task.id,
+            task.task_path,
+            extra=task_log_fields(task, worker_id),
+        )
+        self.stdout.write(self.style.ERROR(f"  Error running task: {error}"))
+        self._count_failure()
+
+    def _record_result(self, result, verbosity):
         if result.status != TaskResultStatus.SUCCESSFUL:
-            self.tasks_failed += 1
+            self._count_failure()
         self._report_result(result.status, verbosity)
-        return True
+
+    def _count_failure(self):
+        with self._failures_lock:
+            self.tasks_failed += 1
 
     def _receive_and_run(
         self,
@@ -542,8 +702,16 @@ def _receive_and_run(
         wait_seconds,
         max_messages,
         verbosity,
+        executors=None,
     ):
-        """Receive messages from the broker and run the tasks they name."""
+        """
+        Receive messages from the broker and run the tasks they name.
+
+        Returns:
+            Without executors, how many tasks were run. With them, how many
+            messages were handed out; the executors report the runs back
+            through the loop's ``collect()``.
+        """
         if max_messages < 1:
             return 0
 
@@ -567,11 +735,29 @@ def _receive_and_run(
             self.stdout.write(self.style.ERROR(f"\nError receiving from broker: {e}"))
             return 0
 
-        return sum(
-            1
-            for message in messages or []
-            if self._run_broker_message(broker, message, worker_id, verbosity)
-        )
+        messages = list(messages or [])
+        if executors is None:
+            return sum(
+                1
+                for message in messages
+                if self._run_broker_message(broker, message, worker_id, verbosity)
+            )
+
+        for message in messages:
+            # remaining() capped the receive by the idle executors, so one
+            # is free for each message.
+            index = executors.acquire()
+            executors.submit(
+                index,
+                functools.partial(
+                    self._run_broker_message,
+                    broker,
+                    message,
+                    executors.worker_ids[index],
+                    verbosity,
+                ),
+            )
+        return len(messages)
 
     def _run_broker_message(self, broker, message, worker_id, verbosity):
         """
@@ -602,7 +788,7 @@ def _run_broker_message(self, broker, message, worker_id, verbosity):
                 extra=self._broker_task_log_fields(message, worker_id),
             )
             self.stdout.write(self.style.ERROR(f"  Error running task: {e}"))
-            self.tasks_failed += 1
+            self._count_failure()
             self._nack(broker, message)
             return False
 
@@ -613,9 +799,7 @@ def _run_broker_message(self, broker, message, worker_id, verbosity):
                 self.stdout.write("  Task is not ready to run; nothing to do")
             return False
 
-        if result.status != TaskResultStatus.SUCCESSFUL:
-            self.tasks_failed += 1
-        self._report_result(result.status, verbosity)
+        self._record_result(result, verbosity)
         return True
 
     def _broker_task_log_fields(self, message, worker_id):
diff --git a/django_database_task/threads.py b/django_database_task/threads.py
new file mode 100644
index 0000000..ccb9d4c
--- /dev/null
+++ b/django_database_task/threads.py
@@ -0,0 +1,156 @@
+"""
+Executor threads for ``run_database_tasks --threads M``.
+
+The main thread of the worker stays the one that fetches and receives: it
+is the only thread polling the database, the only one on the broker's
+connection, and the one the signal handlers run in. What ``--threads``
+changes is that a task it found is run by one of M executor threads instead
+of inline, so M tasks are in progress at once while there is still one
+poller per process.
+
+Each executor has a worker id of its own, derived from the process's, so
+the ``worker_ids`` recorded on a task say which thread ran it. Django keeps
+database connections per thread, so each executor has a connection of its
+own too; after every task it closes the ones that outlived ``CONN_MAX_AGE``
+or broke, the way a request does, since a worker has no ``request_finished``
+to do it for it.
+
+Tasks run under ``--threads`` share one process. A task that is not
+thread-safe, calls ``os._exit()`` or crashes the interpreter takes the other
+M-1 with it; that is the trade the option makes, and the README says so.
+"""
+
+import logging
+import queue
+import threading
+
+from django.db import close_old_connections, connections
+
+logger = logging.getLogger("django_database_task")
+
+
+class ExecutorThreads:
+    """
+    M threads that run the jobs the main thread hands them.
+
+    Only the main thread calls :meth:`acquire`, :meth:`submit`,
+    :meth:`release` and :meth:`collect`; the executors put themselves back
+    on the idle list and their outcomes on the results queue. A job is a
+    callable taking no arguments; what it returns is handed back by
+    :meth:`collect`.
+
+    Args:
+        count: How many executors.
+        worker_id: The process's worker id; executor ``n`` is
+            ``f"{worker_id}-t{n}"``.
+    """
+
+    def __init__(self, count, worker_id):
+        if count < 1:
+            raise ValueError("count must be at least 1")
+        self.count = count
+        self.worker_ids = [f"{worker_id}-t{n}" for n in range(1, count + 1)]
+        self._inboxes = [queue.Queue(maxsize=1) for _ in range(count)]
+        self._idle = queue.Queue()
+        for index in range(count):
+            self._idle.put(index)
+        self._results = queue.Queue()
+        #: Jobs handed out and not yet collected. Main thread only.
+        self._busy = 0
+        self._threads = [
+            threading.Thread(
+                target=self._run,
+                args=(index,),
+                name=f"executor-{index + 1}",
+                daemon=True,
+            )
+            for index in range(count)
+        ]
+
+    def start(self):
+        for thread in self._threads:
+            thread.start()
+        return self
+
+    def stop(self):
+        """Let every executor finish its job and exit, and wait for them."""
+        for inbox in self._inboxes:
+            inbox.put(None)
+        for thread in self._threads:
+            thread.join()
+
+    @property
+    def busy(self):
+        """How many jobs are out: submitted and not yet collected."""
+        return self._busy
+
+    def idle_count(self):
+        """How many executors are free to take a job right now."""
+        return self._idle.qsize()
+
+    def acquire(self):
+        """
+        Take an idle executor, returning its index, or None if none is idle.
+
+        The index is held until :meth:`submit` hands it a job or
+        :meth:`release` gives it back, so the caller can do work in between
+        (claim a task with the executor's worker id) knowing the slot is its.
+        """
+        try:
+            return self._idle.get_nowait()
+        except queue.Empty:
+            return None
+
+    def release(self, index):
+        """Give back an executor taken with :meth:`acquire` and not used."""
+        self._idle.put(index)
+
+    def submit(self, index, job):
+        """Hand ``job`` to the executor taken with :meth:`acquire`."""
+        self._busy += 1
+        self._inboxes[index].put(job)
+
+    def collect(self, timeout=None):
+        """
+        Return the outcomes of the jobs that have finished since last time.
+
+        With ``timeout``, wait up to that many seconds for the first one when
+        none has finished yet; without it, return at once.
+        """
+        outcomes = []
+        try:
+            if timeout:
+                outcomes.append(self._results.get(timeout=timeout))
+            while True:
+                outcomes.append(self._results.get_nowait())
+        except queue.Empty:
+            pass
+        self._busy -= len(outcomes)
+        return outcomes
+
+    def _run(self, index):
+        inbox = self._inboxes[index]
+        try:
+            while True:
+                job = inbox.get()
+                if job is None:
+                    break
+                try:
+                    outcome = job()
+                except BaseException:
+                    # The jobs report their own failures; this is a last
+                    # resort so that a slot is never lost to a thread that
+                    # died, and so that SystemExit raised by task code ends
+                    # the task rather than the thread.
+                    logger.exception(
+                        "Executor thread %s failed",
+                        self.worker_ids[index],
+                        extra={"worker_id": self.worker_ids[index]},
+                    )
+                    outcome = None
+                finally:
+                    close_old_connections()
+                self._results.put(outcome)
+                self._idle.put(index)
+        finally:
+            connections.close_all()
diff --git a/tests/postgres/test_threads.py b/tests/postgres/test_threads.py
new file mode 100644
index 0000000..00e07b7
--- /dev/null
+++ b/tests/postgres/test_threads.py
@@ -0,0 +1,179 @@
+"""
+``--threads M`` at work, against PostgreSQL.
+
+The executor threads claim and write on connections of their own, so they
+need a database with row locks, which is why the command refuses several
+threads on SQLite; tests/test_commands.py covers that refusal, and
+tests/test_threads.py the pool on its own. Here the command runs with
+threads for real: tasks that can only finish together, one worker id per
+thread, the cap, the failure count, broker messages spread over the
+threads, and the shutdown that lets the running tasks finish.
+
+Run the suite with DJANGO_DATABASE_ENGINE=postgresql to include them.
+"""
+
+import sys
+import threading
+from io import StringIO
+from unittest.mock import patch
+
+import pytest
+from django.core.management import call_command
+from django.db import connections
+from django.tasks.base import TaskResultStatus
+
+from django_database_task.brokers import BrokerMessage
+from django_database_task.models import DatabaseTask
+from tests import tasks as test_tasks
+from tests.tasks import (
+    barrier_task,
+    failing_task,
+    shutdown_signal_task,
+    simple_task,
+    slow_task,
+)
+from tests.test_commands import FakePullBroker, make_backend, run_worker
+
+if connections["default"].vendor != "postgresql":
+    pytest.skip(
+        "needs a PostgreSQL database; run the suite with "
+        "DJANGO_DATABASE_ENGINE=postgresql",
+        allow_module_level=True,
+    )
+
+posix_signals = pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="os.kill() cannot deliver a signal to the test process on Windows",
+)
+
+# Real commits: the executor threads work on connections of their own, and
+# must see the rows the test enqueued and be free to write theirs while the
+# test's connection is not holding a transaction.
+pytestmark = pytest.mark.django_db(transaction=True)
+
+
+class TestThreads:
+    def test_tasks_run_at_the_same_time(self):
+        """Three tasks that each wait for the other two can only pass together."""
+        test_tasks.barrier = threading.Barrier(3, timeout=10)
+        results = [barrier_task.enqueue() for _ in range(3)]
+        out = StringIO()
+
+        call_command("run_database_tasks", threads=3, stdout=out)
+
+        statuses = {DatabaseTask.objects.get(id=r.id).status for r in results}
+        assert statuses == {TaskResultStatus.SUCCESSFUL}
+        output = out.getvalue()
+        assert "Threads: 3" in output
+        assert "Total tasks processed: 3" in output
+        assert "Tasks failed" not in output
+
+    def test_each_thread_records_its_own_worker_id(self):
+        test_tasks.barrier = threading.Barrier(2, timeout=10)
+        results = [barrier_task.enqueue() for _ in range(2)]
+
+        call_command("run_database_tasks", threads=2, stdout=StringIO())
+
+        worker_ids = sorted(
+            DatabaseTask.objects.get(id=r.id).worker_ids_json[0] for r in results
+        )
+        assert [w.rsplit("-", 1)[1] for w in worker_ids] == ["t1", "t2"]
+        assert worker_ids[0].rsplit("-", 1)[0] == worker_ids[1].rsplit("-", 1)[0]
+
+    def test_max_tasks_counts_the_tasks_handed_out(self):
+        for _ in range(5):
+            slow_task.enqueue(seconds=0.1)
+        out = StringIO()
+
+        call_command("run_database_tasks", threads=2, max_tasks=3, stdout=out)
+
+        assert "Reached max tasks limit: 3" in out.getvalue()
+        assert "Total tasks processed: 3" in out.getvalue()
+        assert DatabaseTask.objects.filter(status=TaskResultStatus.READY).count() == 2
+
+    def test_failures_are_counted_across_threads(self):
+        for _ in range(4):
+            failing_task.enqueue()
+        out = StringIO()
+
+        with pytest.raises(SystemExit) as exc_info:
+            call_command(
+                "run_database_tasks", threads=2, failed_exit_code=5, stdout=out
+            )
+
+        assert exc_info.value.code == 5
+        assert "Tasks failed: 4" in out.getvalue()
+        assert "Total tasks processed: 4" in out.getvalue()
+
+    def test_connections_are_closed_after_every_task(self):
+        for _ in range(3):
+            simple_task.enqueue(1, 2)
+
+        with patch("django_database_task.threads.close_old_connections") as close:
+            call_command("run_database_tasks", threads=2, stdout=StringIO())
+
+        assert close.call_count == 3
+
+    def test_output_ends_after_the_last_task(self):
+        slow_task.enqueue(seconds=0.3)
+        out = StringIO()
+
+        call_command("run_database_tasks", threads=2, stdout=out)
+
+        output = out.getvalue()
+        assert output.index("Task completed successfully") < output.index(
+            "No more tasks to process."
+        )
+
+    def test_broker_messages_are_spread_over_the_threads(self):
+        test_tasks.barrier = threading.Barrier(3, timeout=10)
+        results = [barrier_task.enqueue() for _ in range(3)]
+        broker = FakePullBroker(
+            batches=[[BrokerMessage(task_id=r.id) for r in results]]
+        )
+
+        output = run_worker(
+            make_backend(broker),
+            source="broker",
+            threads=3,
+            max_messages=10,
+            wait_time=0,
+        )
+
+        statuses = {DatabaseTask.objects.get(id=r.id).status for r in results}
+        assert statuses == {TaskResultStatus.SUCCESSFUL}
+        assert sorted(broker.acked) == sorted(r.id for r in results)
+        assert "Total tasks processed: 3" in output
+
+    def test_receive_asks_for_no_more_than_the_free_threads(self):
+        broker = FakePullBroker(batches=[])
+
+        run_worker(
+            make_backend(broker),
+            source="broker",
+            threads=2,
+            max_messages=10,
+            wait_time=0,
+        )
+
+        assert broker.received[0]["max_messages"] == 2
+
+    @posix_signals
+    def test_running_tasks_finish_before_shutdown(self):
+        """A signal lets the tasks in the threads finish and starts no more."""
+        first = slow_task.enqueue(seconds=0.5)
+        second = shutdown_signal_task.enqueue()
+        third = simple_task.enqueue(1, 2)
+        out = StringIO()
+
+        call_command("run_database_tasks", threads=2, stdout=out)
+
+        assert DatabaseTask.objects.get(id=first.id).status == (
+            TaskResultStatus.SUCCESSFUL
+        )
+        assert DatabaseTask.objects.get(id=second.id).status == (
+            TaskResultStatus.SUCCESSFUL
+        )
+        assert DatabaseTask.objects.get(id=third.id).status == TaskResultStatus.READY
+        assert "Shutdown complete" in out.getvalue()
+        assert "Total tasks processed: 2" in out.getvalue()
diff --git a/tests/tasks.py b/tests/tasks.py
index ef0e01c..2ed7cad 100644
--- a/tests/tasks.py
+++ b/tests/tasks.py
@@ -122,3 +122,17 @@ def shutdown_aware_task(iterations=10):
             break
         completed += 1
     return completed
+
+
+#: Set by a test to a threading.Barrier; barrier_task() waits on it, so the
+#: tasks only finish if as many of them are running at the same time.
+barrier = None
+
+
+@task
+def barrier_task():
+    """Task that waits for the other parties of ``barrier`` before returning."""
+    from . import tasks
+
+    tasks.barrier.wait()
+    return "met"
diff --git a/tests/test_commands.py b/tests/test_commands.py
index de934c9..5a4bd93 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -1306,7 +1306,7 @@ def run_workers(self, code=0, **options):
         out = StringIO()
         with (
             patch(f"{COMMAND_MODULE}.worker_arguments", return_value=stub_worker(code)),
-            patch.object(RunDatabaseTasks, "_check_database_supports_workers"),
+            patch.object(RunDatabaseTasks, "_check_database_supports_concurrency"),
         ):
             call_command("run_database_tasks", stdout=out, stderr=out, **options)
         return out.getvalue()
@@ -1379,3 +1379,37 @@ def stop():
         assert "; restarting." in output
         assert "Received SIGTERM: stopping" in output
         assert "Shutdown complete: every worker exited." in output
+
+
+@pytest.mark.django_db
+class TestRunDatabaseTasksThreads:
+    """
+    What the command checks for --threads before starting any thread.
+
+    The threads at work are tested against PostgreSQL in
+    tests/postgres/test_threads.py, and the pool on its own in
+    tests/test_threads.py.
+    """
+
+    def test_threads_must_be_positive(self):
+        with pytest.raises(CommandError, match="--threads must be at least 1"):
+            call_command("run_database_tasks", threads=0, stdout=StringIO())
+
+    def test_one_thread_is_the_worker_itself(self):
+        result = simple_task.enqueue(1, 2)
+        out = StringIO()
+
+        call_command("run_database_tasks", threads=1, stdout=out)
+
+        assert "Threads:" not in out.getvalue()
+        worker_ids = DatabaseTask.objects.get(id=result.id).worker_ids_json
+        assert len(worker_ids) == 1
+        assert "-t" not in worker_ids[0]
+
+    @pytest.mark.skipif(
+        connections["default"].vendor != "sqlite",
+        reason="the refusal is for SQLite",
+    )
+    def test_several_threads_are_refused_on_sqlite(self):
+        with pytest.raises(CommandError, match="--threads 2 needs a database"):
+            call_command("run_database_tasks", threads=2, stdout=StringIO())
diff --git a/tests/test_threads.py b/tests/test_threads.py
new file mode 100644
index 0000000..1b9e97c
--- /dev/null
+++ b/tests/test_threads.py
@@ -0,0 +1,102 @@
+"""Tests for the executor threads behind ``--threads``, without a database."""
+
+import threading
+import time
+from unittest.mock import patch
+
+import pytest
+
+from django_database_task.threads import ExecutorThreads
+
+
+@pytest.fixture
+def executors():
+    pool = ExecutorThreads(2, "host-abcd1234").start()
+    yield pool
+    pool.stop()
+
+
+def wait_for(condition, timeout=10):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if condition():
+            return
+        time.sleep(0.01)
+    pytest.fail("condition not met")
+
+
+class TestExecutorThreads:
+    def test_rejects_less_than_one(self):
+        with pytest.raises(ValueError):
+            ExecutorThreads(0, "w")
+
+    def test_worker_ids_are_derived_from_the_process_id(self, executors):
+        assert executors.worker_ids == ["host-abcd1234-t1", "host-abcd1234-t2"]
+
+    def test_jobs_run_and_report_back(self, executors):
+        ran = []
+        assert executors.idle_count() == 2
+
+        with patch("django_database_task.threads.close_old_connections") as close:
+            for value in ("a", "b"):
+                index = executors.acquire()
+                executors.submit(index, lambda value=value: ran.append(value) or value)
+            assert executors.busy == 2
+            assert executors.acquire() is None
+
+            outcomes = []
+            while executors.busy:
+                outcomes += executors.collect(timeout=5)
+
+        assert sorted(outcomes) == ["a", "b"]
+        assert sorted(ran) == ["a", "b"]
+        assert executors.idle_count() == 2
+        assert close.call_count == 2
+
+    def test_jobs_run_at_the_same_time(self, executors):
+        barrier = threading.Barrier(2, timeout=5)
+
+        def meet():
+            barrier.wait()
+            return "met"
+
+        for _ in range(2):
+            executors.submit(executors.acquire(), meet)
+
+        outcomes = []
+        while executors.busy:
+            outcomes += executors.collect(timeout=5)
+
+        assert outcomes == ["met", "met"]
+
+    def test_release_gives_the_slot_back(self, executors):
+        index = executors.acquire()
+        assert executors.idle_count() == 1
+        executors.release(index)
+        assert executors.idle_count() == 2
+        assert executors.busy == 0
+
+    def test_a_job_that_raises_does_not_lose_the_slot(self, executors, caplog):
+        def boom():
+            raise SystemExit(3)
+
+        executors.submit(executors.acquire(), boom)
+        outcomes = executors.collect(timeout=5)
+
+        assert outcomes == [None]
+        assert executors.idle_count() == 2
+        assert "Executor thread host-abcd1234-t" in caplog.text
+
+    def test_collect_without_timeout_does_not_wait(self, executors):
+        started = time.monotonic()
+        assert executors.collect() == []
+        assert time.monotonic() - started < 1
+
+    def test_stop_waits_for_the_running_job(self):
+        pool = ExecutorThreads(1, "w").start()
+        done = []
+        pool.submit(pool.acquire(), lambda: time.sleep(0.2) or done.append(True))
+
+        pool.stop()
+
+        assert done == [True]