你這種工作邏輯類似于流水線；
可以開(kāi)N個(gè)消費(fèi)者，通過(guò)隊(duì)列來(lái)實(shí)現(xiàn)你要的功能；
工作流：1隊(duì)列 --->>> 2隊(duì)列 --->>> 3隊(duì)列 --->>> 4隊(duì)列 --->>> 5隊(duì)列 --->>> 6隊(duì)列

以TaskProcess 類為例，這個(gè)類只做2個(gè)事情

1. 構(gòu)造方法傳入任務(wù)類型或者處理的隊(duì)列名
2. onWorkerStart運(yùn)行后，添加每秒執(zhí)行的定時(shí)器，去處理當(dāng)前任務(wù)隊(duì)列；

按你的流水線，在進(jìn)程配置內(nèi)新增進(jìn)程配置（一定要設(shè)置構(gòu)造函數(shù) 參數(shù)）
完畢。

進(jìn)程配置

'pipeline_1' => [
    'count' => 1,
    'handler' => \process\TaskProcess::class,
    'constructor' => [
        'type' => '任務(wù)類型1',
        'queue' => '隊(duì)列1'
    ],
],
'pipeline_2' => [
    'count' => 5,
    // 根據(jù)任務(wù)繁重情況設(shè)置進(jìn)程數(shù)
    'handler' => \process\TaskProcess::class,
    'constructor' => [
        'type' => '任務(wù)類型2',
        'queue' => '隊(duì)列2'
    ],
],
'pipeline_3' => [
    'count' => 3,
    // 根據(jù)任務(wù)繁重情況設(shè)置進(jìn)程數(shù)
    'handler' => \process\TaskProcess::class,
    'constructor' => [
        'type' => '任務(wù)類型3',
        'queue' => '隊(duì)列3'
    ],
],

<?php

namespace process;

use support\Redis;
use Workerman\Timer;

class TaskProcess
{
    /**
     * @var string
     */
    protected string $type;
    /**
     * @var string
     */
    protected string $queue;

    /**
     * 構(gòu)造函數(shù)
     * @param string $type
     * @param string $queue
     */
    public function __construct(string $type, string $queue)
    {
        $this->type = $type;
        $this->queue = $queue;
    }

    /**
     * 根據(jù)類型，得到下一個(gè)工作流隊(duì)列名稱
     * @return string
     */
    public function getNextQueue(): string
    {
        //todo...
        switch ($this->type) {
            case '':
                return '';
            default:
                return '';
        }
    }

    /**
     * 進(jìn)程啟動(dòng)時(shí)執(zhí)行
     * @return void
     */
    public function onWorkerStart(): void
    {
        Timer::add(1, function () {
            //當(dāng)前工作流隊(duì)列內(nèi)有任務(wù)
            if (Redis::lLen($this->queue)) {
                $data = Redis::lPop($this->queue);
                //todo... 完成當(dāng)前流程，得到結(jié)果
                $result = call_user_func([$this, $this->type], $data);
                //todo... 投遞到下個(gè)工作流隊(duì)列
                Redis::rPush($this->getNextQueue(), $result);
            }
        });
    }

    /**
     * step1
     * @param $data
     * @return void
     */
    protected function step1($data): void
    {
        //todo... 業(yè)務(wù)邏輯
    }

    /**
     * step2
     * @param $data
     * @return void
     */
    protected function step2($data): void
    {
        //todo... 業(yè)務(wù)邏輯
    }

    /**
     * step3
     * @param $data
     * @return void
     */
    protected function step3($data): void
    {
        //todo... 業(yè)務(wù)邏輯
    }

    /**
     * step4
     * @param $data
     * @return void
     */
    protected function step4($data): void
    {
        //todo... 業(yè)務(wù)邏輯
    }
}

需要阻塞，數(shù)據(jù)來(lái)了立馬執(zhí)行立馬傳給下一個(gè)進(jìn)程，分散到多個(gè)進(jìn)程的目的就是為了最大化減少運(yùn)行時(shí)間。

子進(jìn)程A：--> 運(yùn)行10秒計(jì)算任務(wù) --> 等待數(shù)據(jù)1，收到后馬上處理數(shù)據(jù)然后返回?cái)?shù)據(jù)2 --> 運(yùn)行10秒計(jì)算任務(wù) --> 和主進(jìn)程通信 -->……
子進(jìn)程B：--> 運(yùn)行10秒計(jì)算任務(wù) --> 等待數(shù)據(jù)2，處理成數(shù)據(jù)3返回 --> 運(yùn)行10秒計(jì)算任務(wù) --> 和主進(jìn)程通信 -->……
子進(jìn)程C：--> 運(yùn)行10秒計(jì)算任務(wù) --> 等待數(shù)據(jù)3，處理成數(shù)據(jù)4返回 --> 運(yùn)行10秒計(jì)算任務(wù) --> 和主進(jìn)程通信 -->……

最終的目的是要分散到64個(gè)進(jìn)程同時(shí)計(jì)算

為什么非要強(qiáng)行用拖拉機(jī)干掛車的事?

主進(jìn)程，子進(jìn)程。
你應(yīng)該是沒(méi)看完手冊(cè)，
webman，workerman 你知道主進(jìn)程是干嘛用的嗎，
你開(kāi)的每一個(gè)進(jìn)程都是平級(jí)的，workerman,webman的主進(jìn)程用來(lái)

for ($process_sub>= 你設(shè)置的進(jìn)程數(shù)){
創(chuàng)建一個(gè)子進(jìn)程
}

workerman，webman的主進(jìn)程主要是用來(lái)supervise用的，你還想主進(jìn)程做啥呢。

你要是希望一個(gè)一個(gè)順序處理，應(yīng)該用到隊(duì)列

你這樣的想法，實(shí)際上是線程模型的思想，線程模型的思想主要是讓主線程分配，子線程執(zhí)行，主線程回收數(shù)據(jù)；
進(jìn)程模型一般不會(huì)有這樣的思路，通常來(lái)說(shuō)不會(huì)通過(guò)主進(jìn)程來(lái)進(jìn)行任務(wù)的分配，而是通過(guò)一些共同特征，讓子進(jìn)程自己通過(guò)特征來(lái)獲取自己應(yīng)該獲取的任務(wù)，而主進(jìn)程只負(fù)責(zé)一件事情，就是子進(jìn)程的正常運(yùn)行和整體服務(wù)的正常運(yùn)行；
如果你想要強(qiáng)行實(shí)現(xiàn)主進(jìn)程分配任務(wù)讓固定子進(jìn)程進(jìn)行處理，我建議每一個(gè)子進(jìn)程都與主進(jìn)程分別建立一條專屬的channel，然后通過(guò)發(fā)布訂閱的方式進(jìn)行業(yè)務(wù)處理

我覺(jué)得你這個(gè)需要用go channel實(shí)現(xiàn)很簡(jiǎn)單