MCP-771: [MCP-608] Multiple consumers prototype development

Author: adifucan

app/code/Magento/MessageQueue/Console/StartConsumerCommand.php

@@ -23,6 +23,7 @@ class StartConsumerCommand extends Command
     const OPTION_BATCH_SIZE = 'batch-size';
     const OPTION_AREACODE = 'area-code';
     const OPTION_SINGLE_THREAD = 'single-thread';
+    const OPTION_MULTI_PROCESS = 'multi-process';
     const PID_FILE_PATH = 'pid-file-path';
     const COMMAND_QUEUE_CONSUMERS_START = 'queue:consumers:start';
 
@@ -88,9 +89,7 @@ protected function execute(InputInterface $input, OutputInterface $output)
 
         $consumer = $this->consumerFactory->get($consumerName, $batchSize);
         $consumer->process($numberOfMessages);
-        if ($singleThread) {
-            $this->lockManager->unlock(md5($consumerName)); //phpcs:ignore
-        }
+        $this->lockManager->unlock(md5($consumerName)); //phpcs:ignore
 
         return \Magento\Framework\Console\Cli::RETURN_SUCCESS;
     }
@@ -133,6 +132,12 @@ protected function configure()
             InputOption::VALUE_NONE,
             'This option prevents running multiple copies of one consumer simultaneously.'
         );
+        $this->addOption(
+            self::OPTION_MULTI_PROCESS,
+            null,
+            InputOption::VALUE_OPTIONAL,
+            'The number of processes per consumer.'
+        );
         $this->addOption(
             self::PID_FILE_PATH,
             null,
@@ -166,6 +171,9 @@ protected function configure()
 To save PID enter path (This option is deprecated, use --single-thread instead):
 
     <comment>%command.full_name% someConsumer --pid-file-path='/var/someConsumer.pid'</comment>
+To define the number of processes per consumer:
+
+    <comment>%command.full_name% someConsumer --multi-process'</comment>
 HELP
         );
         parent::configure();

app/code/Magento/MessageQueue/Model/Cron/ConsumersRunner.php

@@ -111,6 +111,7 @@ public function __construct(
     public function run()
     {
         $runByCron = $this->deploymentConfig->get('cron_consumers_runner/cron_run', true);
+        $multipleProcesses = $this->deploymentConfig->get('cron_consumers_runner/multiple_processes', []);
 
         if (!$runByCron) {
             return;
@@ -125,19 +126,47 @@ public function run()
                 continue;
             }
 
-            $arguments = [
-                $consumer->getName(),
-                '--single-thread'
-            ];
-
-            if ($maxMessages) {
-                $arguments[] = '--max-messages=' . min($consumer->getMaxMessages() ?? $maxMessages, $maxMessages);
+            if (array_key_exists($consumer->getName(), $multipleProcesses)) {
+                $numberOfProcesses = $multipleProcesses[$consumer->getName()];
+
+                for ($i = 1; $i <= $numberOfProcesses; $i++) {
+                    if ($this->lockManager->isLocked(md5($consumer->getName()))) { //phpcs:ignore
+                        continue;
+                    }
+                    $arguments = [
+                        $consumer->getName(),
+                        '--multi-process=' . $i
+                    ];
+
+                    if ($maxMessages) {
+                        $arguments[] = '--max-messages=' . min($consumer->getMaxMessages() ?? $maxMessages, $maxMessages);
+                    }
+
+                    $command = $php . ' ' . BP . '/bin/magento queue:consumers:start %s %s'
+                        . ($maxMessages ? ' %s' : '');
+
+                    $this->shellBackground->execute($command, $arguments);
+                }
+                $this->lockManager->lock(md5($consumer->getName()));
+            } else {
+                if ($this->lockManager->isLocked(md5($consumer->getName()))) { //phpcs:ignore
+                    return false;
+                }
+
+                $arguments = [
+                    $consumer->getName(),
+                    '--single-thread'
+                ];
+
+                if ($maxMessages) {
+                    $arguments[] = '--max-messages=' . min($consumer->getMaxMessages() ?? $maxMessages, $maxMessages);
+                }
+
+                $command = $php . ' ' . BP . '/bin/magento queue:consumers:start %s %s'
+                    . ($maxMessages ? ' %s' : '');
+
+                $this->shellBackground->execute($command, $arguments);
             }
-
-            $command = $php . ' ' . BP . '/bin/magento queue:consumers:start %s %s'
-                . ($maxMessages ? ' %s' : '');
-
-            $this->shellBackground->execute($command, $arguments);
         }
     }
 
@@ -157,10 +186,6 @@ private function canBeRun(ConsumerConfigItemInterface $consumerConfig, array $al
             return false;
         }
 
-        if ($this->lockManager->isLocked(md5($consumerName))) { //phpcs:ignore
-            return false;
-        }
-
         $connectionName = $consumerConfig->getConnection();
         try {
             $this->mqConnectionTypeResolver->getConnectionType($connectionName);