diff --git a/queue_job/README.rst b/queue_job/README.rst
index b27a1675ad..eabb6bdba5 100644
--- a/queue_job/README.rst
+++ b/queue_job/README.rst
@@ -128,7 +128,10 @@ Configuration
   - Adjust environment variables (optional):
 
     - ``ODOO_QUEUE_JOB_CHANNELS=root:4`` or any other channels
-      configuration. The default is ``root:1``
+    - ``ODOO_QUEUE_JOB_MAX_CAPACITY=4``, max number of concurrent jobs
+      (not used if ``ODOO_QUEUE_JOB_CHANNELS`` is set)
+    - ``ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2``, max number of concurrent
+      jobs per DB (not used if ``ODOO_QUEUE_JOB_CHANNELS`` is set)
     - ``ODOO_QUEUE_JOB_PORT=8069``, default ``--http-port``
     - ``ODOO_QUEUE_JOB_SCHEME=https``, default ``http``
     - ``ODOO_QUEUE_JOB_HOST=load-balancer``, default
@@ -138,7 +141,8 @@ Configuration
     - Start Odoo with ``--load=web,queue_job`` and ``--workers`` greater
       than 1. [1]_
 
-- Using the Odoo configuration file:
+- Using the Odoo configuration file (set either ``channels``, either
+  ``max_capacity`` and ``db_max_capacity``)
 
 .. code:: ini
 
@@ -150,12 +154,17 @@ Configuration
    (...)
    [queue_job]
    channels = root:2
+   max_capacity = 8
+   db_max_capacity = 3
    scheme = https
    host = load-balancer
    port = 443
    http_auth_user = jobrunner
    http_auth_password = s3cr3t
 
+``db_max_capacity`` may be an integer or a pattern such as
+``prod_*:20,staging:2,*:5``
+
 - Confirm the runner is starting correctly by checking the odoo log
   file:
 
diff --git a/queue_job/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py
index e2561b0e74..2a1168707b 100644
--- a/queue_job/jobrunner/__init__.py
+++ b/queue_job/jobrunner/__init__.py
@@ -20,7 +20,7 @@
     queue_job_config = config.misc.get("queue_job", {})
 
 
-from .runner import QueueJobRunner, _channels
+from .runner import QueueJobRunner, _channels, _max_capacity
 
 _logger = logging.getLogger(__name__)
 
@@ -87,7 +87,12 @@ def signal_time_expired_handler(self, n, stack):
 
 
 def _is_runner_enabled():
-    return not _channels().strip().startswith("root:0")
+    channel_config = _channels()
+    if channel_config and channel_config.strip().startswith("root:0"):
+        return False
+    elif channel_config:
+        return True
+    return _max_capacity() != 0
 
 
 def _start_runner_thread(server_type):
@@ -100,7 +105,8 @@ def _start_runner_thread(server_type):
         else:
             _logger.info(
                 "jobrunner thread (in %s) NOT started, "
-                "because the root channel's capacity is set to 0",
+                "because the root channel's capacity or the max capacity "
+                "is set to 0",
                 server_type,
             )
 
diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py
index e3cb480420..4d7b441e26 100644
--- a/queue_job/jobrunner/channels.py
+++ b/queue_job/jobrunner/channels.py
@@ -3,6 +3,7 @@
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 import logging
 from collections import namedtuple
+from dataclasses import asdict, dataclass
 from functools import total_ordering
 from heapq import heappop, heappush
 from weakref import WeakValueDictionary
@@ -10,12 +11,24 @@
 from ..exception import ChannelNotFound
 from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES
 
+RELOAD_PAYLOAD = "reload"
 NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED)
 JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq")
 
 _logger = logging.getLogger(__name__)
 
 
+@dataclass
+class ChannelConfig:
+    """Configuration of a channel"""
+
+    name: str
+    capacity: int = 0
+    sequential: bool = False
+    throttle: int = 0
+    paused: bool = False
+
+
 class PriorityQueue:
     """A priority queue that supports removing arbitrary objects.
 
@@ -965,6 +978,11 @@ def simple_configure(self, config_string):
         for config in ChannelManager.parse_simple_config(config_string):
             self.get_channel_from_config(config)
 
+    def configure(self, configs):
+        """Configure the channel manager from list of :class:`ChannelConfig`"""
+        for config in configs:
+            self.get_channel_from_config(asdict(config))
+
     def get_channel_from_config(self, config):
         """Return a Channel object from a parsed configuration.
 
@@ -1115,3 +1133,8 @@ def get_jobs_to_run(self, now):
 
     def get_wakeup_time(self):
         return self._root_channel.get_wakeup_time()
+
+    @property
+    def running_count(self) -> int:
+        """Number of jobs currently running"""
+        return len(self._root_channel._running)
diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py
index 95e134ba44..d25adf1712 100644
--- a/queue_job/jobrunner/runner.py
+++ b/queue_job/jobrunner/runner.py
@@ -19,11 +19,13 @@
   anonymous ``/queue_job/runjob`` HTTP request.
 """
 
+import fnmatch
 import logging
 import os
 import selectors
 import threading
 import time
+from collections import deque
 from contextlib import closing, contextmanager
 
 import psycopg2
@@ -34,7 +36,7 @@
 from odoo.tools import config
 
 from . import queue_job_config
-from .channels import ENQUEUED, NOT_DONE, ChannelManager
+from .channels import ENQUEUED, NOT_DONE, RELOAD_PAYLOAD, ChannelConfig, ChannelManager
 
 SELECT_TIMEOUT = 60
 ERROR_RECOVERY_DELAY = 5
@@ -57,14 +59,92 @@ class MasterElectionLost(Exception):
 # so we check it in addition to the environment variables.
 
 
-def _channels():
+def _max_capacity() -> int:
+    """Maximum number of jobs running at the same time across all databases
+
+    It comes from the ``ODOO_QUEUE_JOB_MAX_CAPACITY`` environment
+    variable, then ``max_capacity`` in the ``[queue_job]`` section of the
+    configuration file.
+
+    If none is configured, the max capacity is 0.
+    """
+    value = os.environ.get("ODOO_QUEUE_JOB_MAX_CAPACITY") or queue_job_config.get(
+        "max_capacity"
+    )
+    if value:
+        return int(value)
+    return 0
+
+
+def _db_max_capacity() -> str:
     return (
-        os.environ.get("ODOO_QUEUE_JOB_CHANNELS")
-        or queue_job_config.get("channels")
-        or "root:1"
+        os.environ.get("ODOO_QUEUE_JOB_DB_MAX_CAPACITY")
+        or queue_job_config.get("db_max_capacity")
+        or ""
     )
 
 
+def parse_db_max_capacity(spec):
+    """Parse a per-database max capacity configuration string
+
+    The string is a comma-separated list of ``pattern:capacity`` items, where
+    ``pattern`` matches database names with fnmatch wildcards.
+
+    The first matching pattern wins, so specific patterns must be first in the
+    string.
+
+    A single integer is applied to all databases, as a shorthand for
+    ``*:capacity``.
+
+    >>> parse_db_max_capacity('prod_*:20,staging:2,*:5')
+    [('prod_*', 20), ('staging', 2), ('*', 5)]
+    >>> parse_db_max_capacity('8')
+    [('*', 8)]
+    >>> parse_db_max_capacity('')
+    []
+    >>> parse_db_max_capacity(None)
+    []
+    """
+    rules = []
+    if not spec:
+        return rules
+    for item in spec.replace("\n", ",").split(","):
+        item = item.strip()
+        if not item:
+            continue
+        pattern, sep, capacity = item.rpartition(":")
+        if not sep:
+            pattern = "*"
+        try:
+            rules.append((pattern.strip(), int(capacity)))
+        except ValueError as ex:
+            raise ValueError(f"Invalid db max capacity {spec}: {capacity}") from ex
+    return rules
+
+
+def db_max_capacity_for(db_name, rules, default=None):
+    """Max capacity of a database, first match wins
+
+    >>> rules = parse_db_max_capacity('prod_*:20,staging:2,*:5')
+    >>> db_max_capacity_for('prod_foo', rules)
+    20
+    >>> db_max_capacity_for('staging', rules)
+    2
+    >>> db_max_capacity_for('dev', rules)
+    5
+    >>> db_max_capacity_for('dev', [], default=7)
+    7
+    """
+    for pattern, capacity in rules:
+        if fnmatch.fnmatch(db_name, pattern):
+            return capacity
+    return default
+
+
+def _channels():
+    return os.environ.get("ODOO_QUEUE_JOB_CHANNELS") or queue_job_config.get("channels")
+
+
 def _odoo_now():
     # important: this must return the same as postgresql
     # EXTRACT(EPOCH FROM TIMESTAMP dt)
@@ -123,9 +203,11 @@ def __init__(self, db_name):
         try:
             self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
             self.has_queue_job = self._has_queue_job()
+            self.has_channel_config_columns = False
             if self.has_queue_job:
                 self._acquire_master_lock()
                 self._initialize()
+                self.has_channel_config_columns = self._has_channel_config_columns()
         except BaseException:
             self.close()
             raise
@@ -182,6 +264,45 @@ def _initialize(self):
         with closing(self.conn.cursor()) as cr:
             cr.execute("LISTEN queue_job")
 
+    def _has_channel_config_columns(self):
+        with closing(self.conn.cursor()) as cr:
+            cr.execute(
+                """
+                SELECT count(*)
+                FROM information_schema.columns
+                WHERE table_name = %s
+                AND column_name IN ('capacity', 'sequential', 'throttle', 'paused')
+                """,
+                ("queue_job_channel",),
+            )
+            return cr.fetchone()[0] == 4
+
+    def load_channels_config(self):
+        """Return the channels configuration stored in the database"""
+        if not self.has_channel_config_columns:
+            return None
+        with closing(self.conn.cursor()) as cr:
+            cr.execute(
+                "SELECT complete_name, "
+                "COALESCE(capacity, 0), "
+                "COALESCE(sequential, false), "
+                "COALESCE(throttle, 0), "
+                "COALESCE(paused, false) "
+                "FROM queue_job_channel "
+            )
+            rows = cr.fetchall()
+            configs = [
+                ChannelConfig(
+                    name=name,
+                    capacity=capacity,
+                    sequential=sequential,
+                    throttle=throttle,
+                    paused=paused,
+                )
+                for name, capacity, sequential, throttle, paused in rows
+            ]
+            return configs
+
     @contextmanager
     def select_jobs(self, where, args):
         # pylint: disable=sql-injection
@@ -321,16 +442,40 @@ def __init__(
         user=None,
         password=None,
         channel_config_string=None,
+        max_capacity=None,
+        db_max_capacity=None,
     ):
         self.scheme = scheme
         self.host = host
         self.port = port
         self.user = user
         self.password = password
-        self.channel_manager = ChannelManager()
+
         if channel_config_string is None:
             channel_config_string = _channels()
-        self.channel_manager.simple_configure(channel_config_string)
+
+        self._server_side_channel_manager = None
+        if channel_config_string:
+            channel_manager = ChannelManager()
+            channel_manager.simple_configure(channel_config_string)
+            self._server_side_channel_manager = channel_manager
+            # max_capacity is always equal to the root channel in server-side
+            # configuration
+            max_capacity = channel_manager.get_channel_by_name("root").capacity
+
+        if max_capacity is None:
+            max_capacity = _max_capacity()
+        self.max_capacity = max_capacity
+
+        if db_max_capacity is None:
+            db_max_capacity = _db_max_capacity()
+        self.db_max_capacity_rules = parse_db_max_capacity(db_max_capacity)
+
+        self._channel_manager_by_db = {}
+        self._channel_managers = []
+
+        self._round_robin_offset = 0
+
         self.db_by_name = {}
         self._stop = False
         self._stop_pipe = os.pipe()
@@ -387,11 +532,78 @@ def close_databases(self, remove_jobs=True):
         for db_name, db in self.db_by_name.items():
             try:
                 if remove_jobs:
-                    self.channel_manager.remove_db(db_name)
+                    self._channel_manager_by_db[db_name].remove_db(db_name)
                 db.close()
             except Exception:
                 _logger.warning("error closing database %s", db_name, exc_info=True)
         self.db_by_name = {}
+        self._channel_manager_by_db = {}
+        self._channel_managers = []
+
+    @staticmethod
+    def _unique_channel_managers(channel_managers):
+        seen = set()
+        result = []
+        for channel_manager in channel_managers:
+            if id(channel_manager) not in seen:
+                seen.add(id(channel_manager))
+                result.append(channel_manager)
+        return result
+
+    def _build_channel_manager(self, db):
+        """Build and configure the channel manager of a database"""
+        db_max = db_max_capacity_for(
+            db.db_name, self.db_max_capacity_rules, default=self.max_capacity
+        )
+        channel_manager = ChannelManager()
+
+        channels_config = db.load_channels_config()
+        if channels_config is None:
+            # database not updated to the proper schema,
+            # no job execution until it is properly upgraded
+            _logger.error(
+                "database %s schema is outdated, -u queue_job required", db.db_name
+            )
+            channel_manager.configure([ChannelConfig(name="root", capacity=0)])
+            return channel_manager
+
+        root_config = next(
+            (config for config in channels_config if config.name == "root"), None
+        )
+        if root_config is None:
+            root_config = ChannelConfig("root")
+            channels_config.insert(0, root_config)
+
+        if not db_max:
+            # if a database is set at 0, it does not run any jobs, pause it
+            root_config.paused = True
+        elif not root_config.capacity:
+            root_config.capacity = db_max
+        else:
+            root_config.capacity = min(root_config.capacity, db_max)
+        channel_manager.configure(channels_config)
+        return channel_manager
+
+    def _reconfigure_db(self, db_name):
+        """Rebuild the channel manager for a database and reload its jobs"""
+        db = self.db_by_name.get(db_name)
+        if db is None:
+            return
+        if self._server_side_channel_manager:
+            channel_manager = self._server_side_channel_manager
+        else:
+            channel_manager = self._build_channel_manager(db)
+        with db.select_jobs("state in %s", (NOT_DONE,)) as cr:
+            for job_data in cr:
+                channel_manager.notify(db_name, *job_data)
+        self._register_channel_manager(db_name, channel_manager)
+        _logger.info("channels configuration loaded for db %s", db_name)
+
+    def _register_channel_manager(self, db_name, channel_manager):
+        self._channel_manager_by_db[db_name] = channel_manager
+        self._channel_managers = self._unique_channel_managers(
+            self._channel_manager_by_db.values()
+        )
 
     def initialize_databases(self):
         for db_name in sorted(self.get_db_names()):
@@ -399,9 +611,7 @@ def initialize_databases(self):
             db = Database(db_name)
             if db.has_queue_job:
                 self.db_by_name[db_name] = db
-                with db.select_jobs("state in %s", (NOT_DONE,)) as cr:
-                    for job_data in cr:
-                        self.channel_manager.notify(db_name, *job_data)
+                self._reconfigure_db(db_name)
                 _logger.info("queue job runner ready for db %s", db_name)
             else:
                 db.close()
@@ -411,24 +621,71 @@ def requeue_dead_jobs(self):
             if db.has_queue_job:
                 db.requeue_dead_jobs()
 
-    def run_jobs(self):
-        now = _odoo_now()
-        for job in self.channel_manager.get_jobs_to_run(now):
-            if self._stop:
-                break
-            _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name)
-            self.db_by_name[job.db_name].set_job_enqueued(job.uuid)
-            _async_http_get(
-                self.scheme,
-                self.host,
-                self.port,
-                self.user,
-                self.password,
-                job.db_name,
-                job.uuid,
+    def _all_running_count(self) -> int:
+        return sum(
+            channel_manager.running_count for channel_manager in self._channel_managers
+        )
+
+    def _dispatch_job(self, job):
+        _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name)
+        self.db_by_name[job.db_name].set_job_enqueued(job.uuid)
+        _async_http_get(
+            self.scheme,
+            self.host,
+            self.port,
+            self.user,
+            self.password,
+            job.db_name,
+            job.uuid,
+        )
+
+    def _round_robin_jobs(self, channel_managers, now):
+        managers_count = len(channel_managers)
+        # Ensure the channel managers are ordered in way that all have
+        # equal chances to enqueue jobs. The manager that was last last
+        # time will be first the next time and so on.
+        job_generators = deque(
+            (
+                # store the actual position of the channel manager in the channel
+                # managers list
+                index % managers_count,
+                channel_managers[index % managers_count].get_jobs_to_run(now),
             )
+            for index in range(
+                self._round_robin_offset, self._round_robin_offset + managers_count
+            )
+        )
+        while job_generators:
+            index, job_generator = job_generators.popleft()
+            job = next(job_generator, None)
+            if job is None:
+                # generator exhausted for this tick, remove from the deque
+                # it will come back next time
+                continue
+            job_generators.append((index, job_generator))
+            self._round_robin_offset = index + 1
+            yield job
+
+    def run_jobs(self):
+        channel_managers = self._channel_managers
+        if not channel_managers:
+            return
+
+        jobs = self._round_robin_jobs(channel_managers, _odoo_now())
+        while not self._stop:
+            if self.max_capacity and self._all_running_count() >= self.max_capacity:
+                _logger.debug(
+                    "max capacity of %s reached, waiting for capacity",
+                    self.max_capacity,
+                )
+                return
+            job = next(jobs, None)
+            if job is None:
+                return
+            self._dispatch_job(job)
 
     def process_notifications(self):
+        reload_db_names = set()
         for db in self.db_by_name.values():
             if not db.conn.notifies:
                 # If there are no activity in the queue_job table it seems that
@@ -440,13 +697,29 @@ def process_notifications(self):
                 if self._stop:
                     break
                 notification = db.conn.notifies.pop()
-                uuid = notification.payload
+                payload = notification.payload
+                if payload == RELOAD_PAYLOAD and not self._server_side_channel_manager:
+                    reload_db_names.add(db.db_name)
+                    continue
+
+                uuid = payload
+                channel_manager = self._channel_manager_by_db[db.db_name]
                 with db.select_jobs("uuid = %s", (uuid,)) as cr:
                     job_datas = cr.fetchone()
                     if job_datas:
-                        self.channel_manager.notify(db.db_name, *job_datas)
+                        channel_manager.notify(db.db_name, *job_datas)
                     else:
-                        self.channel_manager.remove_job(uuid)
+                        channel_manager.remove_job(uuid)
+
+        for db_name in reload_db_names:
+            self._reconfigure_db(db_name)
+
+    def next_wakeup_time(self):
+        wakeup_times = [
+            channel_manager.get_wakeup_time()
+            for channel_manager in self._channel_managers
+        ]
+        return min(wakeup_times, default=0)
 
     def wait_notification(self):
         for db in self.db_by_name.values():
@@ -458,7 +731,7 @@ def wait_notification(self):
         conns = [db.conn for db in self.db_by_name.values()]
         conns.append(self._stop_pipe[0])
         # look if the channels specify a wakeup time
-        wakeup_time = self.channel_manager.get_wakeup_time()
+        wakeup_time = self.next_wakeup_time()
         if not wakeup_time:
             # this could very well be no timeout at all, because
             # any activity in the job queue will wake us up, but
diff --git a/queue_job/models/queue_job_channel.py b/queue_job/models/queue_job_channel.py
index 4aabb0188c..47e5bd62a8 100644
--- a/queue_job/models/queue_job_channel.py
+++ b/queue_job/models/queue_job_channel.py
@@ -4,12 +4,19 @@
 
 from odoo import _, api, exceptions, fields, models
 
+from ..jobrunner.channels import RELOAD_PAYLOAD
+
 
 class QueueJobChannel(models.Model):
     _name = "queue.job.channel"
     _description = "Job Channels"
     _rec_name = "complete_name"
 
+    # fields that trigger a reload of the jobrunner for this database when changed
+    _JOBRUNNER_CONFIG_FIELDS = frozenset(
+        ("capacity", "sequential", "throttle", "paused", "name", "parent_id")
+    )
+
     name = fields.Char()
     complete_name = fields.Char(
         compute="_compute_complete_name", store=True, readonly=True, recursive=True
@@ -25,11 +32,44 @@ class QueueJobChannel(models.Model):
     removal_interval = fields.Integer(
         default=lambda self: self.env["queue.job"]._removal_interval, required=True
     )
+    capacity = fields.Integer(
+        help="Maximum number of jobs running at the same time in this channel. "
+        "0 means no limit, but they are still limited by the capacity of the parent "
+        "channel. On the root channel, 0 is limited by the global server-side "
+        "configuration."
+    )
+    sequential = fields.Boolean(
+        help="Jobs are executed one after the other and failed jobs block the channel. "
+        "Requires a capacity of 1."
+    )
+    throttle = fields.Integer(
+        help="Minimum delay in seconds between the start of two jobs in this channel."
+    )
+    paused = fields.Boolean(
+        help="A paused channel (an its sub-channels) do not execute any jobs until "
+        "resumed."
+    )
 
     _sql_constraints = [
         ("name_uniq", "unique(complete_name)", "Channel complete name must be unique")
     ]
 
+    @api.constrains("capacity", "sequential", "throttle")
+    def _check_jobrunner_configuration(self):
+        for record in self:
+            if record.capacity < 0:
+                raise exceptions.ValidationError(
+                    self.env._("The capacity of a channel cannot be negative.")
+                )
+            if record.throttle < 0:
+                raise exceptions.ValidationError(
+                    self.env._("The throttle of a channel cannot be negative.")
+                )
+            if record.sequential and record.capacity != 1:
+                raise exceptions.ValidationError(
+                    self.env._("A sequential channel must have a capacity of 1.")
+                )
+
     @api.depends("name", "parent_id.complete_name")
     def _compute_complete_name(self):
         for record in self:
@@ -70,6 +110,7 @@ def create(self, vals_list):
                 new_vals_list.append(vals)
             vals_list = new_vals_list
         records |= super().create(vals_list)
+        records._notify_channel_config_changed()
         return records
 
     def write(self, values):
@@ -80,10 +121,19 @@ def write(self, values):
                 and ("name" in values or "parent_id" in values)
             ):
                 raise exceptions.UserError(_("Cannot change the root channel"))
-        return super().write(values)
+        res = super().write(values)
+        if self._JOBRUNNER_CONFIG_FIELDS.intersection(values):
+            self._notify_channel_config_changed()
+        return res
 
     def unlink(self):
         for channel in self:
             if channel.name == "root":
                 raise exceptions.UserError(_("Cannot remove the root channel"))
-        return super().unlink()
+        res = super().unlink()
+        self._notify_channel_config_changed()
+        return res
+
+    def _notify_channel_config_changed(self):
+        """Notify the jobrunner to reload its configuration"""
+        self.env.cr.execute("SELECT pg_notify('queue_job', %s)", (RELOAD_PAYLOAD,))
diff --git a/queue_job/readme/CONFIGURE.md b/queue_job/readme/CONFIGURE.md
index 7239106218..b9256dc52a 100644
--- a/queue_job/readme/CONFIGURE.md
+++ b/queue_job/readme/CONFIGURE.md
@@ -1,7 +1,8 @@
 - Using environment variables and command line:
   - Adjust environment variables (optional):
     - `ODOO_QUEUE_JOB_CHANNELS=root:4` or any other channels
-      configuration. The default is `root:1`
+    - `ODOO_QUEUE_JOB_MAX_CAPACITY=4`, max number of concurrent jobs (not used if `ODOO_QUEUE_JOB_CHANNELS` is set)
+    - `ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2`, max number of concurrent jobs per DB (not used if `ODOO_QUEUE_JOB_CHANNELS` is set)
     - `ODOO_QUEUE_JOB_PORT=8069`, default `--http-port`
     - `ODOO_QUEUE_JOB_SCHEME=https`, default `http`
     - `ODOO_QUEUE_JOB_HOST=load-balancer`, default `--http-interface`
@@ -10,7 +11,7 @@
     - `ODOO_QUEUE_JOB_HTTP_AUTH_PASSWORD=s3cr3t`, default empty
     - Start Odoo with `--load=web,queue_job` and `--workers` greater than
       1.[^1]
-- Using the Odoo configuration file:
+- Using the Odoo configuration file (set either `channels`, either `max_capacity` and `db_max_capacity`)
 
 ``` ini
 [options]
@@ -21,6 +22,8 @@ server_wide_modules = web,queue_job
 (...)
 [queue_job]
 channels = root:2
+max_capacity = 8
+db_max_capacity = 3
 scheme = https
 host = load-balancer
 port = 443
@@ -28,10 +31,12 @@ http_auth_user = jobrunner
 http_auth_password = s3cr3t
 ```
 
+`db_max_capacity` may be an integer or a pattern such as `prod_*:20,staging:2,*:5`
+
 - Confirm the runner is starting correctly by checking the odoo log
   file:
 
-``` 
+```
 ...INFO...queue_job.jobrunner.runner: starting
 ...INFO...queue_job.jobrunner.runner: initializing database connections
 ...INFO...queue_job.jobrunner.runner: queue job runner ready for db <dbname>
diff --git a/queue_job/static/description/index.html b/queue_job/static/description/index.html
index c5920c7ac7..392f4eba19 100644
--- a/queue_job/static/description/index.html
+++ b/queue_job/static/description/index.html
@@ -3,7 +3,7 @@
 <head>
 <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
 <meta name="generator" content="Docutils: https://docutils.sourceforge.io/" />
-<title>README.rst</title>
+<title>Job Queue</title>
 <style type="text/css">
 
 /*
@@ -487,8 +487,11 @@ <h2><a class="toc-backref" href="#toc-entry-3">Configuration</a></h2>
 <ul class="simple">
 <li>Using environment variables and command line:<ul>
 <li>Adjust environment variables (optional):<ul>
-<li><tt class="docutils literal">ODOO_QUEUE_JOB_CHANNELS=root:4</tt> or any other channels
-configuration. The default is <tt class="docutils literal">root:1</tt></li>
+<li><tt class="docutils literal">ODOO_QUEUE_JOB_CHANNELS=root:4</tt> or any other channels</li>
+<li><tt class="docutils literal">ODOO_QUEUE_JOB_MAX_CAPACITY=4</tt>, max number of concurrent jobs
+(not used if <tt class="docutils literal">ODOO_QUEUE_JOB_CHANNELS</tt> is set)</li>
+<li><tt class="docutils literal">ODOO_QUEUE_JOB_DB_MAX_CAPACITY=2</tt>, max number of concurrent
+jobs per DB (not used if <tt class="docutils literal">ODOO_QUEUE_JOB_CHANNELS</tt> is set)</li>
 <li><tt class="docutils literal">ODOO_QUEUE_JOB_PORT=8069</tt>, default <tt class="docutils literal"><span class="pre">--http-port</span></tt></li>
 <li><tt class="docutils literal">ODOO_QUEUE_JOB_SCHEME=https</tt>, default <tt class="docutils literal">http</tt></li>
 <li><tt class="docutils literal"><span class="pre">ODOO_QUEUE_JOB_HOST=load-balancer</span></tt>, default
@@ -501,7 +504,8 @@ <h2><a class="toc-backref" href="#toc-entry-3">Configuration</a></h2>
 </li>
 </ul>
 </li>
-<li>Using the Odoo configuration file:</li>
+<li>Using the Odoo configuration file (set either <tt class="docutils literal">channels</tt>, either
+<tt class="docutils literal">max_capacity</tt> and <tt class="docutils literal">db_max_capacity</tt>)</li>
 </ul>
 <pre class="code ini literal-block">
 <span class="k">[options]</span><span class="w">
@@ -512,12 +516,16 @@ <h2><a class="toc-backref" href="#toc-entry-3">Configuration</a></h2>
 </span><span class="na">(...)</span><span class="w">
 </span><span class="k">[queue_job]</span><span class="w">
 </span><span class="na">channels</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">root:2</span><span class="w">
+</span><span class="na">max_capacity</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">8</span><span class="w">
+</span><span class="na">db_max_capacity</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">3</span><span class="w">
 </span><span class="na">scheme</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">https</span><span class="w">
 </span><span class="na">host</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">load-balancer</span><span class="w">
 </span><span class="na">port</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">443</span><span class="w">
 </span><span class="na">http_auth_user</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">jobrunner</span><span class="w">
 </span><span class="na">http_auth_password</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">s3cr3t</span>
 </pre>
+<p><tt class="docutils literal">db_max_capacity</tt> may be an integer or a pattern such as
+<tt class="docutils literal"><span class="pre">prod_*:20,staging:2,*:5</span></tt></p>
 <ul class="simple">
 <li>Confirm the runner is starting correctly by checking the odoo log
 file:</li>
diff --git a/queue_job/tests/test_model_job_channel.py b/queue_job/tests/test_model_job_channel.py
index 20ebbc0bfe..1277071ef4 100644
--- a/queue_job/tests/test_model_job_channel.py
+++ b/queue_job/tests/test_model_job_channel.py
@@ -1,9 +1,12 @@
 # copyright 2018 Camptocamp
 # license lgpl-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 
+from unittest import mock
+
 from psycopg2 import IntegrityError
 
 import odoo
+from odoo import exceptions
 from odoo.tests import common
 
 
@@ -57,3 +60,83 @@ def test_channel_display_name(self):
             {"name": "test", "parent_id": self.root_channel.id}
         )
         self.assertEqual(channel.display_name, channel.complete_name)
+
+    def test_capacity_should_not_be_negative(self):
+        with self.assertRaisesRegex(
+            exceptions.ValidationError,
+            "The capacity of a channel cannot be negative.",
+        ):
+            self.Channel.create(
+                {
+                    "name": "test_capacity",
+                    "parent_id": self.root_channel.id,
+                    "capacity": -1,
+                }
+            )
+
+    def test_throttle_should_not_be_negative(self):
+        with self.assertRaisesRegex(
+            exceptions.ValidationError,
+            "The throttle of a channel cannot be negative.",
+        ):
+            self.Channel.create(
+                {
+                    "name": "test_throttle",
+                    "parent_id": self.root_channel.id,
+                    "throttle": -1,
+                }
+            )
+
+    def test_sequential_should_have_capacity_one(self):
+        with self.assertRaisesRegex(
+            exceptions.ValidationError,
+            "A sequential channel must have a capacity of 1.",
+        ):
+            self.Channel.create(
+                {
+                    "name": "test_sequential",
+                    "parent_id": self.root_channel.id,
+                    "sequential": True,
+                    "capacity": 2,
+                }
+            )
+
+    def _patch_notify(self):
+        return mock.patch.object(
+            type(self.Channel), "_notify_channel_config_changed", autospec=True
+        )
+
+    def test_notify_create_channel(self):
+        with self._patch_notify() as notify:
+            self.Channel.create(
+                {
+                    "name": "create_notify",
+                    "parent_id": self.root_channel.id,
+                    "capacity": 2,
+                }
+            )
+        notify.assert_called_once()
+
+    def test_notify_write_jobrunner_config(self):
+        channel = self.Channel.create(
+            {"name": "write_notify", "parent_id": self.root_channel.id}
+        )
+        with self._patch_notify() as notify:
+            channel.capacity = 3
+        notify.assert_called_once()
+
+        with self._patch_notify() as notify:
+            channel.paused = True
+        notify.assert_called_once()
+
+        with self._patch_notify() as notify:
+            channel.removal_interval = 60
+        notify.assert_not_called()
+
+    def test_notify_unlink_channel(self):
+        channel = self.Channel.create(
+            {"name": "unlink_notify", "parent_id": self.root_channel.id}
+        )
+        with self._patch_notify() as notify:
+            channel.unlink()
+        notify.assert_called_once()
diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py
index 1adfa790dd..3816cea8db 100644
--- a/queue_job/tests/test_runner_runner.py
+++ b/queue_job/tests/test_runner_runner.py
@@ -2,12 +2,15 @@
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 import doctest
 import os
+from unittest import mock
 
 from odoo.tests import BaseCase, tagged
 
+from odoo.addons.queue_job.jobrunner import runner
+
 # pylint: disable=odoo-addons-relative-import
 # we are testing, we want to test as we were an external consumer of the API
-from odoo.addons.queue_job.jobrunner import runner
+from odoo.addons.queue_job.jobrunner.channels import ChannelConfig, ChannelManager
 
 
 @tagged("doctest")
@@ -20,6 +23,35 @@ def test_doctest(self):
 
 @tagged("-at_install", "post_install")
 class TestRunner(BaseCase):
+    def setUp(self):
+        super().setUp()
+        # ensure there is no collision with actual environment variables/config
+        env_patcher = mock.patch.dict(
+            "os.environ",
+            {
+                "ODOO_QUEUE_JOB_MAX_CAPACITY": "",
+                "ODOO_QUEUE_JOB_DB_MAX_CAPACITY": "",
+                "ODOO_QUEUE_JOB_CHANNELS": "",
+            },
+        )
+        env_patcher.start()
+        self.addCleanup(env_patcher.stop)
+
+        conf_patcher = mock.patch.object(runner, "queue_job_config", {})
+        conf_patcher.start()
+        self.addCleanup(conf_patcher.stop)
+
+    def _new_channel_manager(self, jobrunner, db_name, channel_config, pending_jobs=0):
+        channel_manager = ChannelManager()
+        channel_manager.configure(channel_config)
+        jobrunner._register_channel_manager(db_name, channel_manager)
+        jobrunner.db_by_name[db_name] = mock.MagicMock(name=f"Database({db_name})")
+        for number in range(pending_jobs):
+            channel_manager.notify(
+                db_name, "root", f"{db_name}-{number}", number, 0, 10, None, "pending"
+            )
+        return channel_manager
+
     @classmethod
     def _is_open_file_descriptor(cls, fd):
         try:
@@ -61,3 +93,69 @@ def test_runner_file_closed_write_descriptor(self):
 
         self.assertFalse(self._is_open_file_descriptor(read_fd))
         self.assertFalse(self._is_open_file_descriptor(write_fd))
+
+    def test_max_capacity_from_env(self):
+        with (
+            mock.patch.dict(
+                os.environ, {"ODOO_QUEUE_JOB_MAX_CAPACITY": "5"}, clear=True
+            ),
+            mock.patch.object(runner, "queue_job_config", {}),
+        ):
+            self.assertEqual(runner._max_capacity(), 5)
+
+    def test_max_capacity_from_odoo_config(self):
+        with (
+            mock.patch.dict(os.environ, {}, clear=True),
+            mock.patch.object(runner, "queue_job_config", {"max_capacity": "3"}),
+        ):
+            self.assertEqual(runner._max_capacity(), 3)
+
+    def test_max_capacity_env_priority_over_odoo_config(self):
+        with (
+            mock.patch.dict(
+                os.environ, {"ODOO_QUEUE_JOB_MAX_CAPACITY": "5"}, clear=True
+            ),
+            mock.patch.object(runner, "queue_job_config", {"max_capacity": "3"}),
+        ):
+            self.assertEqual(runner._max_capacity(), 5)
+
+    def test_max_capacity_default(self):
+        with (
+            mock.patch.dict(os.environ, {}, clear=True),
+            mock.patch.object(runner, "queue_job_config", {}),
+        ):
+            self.assertEqual(runner._max_capacity(), 0)
+
+    def test_max_capacity_equals_to_channel_config_string(self):
+        jobrunner = runner.QueueJobRunner(channel_config_string="root:3")
+        self.assertEqual(jobrunner.max_capacity, 3)
+
+    def test_max_capacity_server_side_has_priority(self):
+        jobrunner = runner.QueueJobRunner(
+            channel_config_string="root:3", max_capacity=1
+        )
+        self.assertEqual(jobrunner.max_capacity, 3)
+
+    def test_max_capacity_channel_config_no_root(self):
+        jobrunner = runner.QueueJobRunner(channel_config_string="sub:3")
+        self.assertEqual(jobrunner.max_capacity, 1)
+
+    def test_no_job_dispatched_when_no_database_channel_manager(self):
+        jobrunner = runner.QueueJobRunner(max_capacity=3)
+        with mock.patch.object(jobrunner, "_dispatch_job") as dispatch:
+            jobrunner.run_jobs()
+        dispatch.assert_not_called()
+
+    def test_global_capacity_across_databases(self):
+        jobrunner = runner.QueueJobRunner(max_capacity=3)
+        manager_a = self._new_channel_manager(
+            jobrunner, "db_a", [ChannelConfig("root", 5)], pending_jobs=4
+        )
+        manager_b = self._new_channel_manager(
+            jobrunner, "db_b", [ChannelConfig("root", 5)], pending_jobs=4
+        )
+        with mock.patch.object(jobrunner, "_dispatch_job") as dispatch:
+            jobrunner.run_jobs()
+        # number of running jobs must be limited by max_capacity
+        self.assertEqual(dispatch.call_count, 3)
+        self.assertEqual(manager_a.running_count + manager_b.running_count, 3)
diff --git a/queue_job/views/queue_job_channel_views.xml b/queue_job/views/queue_job_channel_views.xml
index 50c245716b..a679596438 100644
--- a/queue_job/views/queue_job_channel_views.xml
+++ b/queue_job/views/queue_job_channel_views.xml
@@ -19,6 +19,16 @@
                     <field name="complete_name" />
                     <field name="removal_interval" />
                 </group>
+                <group string="Job Runner">
+                    <p class="text-muted" colspan="2">
+                        These properties will be used only if the server-side
+                        channels configuration is not used.
+                    </p>
+                    <field name="capacity" />
+                    <field name="sequential" />
+                    <field name="throttle" />
+                    <field name="paused" />
+                </group>
                 <group>
                     <field name="job_function_ids" widget="many2many_tags" />
                 </group>
@@ -32,6 +42,8 @@
         <field name="arch" type="xml">
             <list>
                 <field name="complete_name" />
+                <field name="capacity" />
+                <field name="paused" widget="boolean_toggle" />
             </list>
         </field>
     </record>
@@ -44,6 +56,11 @@
                 <field name="name" />
                 <field name="complete_name" />
                 <field name="parent_id" />
+                <filter
+                    string="Paused"
+                    name="paused"
+                    domain="[('paused', '=', True)]"
+                />
             </search>
         </field>
     </record>