Swoole Process 内 UnixSocket 通信

server

<?php

use Swoole\Event;
use Swoole\Process;
use Swoole\Coroutine\Socket;

    $httpServer = new Swoole\Http\Server("127.0.0.1", 8725);

    $workerNum = 1;
    $serverName = 'JesseServer';

    for ($i = 0; $i < $workerNum; $i++) {
        $process = new Process(function(Process $process) use ($i) {
            swoole_set_process_name("task_worker_". $i);
            // 监听pipe的读
            Event::add($process->pipe, function ($process) {
                try {
                    $process->read();
                } catch (\Throwable $e) {
                    throw new Exception($e->getMessage());
                }
            });

            // 进程退出
            Process::signal(SIGTERM,function () use ($process){
                Event::del($process->pipe);
                Process::signal(SIGTERM, null);
                Event::exit();
            });



            // socket通信
            if (file_exists($file = socketPipe($i))) {
                unlink($file);
            }

            $socketServer = new Socket(AF_UNIX, SOCK_STREAM, 0);
            $socketServer->setOption(SOL_SOCKET,SO_LINGER, ['l_linger' => 0, 'l_onoff' => 0]);
            if(! $socketServer->bind($file)) {
                throw new Exception('bind socket failed: '. $socketServer->errMsg);
            }

            if(! $socketServer->listen(2048)){
                throw new Exception(' listen socket failed: '. $socketServer->errMsg);
            }


            while(true) {
                $client = $socketServer->accept(-1);
                var_dump($client);
            }

            Event::wait();
        }, false, SOCK_DGRAM, false);

        $httpServer->addProcess($process);
    }

    $httpServer->on("request", function ($request, $response){

    });
    $httpServer->start();


/**
 * @param int $id
 * @return string
 */
function socketPipe(int $id):string
{
    return dirname(__DIR__)."/Cases/TaskWorker.{$id}.sock";
}

client

<?php

use Swoole\Coroutine\Client;
use HyperfTest\Cases\Protocol;

\Swoole\Coroutine\run(function() {

    $client = new Client(SWOOLE_UNIX_STREAM);
    if (! $client) {
        var_dump('client failed');
    }
    $client->set(
        [
            'open_length_check'     => true,
            'package_length_type'   => 'N',
            'package_length_offset' => 0,
            'package_body_offset'   => 4,
            'package_max_length'    => 1024
        ]
    );

    $client->connect(socketPipe(randomWorkerId()), 0, 3);

    $client->send(Protocol::pack('hello'));

    $ret = $client->recv();

    $client->close();

    if (! empty($ret)) {
        var_dump(Protocol::unpack($ret));
    }
});

/**
 * @param int $id
 * @return string
 */
function socketPipe(int $id):string
{
    return dirname(__DIR__)."/Cases/TaskWorker.{$id}.sock";
}

/**
 * @return int
 */
function randomWorkerId(): int
{
    mt_srand();
    return rand(0, 2);
}

protocol

<?php
/**
 * User: jesse * Date: 2023/8/31 14:34 */
namespace HyperfTest\Cases;

class Protocol
{
  public static function pack(string $data): string
  {
  return pack('N', strlen($data)) . $data;
  }

  public static function packDataLength(string $head): int
  {
  return unpack('N', $head)[1];
  }

  public static function unpack(string $data): string
  {
  return substr($data, 4);
  }
}

现在Client发消息 Server没法收到， 有经验的同学帮忙看看

================================================================

问完AI后正确写法是这样，记录下免费以后忘记

server

php
<?php

use Swoole\Coroutine\Socket;

// 服务器端代码
$serverProcess = new Swoole\Process(function (Swoole\Process $process) {
    // 创建一个Unix域套接字
    $socketFile = '/tmp/php.sock';
    if (file_exists($socketFile)) unlink($socketFile);
    $socket = new Socket(AF_UNIX, SOCK_STREAM, 0);
    $socket->bind($socketFile);
    $socket->listen();

    echo "Server started, waiting for client...\n";

    // 接受客户端连接
    $client = $socket->accept();
    if (!$client) {
        exit("Failed to accept client connection");
    }

    echo "Client connected\n";

    // 读取客户端发送的数据
    $data = $client->recv();

    echo "Received data from client: $data\n";

    // 向客户端发送响应数据
    $response = "Hello from server!";
    $client->send($response);

    // 关闭套接字
    $client->close();
    $socket->close();
}, false, SOCK_STREAM, true);

$serverProcess->start();

// 等待子进程结束
Swoole\Process::wait(true);

client

<?php

use Swoole\Coroutine\Socket;
use HyperfTest\Cases\Protocol;

Swoole\Coroutine\run(function() {
    // 创建一个Unix域套接字
    $socketFile = '/tmp/php.sock';
    $socket = new Socket(AF_UNIX, SOCK_STREAM, 0);
    $socket->setProtocol([
        'open_length_check'     => true,
        'package_length_type'   => 'N',
        'package_length_offset' => 0,
        'package_body_offset'   => 4,
        'package_max_length'    => 2048
    ]);

    // 连接到服务器的Unix域套接字
    if (!$socket->connect($socketFile)) {
        exit("Failed to connect to server");
    }

    echo "Connected to server\n";

    // 向服务器发送数据
    $data = "Hello from client!";
    $socket->send($data);

    // 读取服务器响应数据
    $response = $socket->recv();
    echo "Received response from server: $response\n";

    // 关闭套接字
    $socket->close();
});

本作品采用《CC 协议》，转载必须注明作者和本文链接