PHPでコマンドラインからマルチスレッド処理を実現する方法（pthreads入門）

   php -v

   sudo pecl install pthreads

   extension=pthreads.so

php -m | grep pthreads

<?php
// Threadクラスを拡張してカスタムスレッドクラスを作成
class MyThread extends Thread {
    public function run() {
        // 実行したい処理
        echo "スレッドが実行されています\n";
        sleep(2); // 2秒間の待機
        echo "スレッドの処理が完了しました\n";
    }
}

// スレッドのインスタンスを作成
$thread = new MyThread();

// スレッドの開始
$thread->start();

// スレッドの終了を待つ
$thread->join();

echo "メインスレッドの処理が完了しました\n";
?>

<?php
class MultiThread extends Thread {
    private $threadId;

    public function __construct($id) {
        $this->threadId = $id;
    }

    public function run() {
        echo "スレッド {$this->threadId} が実行されています\n";
        sleep(1);
        echo "スレッド {$this->threadId} の処理が完了しました\n";
    }
}

// 複数のスレッドを作成して実行
$threads = [];
for ($i = 1; $i <= 3; $i++) {
    $threads[$i] = new MultiThread($i);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待つ
foreach ($threads as $thread) {
    $thread->join();
}

echo "すべてのスレッドの処理が完了しました\n";
?>

<?php
class SharedData extends Threaded {
    public $value = 0;
}

class WorkerThread extends Thread {
    private $sharedData;

    public function __construct($sharedData) {
        $this->sharedData = $sharedData;
    }

    public function run() {
        // 共有データを操作
        $this->sharedData->value++;
        echo "スレッドでのデータ: {$this->sharedData->value}\n";
    }
}

// 共有データオブジェクトを作成
$sharedData = new SharedData();

// スレッドを複数作成
$threads = [];
for ($i = 0; $i < 3; $i++) {
    $threads[$i] = new WorkerThread($sharedData);
    $threads[$i]->start();
}

// 全てのスレッドの終了を待つ
foreach ($threads as $thread) {
    $thread->join();
}

echo "メインスレッドでの最終データ: {$sharedData->value}\n";
?>

<?php
class Counter extends Threaded {
    private $value = 0;
    private $lock;

    public function __construct() {
        $this->lock = new Mutex(); // ロックオブジェクトの作成
    }

    public function increment() {
        Mutex::lock($this->lock); // ロックをかける
        $this->value++;
        Mutex::unlock($this->lock); // ロックを解除
    }

    public function getValue() {
        return $this->value;
    }
}

class IncrementThread extends Thread {
    private $counter;

    public function __construct($counter) {
        $this->counter = $counter;
    }

    public function run() {
        $this->counter->increment();
        echo "スレッドでのカウンタ: {$this->counter->getValue()}\n";
    }
}

$counter = new Counter();
$threads = [];

// 複数のスレッドを生成してカウンタをインクリメント
for ($i = 0; $i < 5; $i++) {
    $threads[$i] = new IncrementThread($counter);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

echo "メインスレッドでの最終カウンタ: {$counter->getValue()}\n";
?>

<?php
class SharedCounter extends Threaded {
    private $count = 0;
    private $mutex;

    public function __construct() {
        $this->mutex = new Mutex(); // Mutexオブジェクトを作成
    }

    public function increment() {
        Mutex::lock($this->mutex); // ロックをかける
        $this->count++;
        Mutex::unlock($this->mutex); // ロックを解除
    }

    public function getCount() {
        return $this->count;
    }
}

class CounterThread extends Thread {
    private $counter;

    public function __construct($counter) {
        $this->counter = $counter;
    }

    public function run() {
        for ($i = 0; $i < 1000; $i++) {
            $this->counter->increment();
        }
    }
}

// 共有カウンタを作成
$counter = new SharedCounter();
$threads = [];

// 5つのスレッドでカウンタを同時に増加させる
for ($i = 0; $i < 5; $i++) {
    $threads[$i] = new CounterThread($counter);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

echo "最終カウント: {$counter->getCount()}\n";
?>

<?php
class LimitedResource extends Threaded {
    private $semaphore;

    public function __construct($maxAccess) {
        $this->semaphore = new Semaphore($maxAccess); // 最大アクセス数を指定
    }

    public function accessResource($threadId) {
        Semaphore::wait($this->semaphore); // ロックを取得
        echo "スレッド {$threadId} がリソースにアクセスしています\n";
        sleep(1); // リソースを使用中
        echo "スレッド {$threadId} がリソースから離れました\n";
        Semaphore::post($this->semaphore); // ロックを解放
    }
}

class ResourceThread extends Thread {
    private $resource;
    private $threadId;

    public function __construct($resource, $id) {
        $this->resource = $resource;
        $this->threadId = $id;
    }

    public function run() {
        $this->resource->accessResource($this->threadId);
    }
}

// リソースへの同時アクセス数を2に設定
$resource = new LimitedResource(2);
$threads = [];

// 5つのスレッドでリソースにアクセス
for ($i = 0; $i < 5; $i++) {
    $threads[$i] = new ResourceThread($resource, $i + 1);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

echo "リソースへのすべてのアクセスが完了しました\n";
?>

<?php
class Task extends Thread {
    private $taskId;

    public function __construct($taskId) {
        $this->taskId = $taskId;
    }

    public function run() {
        // タスクの実行内容
        echo "タスク {$this->taskId} が開始されました\n";
        sleep(2); // タスクの処理にかかる時間をシミュレート
        echo "タスク {$this->taskId} が完了しました\n";
    }
}

class ThreadPool {
    private $poolSize;
    private $tasks;
    private $threads;

    public function __construct($poolSize) {
        $this->poolSize = $poolSize;
        $this->tasks = [];
        $this->threads = [];
    }

    public function addTask($task) {
        $this->tasks[] = $task;
    }

    public function execute() {
        // タスクがなくなるまで実行
        while (!empty($this->tasks)) {
            // スレッド数がプールサイズ未満なら新しいスレッドを作成して実行
            if (count($this->threads) < $this->poolSize) {
                $task = array_shift($this->tasks);
                $task->start();
                $this->threads[] = $task;
            }

            // 完了したスレッドを削除
            foreach ($this->threads as $key => $thread) {
                if (!$thread->isRunning()) {
                    $thread->join(); // スレッドの終了を待機
                    unset($this->threads[$key]);
                }
            }
        }

        // 残っているスレッドの終了を待つ
        foreach ($this->threads as $thread) {
            $thread->join();
        }
    }
}

// スレッドプールを作成
$pool = new ThreadPool(3); // 同時に実行するスレッドの数を3に設定

// タスクを追加
for ($i = 1; $i <= 10; $i++) {
    $pool->addTask(new Task($i));
}

// タスクを実行
$pool->execute();

echo "すべてのタスクが完了しました\n";
?>

<?php
class ErrorProneTask extends Thread {
    private $taskId;

    public function __construct($taskId) {
        $this->taskId = $taskId;
    }

    public function run() {
        try {
            // 意図的にエラーを発生させる
            if ($this->taskId % 2 == 0) {
                throw new Exception("タスク {$this->taskId} でエラーが発生しました");
            }
            echo "タスク {$this->taskId} は正常に終了しました\n";
        } catch (Exception $e) {
            // 例外をキャッチしてエラーメッセージを表示
            echo "エラー: " . $e->getMessage() . "\n";
        }
    }
}

$threads = [];

// 5つのタスクを実行
for ($i = 1; $i <= 5; $i++) {
    $threads[$i] = new ErrorProneTask($i);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

echo "すべてのタスクが終了しました\n";
?>

<?php
class SharedErrorLog extends Threaded {
    public function logError($message) {
        $this->synchronized(function() use ($message) {
            $this[] = $message;
        });
    }

    public function getErrors() {
        return $this;
    }
}

class LoggingTask extends Thread {
    private $taskId;
    private $errorLog;

    public function __construct($taskId, $errorLog) {
        $this->taskId = $taskId;
        $this->errorLog = $errorLog;
    }

    public function run() {
        try {
            // エラーを発生させる
            if ($this->taskId % 2 == 0) {
                throw new Exception("タスク {$this->taskId} でエラーが発生しました");
            }
            echo "タスク {$this->taskId} は正常に完了しました\n";
        } catch (Exception $e) {
            // エラーログにメッセージを追加
            $this->errorLog->logError($e->getMessage());
        }
    }
}

$errorLog = new SharedErrorLog();
$threads = [];

// タスクを5つ実行
for ($i = 1; $i <= 5; $i++) {
    $threads[$i] = new LoggingTask($i, $errorLog);
    $threads[$i]->start();
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

// 共有エラーログの表示
$errors = $errorLog->getErrors();
if (count($errors) > 0) {
    echo "発生したエラー:\n";
    foreach ($errors as $error) {
        echo "- $error\n";
    }
} else {
    echo "エラーは発生しませんでした\n";
}
?>

<?php
class WebScraper extends Thread {
    private $url;
    private $result;

    public function __construct($url) {
        $this->url = $url;
    }

    public function run() {
        // ページの内容を取得
        $this->result = file_get_contents($this->url);
        echo "URL: {$this->url} の取得が完了しました\n";
    }

    public function getResult() {
        return $this->result;
    }
}

$urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    // ...他のURL
];

$threads = [];

// 各URLに対してスレッドを作成して実行
foreach ($urls as $url) {
    $thread = new WebScraper($url);
    $thread->start();
    $threads[] = $thread;
}

// 全てのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
    // 結果の取得
    $result = $thread->getResult();
    echo "スクレイピング結果のサイズ: " . strlen($result) . " バイト\n";
}

echo "全てのWebスクレイピングが完了しました\n";
?>

<?php
class ImageProcessor extends Thread {
    private $imagePath;
    private $outputPath;

    public function __construct($imagePath, $outputPath) {
        $this->imagePath = $imagePath;
        $this->outputPath = $outputPath;
    }

    public function run() {
        // 画像の読み込み
        $image = imagecreatefromjpeg($this->imagePath);
        if ($image === false) {
            echo "画像の読み込みに失敗しました: {$this->imagePath}\n";
            return;
        }

        // 画像のリサイズ
        $width = imagesx($image);
        $height = imagesy($image);
        $newWidth = $width / 2;
        $newHeight = $height / 2;
        $resizedImage = imagescale($image, $newWidth, $newHeight);

        // 画像の保存
        imagejpeg($resizedImage, $this->outputPath);
        imagedestroy($image);
        imagedestroy($resizedImage);

        echo "画像の処理が完了しました: {$this->outputPath}\n";
    }
}

$imagePaths = [
    "images/photo1.jpg",
    "images/photo2.jpg",
    "images/photo3.jpg",
    // ...他の画像ファイル
];

$threads = [];

// 画像ごとにスレッドを作成して実行
foreach ($imagePaths as $index => $path) {
    $outputPath = "output/resized_{$index}.jpg";
    $thread = new ImageProcessor($path, $outputPath);
    $thread->start();
    $threads[] = $thread;
}

// 全てのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
}

echo "すべての画像処理が完了しました\n";
?>

<?php
class DatabaseQuery extends Thread {
    private $query;
    private $result;

    public function __construct($query) {
        $this->query = $query;
    }

    public function run() {
        // データベース接続（例としてPDOを使用）
        $pdo = new PDO('mysql:host=localhost;dbname=testdb', 'username', 'password');
        $statement = $pdo->query($this->query);
        $this->result = $statement->fetchAll(PDO::FETCH_ASSOC);
        echo "クエリの実行が完了しました: {$this->query}\n";
    }

    public function getResult() {
        return $this->result;
    }
}

$queries = [
    "SELECT * FROM users WHERE age > 30",
    "SELECT * FROM orders WHERE status = 'pending'",
    "SELECT * FROM products WHERE stock < 10",
    // ...他のクエリ
];

$threads = [];

// クエリごとにスレッドを作成して実行
foreach ($queries as $query) {
    $thread = new DatabaseQuery($query);
    $thread->start();
    $threads[] = $thread;
}

// すべてのスレッドの終了を待機
foreach ($threads as $thread) {
    $thread->join();
    // 結果を取得して表示
    $result = $thread->getResult();
    echo "取得されたデータ: " . print_r($result, true) . "\n";
}

echo "すべてのクエリの実行が完了しました\n";
?>

<?php
require 'vendor/autoload.php';

use React\EventLoop\Factory;
use React\Http\Browser;

$loop = Factory::create();
$client = new Browser($loop);

$urls = [
    'https://example.com/page1',
    'https://example.com/page2',
    'https://example.com/page3',
];

// 各URLに対して非同期リクエストを送信
foreach ($urls as $url) {
    $client->get($url)->then(
        function (Psr\Http\Message\ResponseInterface $response) use ($url) {
            echo "URL: $url のデータサイズ: " . strlen((string) $response->getBody()) . " バイト\n";
        },
        function (Exception $e) {
            echo "エラーが発生しました: " . $e->getMessage() . "\n";
        }
    );
}

$loop->run();
?>

<?php
use parallel\Runtime;

$runtimes = [];
$results = [];

// 複数のランタイムを作成して並列に処理
for ($i = 0; $i < 3; $i++) {
    $runtimes[$i] = new Runtime();
    $results[$i] = $runtimes[$i]->run(function () use ($i) {
        return "タスク {$i} の結果";
    });
}

// 結果を取得して表示
foreach ($results as $result) {
    echo $result->value() . "\n";
}
?>

<?php
$processes = [];

for ($i = 0; $i < 3; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die('プロセスの作成に失敗しました');
    } elseif ($pid) {
        // 親プロセス
        $processes[] = $pid;
    } else {
        // 子プロセス
        echo "プロセス {$i} が開始されました\n";
        sleep(2);
        echo "プロセス {$i} が終了します\n";
        exit(0);
    }
}

// 子プロセスの終了を待機
foreach ($processes as $pid) {
    pcntl_waitpid($pid, $status);
}

echo "すべてのプロセスが終了しました\n";
?>

// ジョブのキュー投入例
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

$data = "Hello, World!";
$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($msg, '', 'task_queue');

echo "ジョブをキューに投入しました\n";

$channel->close();
$connection->close();
?>