// 设置 socket 永不超时
ini_set('default_socket_timeout', -1); 

// 循环监听一个列表
while(true)
{
    // 从队列中取数据，设置为永久不超时
    $data = $redis->brpop(队列名称, 0);
    // 处理数据
    ....... (比如发邮件)
}

$redis->lpush(队列名称, 数据);

<?php
// 连接永不断开
ini_set('default_socket_timeout', -1); 

// 消费者程序
// 模块发邮件
function sendMail($content)
{
    sleep(2);
    file_put_contents('./mails/'.time(), $content);
}

// 监听消息队列，从队列中取出消息然后发邮件
require('./vendor/autoload.php');

// 连接 Redis
$client = new \Predis\Client([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);

echo "发邮件程序已经启动...\n";

while(true)
{   
    // 从队列的右侧取出消息，如果队列中没有消息就阻塞
    $message = $client->brpop('user_emails', 0);
    echo "收到消息，准备发邮件...\n";
    // 取出消息之后就发邮件
    sendMail($message[1]);
    echo "邮件发送成功，开始监听下一个！\n";
} 

<?php
// 负责注册

$user = 'tom';
$password = '12123';

$sql = 'insert into USER ...';

echo '数据库成功！';
echo '注册成功！';

// 监听消息队列，从队列中取出消息然后发邮件
require('./vendor/autoload.php');

// 连接 Redis
$client = new \Predis\Client([
    'scheme' => 'tcp',
    'host'   => '127.0.0.1',
    'port'   => 6379,
]);
// 出消息队列中放消息
$client->lpush('user_emails', '注册成功！欢迎加入！');

echo '完成 ！！！';

        $http = new Swoole\Http\Server("127.0.0.1", 9501);

    $http->on("start", function ($server) {
        echo "Swoole http server is started at http://127.0.0.1:9501\n";
    });

    $http->on("request", function ($request, $response) {
        $response->header("Content-Type", "text/plain");
        $response->end("Hello World\n");
    });

    $http->start();

        $server = new Swoole\Websocket\Server("127.0.0.1", 9502);

    $server->on('open', function($server, $req) {
        echo "connection open: {$req->fd}\n";
    });

    $server->on('message', function($server, $frame) {
        echo "received message: {$frame->data}\n";
        $server->push($frame->fd, json_encode(["hello", "world"]));
    });

    $server->on('close', function($server, $fd) {
        echo "connection close: {$fd}\n";
    });

    $server->start();

        $server = new Swoole\Server("127.0.0.1", 9503);
    $server->on('connect', function ($server, $fd){
        echo "connection open: {$fd}\n";
    });
    $server->on('receive', function ($server, $fd, $reactor_id, $data) {
        $server->send($fd, "Swoole: {$data}");
        $server->close($fd);
    });
    $server->on('close', function ($server, $fd) {
        echo "connection close: {$fd}\n";
    });
    $server->start();

        $serv = new Swoole\Server("127.0.0.1", 9502, SWOOLE_PROCESS, SWOOLE_SOCK_UDP);

    //监听数据接收事件
    $serv->on('Packet', function ($serv, $data, $clientInfo) {
        $serv->sendto($clientInfo['address'], $clientInfo['port'], "Server ".$data);
        var_dump($clientInfo);
    });

    //启动服务器
    $serv->start();

$server = new Swoole\Server("127.0.0.1", 9502);
    $server->set(array('task_worker_num' => 4));
    $server->on('receive', function($server, $fd, $reactor_id, $data) {
        $task_id = $server->task("Async");
        echo "Dispatch AsyncTask: [id=$task_id]\n";
    });
    $server->on('task', function ($server, $task_id, $reactor_id, $data) {
        echo "New AsyncTask[id=$task_id]\n";
        $server->finish("$data -> OK");
    });
    $server->on('finish', function ($server, $task_id, $data) {
        echo "AsyncTask[$task_id] finished: {$data}\n";
    });
    $server->start();

//睡眠 1 万次，读取，写入，检查和删除文件 1 万次，使用 PDO 和 MySQLi 与数据库通信 1 万次，创建 TCP 服务器和多个客户端相互通信 1 万次，
//创建 UDP 服务器和多个客户端到相互通信 1 万次...... 一切都在一个进程一秒内完美完成！

   Swoole\Runtime::enableCoroutine();//此行代码后，文件操作，sleep，Mysqli，PDO，streams等都变成异步IO，见文档'一键协程化'章节
   $s = microtime(true);
    //Co/run()见文档'协程容器'章节
   Co\run(function() {
    // i just want to sleep...
    for ($c = 100; $c--;) {
        go(function () {
            for ($n = 100; $n--;) {
                usleep(1000);
            }
        });
    }

    // 10k file read and write
    for ($c = 100; $c--;) {
        go(function () use ($c) {
            $tmp_filename = "/tmp/test-{$c}.php";
            for ($n = 100; $n--;) {
                $self = file_get_contents(__FILE__);
                file_put_contents($tmp_filename, $self);
                assert(file_get_contents($tmp_filename) === $self);
            }
            unlink($tmp_filename);
        });
    }

    // 10k pdo and mysqli read
    for ($c = 50; $c--;) {
        go(function () {
            $pdo = new PDO('mysql:host=127.0.0.1;dbname=test;charset=utf8', 'root', 'root');
            $statement = $pdo->prepare('SELECT * FROM `user`');
            for ($n = 100; $n--;) {
                $statement->execute();
                assert(count($statement->fetchAll()) > 0);
            }
        });
    }
    for ($c = 50; $c--;) {
        go(function () {
            $mysqli = new Mysqli('127.0.0.1', 'root', 'root', 'test');
            $statement = $mysqli->prepare('SELECT `id` FROM `user`');
            for ($n = 100; $n--;) {
                $statement->bind_result($id);
                $statement->execute();
                $statement->fetch();
                assert($id > 0);
            }
        });
    }

    // php_stream tcp server & client with 12.8k requests in single process
    function tcp_pack(string $data): string
    {
        return pack('n', strlen($data)) . $data;
    }

    function tcp_length(string $head): int
    {
        return unpack('n', $head)[1];
    }

    go(function () {
        $ctx = stream_context_create(['socket' => ['so_reuseaddr' => true, 'backlog' => 128]]);
        $socket = stream_socket_server(
            'tcp://0.0.0.0:9502',
            $errno, $errstr, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN, $ctx
        );
        if (!$socket) {
            echo "$errstr ($errno)\n";
        } else {
            $i = 0;
            while ($conn = stream_socket_accept($socket, 1)) {
                stream_set_timeout($conn, 5);
                for ($n = 100; $n--;) {
                    $data = fread($conn, tcp_length(fread($conn, 2)));
                    assert($data === "Hello Swoole Server #{$n}!");
                    fwrite($conn, tcp_pack("Hello Swoole Client #{$n}!"));
                }
                if (++$i === 128) {
                    fclose($socket);
                    break;
                }
            }
        }
    });
    for ($c = 128; $c--;) {
        go(function () {
            $fp = stream_socket_client("tcp://127.0.0.1:9502", $errno, $errstr, 1);
            if (!$fp) {
                echo "$errstr ($errno)\n";
            } else {
                stream_set_timeout($fp, 5);
                for ($n = 100; $n--;) {
                    fwrite($fp, tcp_pack("Hello Swoole Server #{$n}!"));
                    $data = fread($fp, tcp_length(fread($fp, 2)));
                    assert($data === "Hello Swoole Client #{$n}!");
                }
                fclose($fp);
            }
        });
    }

    // udp server & client with 12.8k requests in single process
    go(function () {
        $socket = new Swoole\Coroutine\Socket(AF_INET, SOCK_DGRAM, 0);
        $socket->bind('127.0.0.1', 9503);
        $client_map = [];
        for ($c = 128; $c--;) {
            for ($n = 0; $n < 100; $n++) {
                $recv = $socket->recvfrom($peer);
                $client_uid = "{$peer['address']}:{$peer['port']}";
                $id = $client_map[$client_uid] = ($client_map[$client_uid] ?? -1) + 1;
                assert($recv === "Client: Hello #{$id}!");
                $socket->sendto($peer['address'], $peer['port'], "Server: Hello #{$id}!");
            }
        }
        $socket->close();
    });
    for ($c = 128; $c--;) {
        go(function () {
            $fp = stream_socket_client("udp://127.0.0.1:9503", $errno, $errstr, 1);
            if (!$fp) {
                echo "$errstr ($errno)\n";
            } else {
                for ($n = 0; $n < 100; $n++) {
                    fwrite($fp, "Client: Hello #{$n}!");
                    $recv = fread($fp, 1024);
                    list($address, $port) = explode(':', (stream_socket_get_name($fp, true)));
                    assert($address === '127.0.0.1' && (int)$port === 9503);
                    assert($recv === "Server: Hello #{$n}!");
                }
                fclose($fp);
            }
        });
    }
  });
  echo 'use ' . (microtime(true) - $s) . ' s';

 Co\run(function(){
        //使用Channel进行协程间通讯
        $chan = new Swoole\Coroutine\Channel(1);
        Swoole\Coroutine::create(function () use ($chan) {
            for($i = 0; $i < 100000; $i++) {
                co::sleep(1.0);
                $chan->push(['rand' => rand(1000, 9999), 'index' => $i]);
                echo "$i\n";
            }
        });
        Swoole\Coroutine::create(function () use ($chan) {
            while(1) {
                $data = $chan->pop();
                var_dump($data);
            }
        });
  });

use Swoft\Http\Message\Request;
use Swoft\Http\Server\Annotation\Mapping\Controller;
use Swoft\Http\Server\Annotation\Mapping\RequestMapping;

/**
 * Class Home
 *
 * @Controller(prefix="home")
 */
class Home
{
    /**
     * 该方法路由地址为 /home/index
     *
     * @RequestMapping(route="/index", method="post")
     *
     * @param Request $request
     */
    public function index(Request $request)
    {
        // TODO:
    }
}