diff --git a/README.md b/README.md
index 942e4155..2997080c 100644
--- a/README.md
+++ b/README.md
@@ -138,6 +138,8 @@ For setting up all classes manually, see the [Manual configuration](docs/guide/e
 To send a message to the queue, get the queue instance and call `push()`. Typically the queue is injected as a dependency:
 
 ```php
+use Yiisoft\Queue\QueueInterface;
+
 final readonly class Foo
 {
     public function __construct(private QueueInterface $queue) {}
@@ -159,11 +161,13 @@ By default, Yii Framework uses [yiisoft/yii-console](https://github.com/yiisoft/
 ```bash
 ./yii queue:run # Handle all existing messages in the queue
 ./yii queue:listen [queueName] # Start a daemon listening for new messages permanently from the specified queue
-./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all queues or specified list of queues (use with caution in production, recommended for dev only)
+./yii queue:listen-all [queueName [queueName2 [...]]] # Start a daemon listening for new messages permanently from all consumer-capable queues or specified list of queues (use with caution in production, recommended for dev only)
 ```
 
 See [Console commands](docs/guide/en/console-commands.md) for more details.
 
+Consumers are represented by `Yiisoft\Queue\QueueConsumerInterface`; it contains `run()` and `listen()`.
+
 > In case you're running the queue in synchronous mode (no adapter), `queue:listen` logs an info message and exits. The messages are processed immediately when pushed.
 
 ## Documentation
diff --git a/docs/guide/en/configuration-manual.md b/docs/guide/en/configuration-manual.md
index 181874e7..acf81055 100644
--- a/docs/guide/en/configuration-manual.md
+++ b/docs/guide/en/configuration-manual.md
@@ -98,9 +98,15 @@ $provider = new PredefinedQueueProvider([
 
 ## Running the queue
 
+Message consumption methods are available on `Yiisoft\Queue\QueueConsumerInterface`.
+The built-in `Queue` implements both `QueueInterface` for producing messages and `QueueConsumerInterface` for consuming them.
+
 ### Processing existing messages
 
 ```php
+use Yiisoft\Queue\QueueConsumerInterface;
+
+/** @var QueueConsumerInterface $queue */
 $queue->run();      // Process all messages
 $queue->run(10);    // Process up to 10 messages
 ```
@@ -108,6 +114,9 @@ $queue->run(10);    // Process up to 10 messages
 ### Listening for new messages
 
 ```php
+use Yiisoft\Queue\QueueConsumerInterface;
+
+/** @var QueueConsumerInterface $queue */
 $queue->listen();   // Run indefinitely
 ```
 
diff --git a/docs/guide/en/console-commands.md b/docs/guide/en/console-commands.md
index 0157e7f5..c3f5443e 100644
--- a/docs/guide/en/console-commands.md
+++ b/docs/guide/en/console-commands.md
@@ -6,7 +6,7 @@ If you are using [yiisoft/config](https://github.com/yiisoft/config) and [yiisof
 
 If you are using [symfony/console](https://github.com/symfony/console) directly, you should register the commands manually.
 
-> **Note:** The default queue name list (used when no queue names are passed to a command) is only available when using [yiisoft/config](https://github.com/yiisoft/config) and [yiisoft/yii-console](https://github.com/yiisoft/yii-console). Without them, you must pass the queue name list explicitly to the command constructor.
+> **Note:** When no queue names are passed to `queue:run` or `queue:listen-all`, the commands use the queue names provided by `QueueProviderInterface::getNames()` and skip queues that don't implement `QueueConsumerInterface`. Explicitly passed queue names must be consumer-capable.
 
 In [yiisoft/app](https://github.com/yiisoft/app) the `yii` console binary is provided out of the box.
 If you are using [yiisoft/yii-console](https://github.com/yiisoft/yii-console) or `symfony/console` without that template, invoke these commands the same way you invoke other console commands in your application.
@@ -17,7 +17,7 @@ The command `queue:run` obtains and handles messages until the queue is empty, t
 
 You can also narrow the scope of processed messages by specifying queue name(s) and maximum number of messages to process:
 
-- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names.
+- Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names.
 - Use `--limit` to limit the number of messages processed. When set, command will exit either when all the messages are processed or when the maximum count is reached.
 
 The full command signature is:
@@ -39,7 +39,7 @@ yii queue:listen [queueName]
 
 The following command iterates through multiple queues and is meant to be used in development environment only, as it consumes a lot of CPU for iterating through queues. You can pass to it:
 
-- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered queue names.
+- `queueName` argument(s). Specify one or more queue names to process. Messages from other queues will be ignored. Defaults to all registered consumer-capable queue names.
 - `--limit` option to limit the number of messages processed before switching to another queue. E.g. you set `--limit` to 500 and right now you have 1000 messages in `queue1`. This command will consume only 500 of them, then it will switch to `queue2` to see if there are any messages there. Defaults to `0` (no limit).
 - `--pause` option to specify the number of seconds to pause between checking queues when no messages are found. Defaults to `1`.
 
diff --git a/docs/guide/en/queue-names-advanced.md b/docs/guide/en/queue-names-advanced.md
index 21cdd8cc..8ccc9a0b 100644
--- a/docs/guide/en/queue-names-advanced.md
+++ b/docs/guide/en/queue-names-advanced.md
@@ -7,6 +7,7 @@ Use this reference when you need to understand how queue names map to adapters,
 - A queue name (string or `BackedEnum`) is passed to `Yiisoft\Queue\Provider\QueueProviderInterface::get($queueName)`.
 - The provider returns a `Yiisoft\Queue\QueueInterface` instance configured for that name.
 - `QueueInterface::getName()` can be used for introspection; it returns the logical name the queue was created with.
+- Queues that can consume messages also implement `Yiisoft\Queue\QueueConsumerInterface`.
 
 ## Provider implementations
 
diff --git a/src/Command/ListenAllCommand.php b/src/Command/ListenAllCommand.php
index 51b55009..e9904a67 100644
--- a/src/Command/ListenAllCommand.php
+++ b/src/Command/ListenAllCommand.php
@@ -11,13 +11,18 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:listen-all',
     'Listens the all the given queues and executes messages as they come. '
         . 'Meant to be used in development environment only. '
-        . 'Listens all configured queues by default in case you\'re using yiisoft/config. '
+        . 'Listens all consumer-capable configured queues by default. '
         . 'Needs to be stopped manually.',
 )]
 final class ListenAllCommand extends Command
@@ -38,7 +43,7 @@ public function configure(): void
             'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
             'Queue name list to connect to',
-            $this->queueProvider->getNames(),
+            [],
         )
             ->addOption(
                 'pause',
@@ -61,10 +66,24 @@ public function configure(): void
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
+        /** @var string[] $queueNames */
+        $queueNames = $input->getArgument('queue');
+        $queueIsRequired = $queueNames !== [];
+        if (!$queueIsRequired) {
+            $queueNames = $this->queueProvider->getNames();
+        }
+
         $queues = [];
         /** @var string $queue */
-        foreach ($input->getArgument('queue') as $queue) {
-            $queues[] = $this->queueProvider->get($queue);
+        foreach ($queueNames as $queue) {
+            $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired);
+            if ($queueConsumer !== null) {
+                $queues[] = $queueConsumer;
+            }
+        }
+
+        if ($queues === []) {
+            return Command::SUCCESS;
         }
 
         $pauseSeconds = (int) $input->getOption('pause');
@@ -86,4 +105,26 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         return 0;
     }
+
+    private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            if (!$required) {
+                return null;
+            }
+
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }
diff --git a/src/Command/ListenCommand.php b/src/Command/ListenCommand.php
index cfc4c9db..c1cdb39c 100644
--- a/src/Command/ListenCommand.php
+++ b/src/Command/ListenCommand.php
@@ -9,7 +9,12 @@
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:listen',
@@ -37,8 +42,26 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $queueName = (string) $input->getArgument('queue');
 
-        $this->queueProvider->get($queueName)->listen();
+        $this->getQueueConsumer($queueName)->listen();
 
         return Command::SUCCESS;
     }
+
+    private function getQueueConsumer(string $name): QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }
diff --git a/src/Command/RunCommand.php b/src/Command/RunCommand.php
index c9c0baa8..09747ad8 100644
--- a/src/Command/RunCommand.php
+++ b/src/Command/RunCommand.php
@@ -10,7 +10,12 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+
+use function get_debug_type;
+use function sprintf;
 
 #[AsCommand(
     'queue:run',
@@ -30,7 +35,7 @@ public function configure(): void
             'queue',
             InputArgument::OPTIONAL | InputArgument::IS_ARRAY,
             'Queue name list to connect to.',
-            $this->queueProvider->getNames(),
+            [],
         )
             ->addOption(
                 'limit',
@@ -44,16 +49,48 @@ public function configure(): void
 
     protected function execute(InputInterface $input, OutputInterface $output): int
     {
+        /** @var string[] $queueNames */
+        $queueNames = $input->getArgument('queue');
+        $queueIsRequired = $queueNames !== [];
+        if (!$queueIsRequired) {
+            $queueNames = $this->queueProvider->getNames();
+        }
+
         /** @var string $queue */
-        foreach ($input->getArgument('queue') as $queue) {
+        foreach ($queueNames as $queue) {
+            $queueConsumer = $this->getQueueConsumer($queue, $queueIsRequired);
+            if ($queueConsumer === null) {
+                continue;
+            }
+
             $output->write("Processing queue $queue... ");
-            $count = $this->queueProvider
-                ->get($queue)
-                ->run((int) $input->getOption('limit'));
+            $count = $queueConsumer->run((int) $input->getOption('limit'));
 
             $output->writeln("Messages processed: $count.");
         }
 
         return 0;
     }
+
+    private function getQueueConsumer(string $name, bool $required): ?QueueConsumerInterface
+    {
+        $queue = $this->queueProvider->get($name);
+
+        if (!$queue instanceof QueueConsumerInterface) {
+            if (!$required) {
+                return null;
+            }
+
+            throw new InvalidQueueConfigException(
+                sprintf(
+                    'Queue "%s" must implement "%s" to consume messages. Got "%s" instead.',
+                    $name,
+                    QueueConsumerInterface::class,
+                    get_debug_type($queue),
+                ),
+            );
+        }
+
+        return $queue;
+    }
 }
diff --git a/src/Debug/QueueConsumerDecorator.php b/src/Debug/QueueConsumerDecorator.php
new file mode 100644
index 00000000..b6223ab0
--- /dev/null
+++ b/src/Debug/QueueConsumerDecorator.php
@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue\Debug;
+
+use Yiisoft\Queue\MessageStatus;
+use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
+use Yiisoft\Queue\QueueInterface;
+
+final class QueueConsumerDecorator implements QueueInterface, QueueConsumerInterface
+{
+    private readonly QueueDecorator $queueDecorator;
+
+    public function __construct(
+        private readonly QueueInterface&QueueConsumerInterface $queue,
+        QueueCollector $collector,
+    ) {
+        $this->queueDecorator = new QueueDecorator($queue, $collector);
+    }
+
+    public function status(string|int $id): MessageStatus
+    {
+        return $this->queueDecorator->status($id);
+    }
+
+    public function push(MessageInterface $message): MessageInterface
+    {
+        return $this->queueDecorator->push($message);
+    }
+
+    public function run(int $max = 0): int
+    {
+        return $this->queue->run($max);
+    }
+
+    public function listen(): void
+    {
+        $this->queue->listen();
+    }
+
+    public function getName(): string
+    {
+        return $this->queueDecorator->getName();
+    }
+}
diff --git a/src/Debug/QueueDecorator.php b/src/Debug/QueueDecorator.php
index de0e901b..3b3f9617 100644
--- a/src/Debug/QueueDecorator.php
+++ b/src/Debug/QueueDecorator.php
@@ -36,16 +36,6 @@ public function push(MessageInterface $message): MessageInterface
         return $message;
     }
 
-    public function run(int $max = 0): int
-    {
-        return $this->queue->run($max);
-    }
-
-    public function listen(): void
-    {
-        $this->queue->listen();
-    }
-
     public function getName(): string
     {
         return $this->queue->getName();
diff --git a/src/Debug/QueueProviderInterfaceProxy.php b/src/Debug/QueueProviderInterfaceProxy.php
index c28d1b20..b40a0b56 100644
--- a/src/Debug/QueueProviderInterfaceProxy.php
+++ b/src/Debug/QueueProviderInterfaceProxy.php
@@ -6,6 +6,7 @@
 
 use BackedEnum;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueProviderInterfaceProxy implements QueueProviderInterface
@@ -19,7 +20,9 @@ public function get(string|BackedEnum $name): QueueInterface
     {
         $queue = $this->queueProvider->get($name);
 
-        return new QueueDecorator($queue, $this->collector);
+        return $queue instanceof QueueConsumerInterface
+            ? new QueueConsumerDecorator($queue, $this->collector)
+            : new QueueDecorator($queue, $this->collector);
     }
 
     public function has(string|BackedEnum $name): bool
diff --git a/src/Queue.php b/src/Queue.php
index b9bc5d48..bb2c6d5a 100644
--- a/src/Queue.php
+++ b/src/Queue.php
@@ -17,7 +17,7 @@
 use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
 
-final class Queue implements QueueInterface
+final class Queue implements QueueInterface, QueueConsumerInterface
 {
     private string $name;
 
diff --git a/src/QueueConsumerInterface.php b/src/QueueConsumerInterface.php
new file mode 100644
index 00000000..320f1b77
--- /dev/null
+++ b/src/QueueConsumerInterface.php
@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yiisoft\Queue;
+
+/**
+ * Consumes messages from a queue.
+ *
+ * This interface complements {@see QueueInterface}, which is responsible for producing messages and checking their
+ * status.
+ */
+interface QueueConsumerInterface
+{
+    /**
+     * Handle all existing messages and exit.
+     *
+     * @return int Number of messages processed.
+     */
+    public function run(int $max = 0): int;
+
+    /**
+     * Listen to the queue and handle messages as they come.
+     */
+    public function listen(): void;
+}
diff --git a/src/QueueInterface.php b/src/QueueInterface.php
index b4ac11e3..66a70b39 100644
--- a/src/QueueInterface.php
+++ b/src/QueueInterface.php
@@ -6,6 +6,11 @@
 
 use Yiisoft\Queue\Message\MessageInterface;
 
+/**
+ * Produces messages by pushing them into a queue.
+ *
+ * For consuming messages from a queue, use {@see QueueConsumerInterface}.
+ */
 interface QueueInterface
 {
     /**
@@ -17,18 +22,6 @@ interface QueueInterface
      */
     public function push(MessageInterface $message): MessageInterface;
 
-    /**
-     * Handle all existing messages and exit
-     *
-     * @return int Number of messages processed.
-     */
-    public function run(int $max = 0): int;
-
-    /**
-     * Listen to the queue and handle messages as they come
-     */
-    public function listen(): void;
-
     /**
      * @param int|string $id A message ID.
      *
diff --git a/stubs/StubQueue.php b/stubs/StubQueue.php
index f34c6d11..a929fcf5 100644
--- a/stubs/StubQueue.php
+++ b/stubs/StubQueue.php
@@ -6,12 +6,13 @@
 
 use Yiisoft\Queue\MessageStatus;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 /**
  * Stub queue that does nothing. Message status is always "done".
  */
-final class StubQueue implements QueueInterface
+final class StubQueue implements QueueInterface, QueueConsumerInterface
 {
     public function __construct(
         private string $name = 'default',
diff --git a/tests/Benchmark/QueueBench.php b/tests/Benchmark/QueueBench.php
index 873f773d..24ca03bf 100644
--- a/tests/Benchmark/QueueBench.php
+++ b/tests/Benchmark/QueueBench.php
@@ -22,6 +22,7 @@
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareConfig;
 use Yiisoft\Queue\Middleware\Push\PushMiddlewareFactory;
 use Yiisoft\Queue\Queue;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Tests\Benchmark\Support\VoidAdapter;
 use Yiisoft\Queue\Worker\Worker;
@@ -29,7 +30,7 @@
 
 final class QueueBench
 {
-    private readonly QueueInterface $queue;
+    private readonly QueueInterface&QueueConsumerInterface $queue;
     private readonly MessageSerializer $serializer;
     private readonly VoidAdapter $adapter;
 
diff --git a/tests/Unit/Command/ListenAllCommandTest.php b/tests/Unit/Command/ListenAllCommandTest.php
index 24175f58..98224873 100644
--- a/tests/Unit/Command/ListenAllCommandTest.php
+++ b/tests/Unit/Command/ListenAllCommandTest.php
@@ -9,16 +9,18 @@
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Cli\LoopInterface;
 use Yiisoft\Queue\Command\ListenAllCommand;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\PredefinedQueueProvider;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class ListenAllCommandTest extends TestCase
 {
     public function testExecute(): void
     {
-        $queue1 = $this->createMock(QueueInterface::class);
+        $queue1 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue1->expects($this->once())->method('run');
-        $queue2 = $this->createMock(QueueInterface::class);
+        $queue2 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue2->expects($this->once())->method('run');
 
         $queueFactory = new PredefinedQueueProvider([
@@ -39,4 +41,45 @@ public function testExecute(): void
 
         $this->assertEquals(0, $exitCode);
     }
+
+    public function testExecuteSkipsProducerOnlyQueuesByDefault(): void
+    {
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
+        $queue->expects($this->once())->method('run');
+
+        $queueFactory = new PredefinedQueueProvider([
+            'producer-only' => $this->createMock(QueueInterface::class),
+            'consumer' => $queue,
+        ]);
+
+        $loop = $this->createMock(LoopInterface::class);
+        $loop->method('canContinue')->willReturn(true, false);
+
+        $command = new ListenAllCommand(
+            $queueFactory,
+            $loop,
+        );
+        $input = new ArrayInput([], $command->getNativeDefinition());
+        $input->setOption('pause', 0);
+        $exitCode = $command->run($input, $this->createMock(OutputInterface::class));
+
+        $this->assertEquals(0, $exitCode);
+    }
+
+    public function testExecuteRequiresConsumerQueues(): void
+    {
+        $queueFactory = new PredefinedQueueProvider([
+            'producer-only' => $this->createMock(QueueInterface::class),
+        ]);
+        $loop = $this->createMock(LoopInterface::class);
+
+        $this->expectException(InvalidQueueConfigException::class);
+        $this->expectExceptionMessage('Queue "producer-only" must implement');
+
+        $command = new ListenAllCommand(
+            $queueFactory,
+            $loop,
+        );
+        $command->run(new ArrayInput(['queue' => ['producer-only']], $command->getNativeDefinition()), $this->createMock(OutputInterface::class));
+    }
 }
diff --git a/tests/Unit/Command/ListenCommandTest.php b/tests/Unit/Command/ListenCommandTest.php
index 46dda71f..ccd77210 100644
--- a/tests/Unit/Command/ListenCommandTest.php
+++ b/tests/Unit/Command/ListenCommandTest.php
@@ -8,14 +8,16 @@
 use Symfony\Component\Console\Input\StringInput;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Command\ListenCommand;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class ListenCommandTest extends TestCase
 {
     public function testExecuteWithDefaultQueue(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('listen');
 
@@ -34,7 +36,7 @@ public function testExecuteWithDefaultQueue(): void
 
     public function testExecuteWithCustomQueue(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('listen');
 
@@ -53,7 +55,7 @@ public function testExecuteWithCustomQueue(): void
 
     public function testExecuteReturnsZero(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('listen');
 
@@ -66,4 +68,16 @@ public function testExecuteReturnsZero(): void
 
         $this->assertSame(0, $exitCode);
     }
+
+    public function testExecuteRequiresConsumerQueue(): void
+    {
+        $queueProvider = $this->createMock(QueueProviderInterface::class);
+        $queueProvider->method('get')->willReturn($this->createMock(QueueInterface::class));
+
+        $this->expectException(InvalidQueueConfigException::class);
+        $this->expectExceptionMessage('Queue "producer-only" must implement');
+
+        $command = new ListenCommand($queueProvider);
+        $command->run(new StringInput('producer-only'), $this->createMock(OutputInterface::class));
+    }
 }
diff --git a/tests/Unit/Command/RunCommandTest.php b/tests/Unit/Command/RunCommandTest.php
index 1ddc02dc..e2fe4625 100644
--- a/tests/Unit/Command/RunCommandTest.php
+++ b/tests/Unit/Command/RunCommandTest.php
@@ -8,15 +8,17 @@
 use Symfony\Component\Console\Input\StringInput;
 use Symfony\Component\Console\Output\OutputInterface;
 use Yiisoft\Queue\Command\RunCommand;
+use Yiisoft\Queue\Provider\InvalidQueueConfigException;
 use Yiisoft\Queue\Provider\PredefinedQueueProvider;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class RunCommandTest extends TestCase
 {
     public function testExecuteWithSingleQueue(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('run')
             ->with($this->equalTo(0))
@@ -43,13 +45,13 @@ public function testExecuteWithSingleQueue(): void
 
     public function testExecuteWithMultipleQueues(): void
     {
-        $queue1 = $this->createMock(QueueInterface::class);
+        $queue1 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue1->expects($this->once())
             ->method('run')
             ->with($this->equalTo(0))
             ->willReturn(3);
 
-        $queue2 = $this->createMock(QueueInterface::class);
+        $queue2 = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue2->expects($this->once())
             ->method('run')
             ->with($this->equalTo(0))
@@ -75,7 +77,7 @@ public function testExecuteWithMultipleQueues(): void
 
     public function testExecuteWithLimitOption(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('run')
             ->with($this->equalTo(100))
@@ -102,7 +104,7 @@ public function testExecuteWithLimitOption(): void
 
     public function testExecuteWithDefaultQueues(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())
             ->method('run')
             ->with($this->equalTo(0))
@@ -126,4 +128,45 @@ public function testExecuteWithDefaultQueues(): void
 
         $this->assertEquals(0, $exitCode);
     }
+
+    public function testExecuteWithDefaultQueuesSkipsProducerOnlyQueues(): void
+    {
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
+        $queue->expects($this->once())
+            ->method('run')
+            ->with($this->equalTo(0))
+            ->willReturn(2);
+
+        $queueProvider = new PredefinedQueueProvider([
+            'producer-only' => $this->createMock(QueueInterface::class),
+            'consumer' => $queue,
+        ]);
+
+        $input = new StringInput('');
+        $output = $this->createMock(OutputInterface::class);
+        $output->expects($this->once())
+            ->method('write')
+            ->with($this->equalTo('Processing queue consumer... '));
+        $output->expects($this->once())
+            ->method('writeln')
+            ->with($this->equalTo('Messages processed: 2.'));
+
+        $command = new RunCommand($queueProvider);
+        $exitCode = $command->run($input, $output);
+
+        $this->assertEquals(0, $exitCode);
+    }
+
+    public function testExecuteRequiresConsumerQueue(): void
+    {
+        $queueProvider = new PredefinedQueueProvider([
+            'producer-only' => $this->createMock(QueueInterface::class),
+        ]);
+
+        $this->expectException(InvalidQueueConfigException::class);
+        $this->expectExceptionMessage('Queue "producer-only" must implement');
+
+        $command = new RunCommand($queueProvider);
+        $command->run(new StringInput('producer-only'), $this->createMock(OutputInterface::class));
+    }
 }
diff --git a/tests/Unit/Debug/QueueDecoratorTest.php b/tests/Unit/Debug/QueueDecoratorTest.php
index 394f4cb7..b095885f 100644
--- a/tests/Unit/Debug/QueueDecoratorTest.php
+++ b/tests/Unit/Debug/QueueDecoratorTest.php
@@ -6,9 +6,11 @@
 
 use PHPUnit\Framework\TestCase;
 use Yiisoft\Queue\Debug\QueueCollector;
+use Yiisoft\Queue\Debug\QueueConsumerDecorator;
 use Yiisoft\Queue\Debug\QueueDecorator;
 use Yiisoft\Queue\MessageStatus;
 use Yiisoft\Queue\Message\MessageInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueDecoratorTest extends TestCase
@@ -88,10 +90,10 @@ public function testStatusCollectsCallLocation(): void
 
     public function testRun(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())->method('run');
         $collector = new QueueCollector();
-        $decorator = new QueueDecorator(
+        $decorator = new QueueConsumerDecorator(
             $queue,
             $collector,
         );
@@ -101,10 +103,10 @@ public function testRun(): void
 
     public function testListen(): void
     {
-        $queue = $this->createMock(QueueInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
         $queue->expects($this->once())->method('listen');
         $collector = new QueueCollector();
-        $decorator = new QueueDecorator(
+        $decorator = new QueueConsumerDecorator(
             $queue,
             $collector,
         );
diff --git a/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php b/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php
index 2ba2ae42..24915332 100644
--- a/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php
+++ b/tests/Unit/Debug/QueueProviderInterfaceProxyTest.php
@@ -6,9 +6,11 @@
 
 use PHPUnit\Framework\TestCase;
 use Yiisoft\Queue\Debug\QueueCollector;
+use Yiisoft\Queue\Debug\QueueConsumerDecorator;
 use Yiisoft\Queue\Debug\QueueDecorator;
 use Yiisoft\Queue\Debug\QueueProviderInterfaceProxy;
 use Yiisoft\Queue\Provider\QueueProviderInterface;
+use Yiisoft\Queue\QueueConsumerInterface;
 use Yiisoft\Queue\QueueInterface;
 
 final class QueueProviderInterfaceProxyTest extends TestCase
@@ -24,6 +26,20 @@ public function testGet(): void
         $this->assertInstanceOf(QueueDecorator::class, $factory->get('test'));
     }
 
+    public function testGetConsumerQueue(): void
+    {
+        $queueFactory = $this->createMock(QueueProviderInterface::class);
+        $queue = $this->createMockForIntersectionOfInterfaces([QueueInterface::class, QueueConsumerInterface::class]);
+        $queueFactory->expects($this->once())->method('get')->willReturn($queue);
+        $collector = new QueueCollector();
+        $factory = new QueueProviderInterfaceProxy($queueFactory, $collector);
+
+        $decoratedQueue = $factory->get('test');
+
+        $this->assertInstanceOf(QueueConsumerDecorator::class, $decoratedQueue);
+        $this->assertInstanceOf(QueueConsumerInterface::class, $decoratedQueue);
+    }
+
     public function testHas(): void
     {
         $queueFactory = $this->createMock(QueueProviderInterface::class);
diff --git a/tests/Unit/QueueTest.php b/tests/Unit/QueueTest.php
index 254c8076..eecfc8a3 100644
--- a/tests/Unit/QueueTest.php
+++ b/tests/Unit/QueueTest.php
@@ -8,6 +8,8 @@
 use Yiisoft\Queue\Message\IdEnvelope;
 use Yiisoft\Queue\Message\GenericMessage;
 use Yiisoft\Queue\MessageStatus;
+use Yiisoft\Queue\QueueConsumerInterface;
+use Yiisoft\Queue\QueueInterface;
 use Yiisoft\Queue\Stubs\InMemoryAdapter;
 use Yiisoft\Queue\Tests\TestCase;
 
@@ -22,6 +24,16 @@ enum TestQueue: string
 
 final class QueueTest extends TestCase
 {
+    public function testContracts(): void
+    {
+        $queue = $this->createQueue();
+
+        self::assertInstanceOf(QueueInterface::class, $queue);
+        self::assertInstanceOf(QueueConsumerInterface::class, $queue);
+        self::assertFalse(method_exists(QueueInterface::class, 'run'));
+        self::assertFalse(method_exists(QueueInterface::class, 'listen'));
+    }
+
     public function testPushSuccessful(): void
     {
         $adapter = new InMemoryAdapter();