{"id":1166,"date":"2025-07-12T17:18:13","date_gmt":"2025-07-12T09:18:13","guid":{"rendered":"https:\/\/www.zhaozhao123.cn\/php\/?p=1166"},"modified":"2025-07-12T17:18:13","modified_gmt":"2025-07-12T09:18:13","slug":"php%e6%b6%88%e6%81%af%e9%98%9f%e5%88%97%e5%ae%9e%e7%8e%b0%e6%96%b9%e6%a1%88%e8%af%a6%e8%a7%a3","status":"publish","type":"post","link":"https:\/\/www.zhaozhao123.cn\/php\/post\/1166.html","title":{"rendered":"PHP\u6d88\u606f\u961f\u5217\u5b9e\u73b0\u65b9\u6848\u8be6\u89e3"},"content":{"rendered":"\n

\u5728PHP\u4e2d\u5b9e\u73b0\u6d88\u606f\u961f\u5217\uff08Message Queue\uff09\u6709\u591a\u79cd\u65b9\u5f0f\uff0c\u4e0b\u9762\u4ecb\u7ecd\u51e0\u79cd\u5e38\u89c1\u7684\u5b9e\u73b0\u65b9\u6848\uff0c\u4ece\u7b80\u5355\u5230\u590d\u6742\u9010\u6b65\u5c55\u5f00\uff1a<\/p>\n\n\n\n

\n\n\n\n

\u4e00\u3001\u57fa\u4e8e\u6570\u636e\u5e93\u7684\u7b80\u6613\u961f\u5217\uff08\u9002\u5408\u8f7b\u91cf\u7ea7\u573a\u666f\uff09<\/h3>\n\n\n\n

\u539f\u7406<\/strong>\uff1a\u5229\u7528\u6570\u636e\u5e93\u8868\u4f5c\u4e3a\u961f\u5217\u5b58\u50a8
\u8868\u7ed3\u6784\u793a\u4f8b<\/strong>\uff1a<\/p>\n\n\n\n

CREATE TABLE `message_queue` (\n  `id` INT AUTO_INCREMENT,\n  `queue_name` VARCHAR(50) NOT NULL DEFAULT 'default',\n  `payload` TEXT NOT NULL,        -- \u5b58\u50a8JSON\u683c\u5f0f\u7684\u4efb\u52a1\u6570\u636e\n  `status` TINYINT NOT NULL DEFAULT 0, -- 0=\u7b49\u5f85, 1=\u5904\u7406\u4e2d, 2=\u5b8c\u6210\n  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB;<\/code><\/pre>\n\n\n\n
\u751f\u4ea7\u8005\u4ee3\u7801<\/strong>\uff1a<\/p>\n\n\n\n
\/\/ \u751f\u4ea7\u8005 producer.php\n$db = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');\n$payload = json_encode(['task' => 'send_email', 'email' => 'user@example.com']);\n\n$stmt = $db->prepare(\"INSERT INTO message_queue (queue_name, payload) VALUES (?, ?)\");\n$stmt->execute(['email_queue', $payload]);<\/code><\/pre>\n\n\n\n
\u6d88\u8d39\u8005\u4ee3\u7801<\/strong>\uff1a<\/p>\n\n\n\n
\/\/ \u6d88\u8d39\u8005 worker.php\n$db = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');\n$db->beginTransaction();\n\n\/\/ \u9501\u5b9a\u5e76\u83b7\u53d6\u4efb\u52a1\uff08\u907f\u514d\u591a\u8fdb\u7a0b\u91cd\u590d\u6d88\u8d39\uff09\n$stmt = $db->prepare(\"SELECT * FROM message_queue \n                     WHERE status = 0 AND queue_name = ?\n                     ORDER BY id ASC LIMIT 1 FOR UPDATE SKIP LOCKED\");\n$stmt->execute(['email_queue']);\n$task = $stmt->fetch(PDO::FETCH_ASSOC);\n\nif ($task) {\n    \/\/ \u6807\u8bb0\u4e3a\u5904\u7406\u4e2d\n    $db->prepare(\"UPDATE message_queue SET status = 1 WHERE id = ?\")\n       ->execute([$task['id']]);\n    $db->commit();\n\n    \/\/ \u5904\u7406\u4efb\u52a1\n    $data = json_decode($task['payload'], true);\n    send_email($data['email']); \/\/ \u5b9e\u9645\u4e1a\u52a1\u903b\u8f91\n\n    \/\/ \u6807\u8bb0\u5b8c\u6210\n    $db->prepare(\"UPDATE message_queue SET status = 2 WHERE id = ?\")\n       ->execute([$task['id']]);\n} else {\n    $db->rollBack();\n    sleep(1); \/\/ \u65e0\u4efb\u52a1\u65f6\u4f11\u7720\n}<\/code><\/pre>\n\n\n\n
\n\n\n\n
\u4e8c\u3001\u4f7f\u7528 Redis \u961f\u5217\uff08\u63a8\u8350\uff0c\u9ad8\u6027\u80fd\uff09<\/h3>\n\n\n\n
\u4f18\u52bf<\/strong>\uff1a\u5185\u5b58\u64cd\u4f5c\uff0c\u652f\u6301\u963b\u585e\u5f39\u51fa\uff0c\u539f\u5b50\u6027\u64cd\u4f5c
\u9700\u8981\u5b89\u88c5<\/strong>\uff1aphp-redis<\/code> \u6269\u5c55<\/p>\n\n\n\n
\u751f\u4ea7\u8005<\/h4>\n\n\n\n
\/\/ producer.php\n$redis = new Redis();\n$redis->connect('127.0.0.1', 6379);\n\n$task = [\n    'type' => 'log_cleanup',\n    'path' => '\/var\/logs'\n];\n\n$redis->lPush('work_queue', json_encode($task)); \/\/ \u5de6\u8fdb<\/code><\/pre>\n\n\n\n
\u6d88\u8d39\u8005\uff08\u963b\u585e\u7248\uff09<\/h4>\n\n\n\n
\/\/ worker.php\n$redis = new Redis();\n$redis->connect('127.0.0.1', 6379);\n\nwhile (true) {\n    \/\/ \u963b\u585e\u53f3\u51fa\uff080\u8868\u793a\u65e0\u9650\u7b49\u5f85\uff09\n    $taskJson = $redis->brPop(['work_queue'], 0)[1]; \n    $task = json_decode($taskJson, true);\n\n    \/\/ \u6267\u884c\u4efb\u52a1\n    cleanup_logs($task['path']); \/\/ \u4e1a\u52a1\u903b\u8f91\n}<\/code><\/pre>\n\n\n\n
\n\n\n\n
\u4e09\u3001\u4e13\u4e1a\u6d88\u606f\u961f\u5217\u7cfb\u7edf\uff08\u751f\u4ea7\u73af\u5883\u63a8\u8350\uff09<\/h3>\n\n\n\n
\u65b9\u68481\uff1aRabbitMQ + PHP AMQP \u6269\u5c55<\/h4>\n\n\n\n
\u5b89\u88c5<\/strong>\uff1a<\/p>\n\n\n\n
# \u5b89\u88c5\u6269\u5c55\npecl install amqp<\/code><\/pre>\n\n\n\n
\u751f\u4ea7\u8005<\/strong>\uff1a<\/p>\n\n\n\n
$connection = new AMQPConnection(['host' => 'localhost']);\n$connection->connect();\n$channel = $connection->channel();\n\n\/\/ \u58f0\u660e\u961f\u5217\n$channel->queue_declare('task_queue', false, true, false, false);\n\n\/\/ \u521b\u5efa\u6d88\u606f\n$msg = new AMQPMessage(json_encode(['data' => '...']), [\n    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT\n]);\n\n\/\/ \u53d1\u5e03\n$channel->basic_publish($msg, '', 'task_queue');\n$channel->close();<\/code><\/pre>\n\n\n\n
\u6d88\u8d39\u8005<\/strong>\uff1a<\/p>\n\n\n\n
$callback = function ($msg) {\n    $task = json_decode($msg->body, true);\n    process_task($task); \/\/ \u4e1a\u52a1\u5904\u7406\n    $msg->ack(); \/\/ \u624b\u52a8\u786e\u8ba4\n};\n\n$channel->basic_consume('task_queue', '', false, false, false, false, $callback);\n\nwhile (count($channel->callbacks)) {\n    $channel->wait();\n}<\/code><\/pre>\n\n\n\n
\u65b9\u68482\uff1aBeanstalkd + Pheanstalk \u5e93<\/h4>\n\n\n\n
\u5b89\u88c5<\/strong>\uff1a<\/p>\n\n\n\n
composer require pda\/pheanstalk<\/code><\/pre>\n\n\n\n
\u751f\u4ea7\u8005<\/strong>\uff1a<\/p>\n\n\n\n
$pheanstalk = Pheanstalk\\Pheanstalk::create('127.0.0.1');\n$pheanstalk->useTube('email_tube')\n           ->put(json_encode(['to' => 'user@example.com']));<\/code><\/pre>\n\n\n\n
\u6d88\u8d39\u8005<\/strong>\uff1a<\/p>\n\n\n\n
$pheanstalk->watch('email_tube');\n\nwhile ($job = $pheanstalk->reserve()) { \/\/ \u963b\u585e\u83b7\u53d6\n    $data = json_decode($job->getData(), true);\n    send_email($data['to']);\n    $pheanstalk->delete($job); \/\/ \u786e\u8ba4\u5220\u9664\n}<\/code><\/pre>\n\n\n\n
\n\n\n\n
\u56db\u3001\u961f\u5217\u7ba1\u7406\u5efa\u8bae<\/h3>\n\n\n\n
    \n
  1. \u5b88\u62a4\u8fdb\u7a0b<\/strong>\uff1a\u4f7f\u7528 Supervisor \u7ba1\u7406\u6d88\u8d39\u8005\u8fdb\u7a0b<\/li>\n<\/ol>\n\n\n\n
       [program:queue_worker]\n   command=php \/path\/to\/worker.php\n   autorestart=true\n   numprocs=4 ; \u542f\u52a84\u4e2a\u8fdb\u7a0b<\/code><\/pre>\n\n\n\n
      \n
    1. \u5931\u8d25\u91cd\u8bd5<\/strong>\uff1a<\/li>\n<\/ol>\n\n\n\n
        \n
      • Redis\uff1a\u7528\u6709\u5e8f\u96c6\u5408\uff08ZSET\uff09\u8bb0\u5f55\u5931\u8d25\u4efb\u52a1+\u91cd\u8bd5\u65f6\u95f4<\/li>\n\n\n\n
      • RabbitMQ\uff1a\u5f00\u542f\u6b7b\u4fe1\u961f\u5217\uff08Dead Letter Exchange\uff09<\/li>\n<\/ul>\n\n\n\n
          \n
        1. \u76d1\u63a7<\/strong>\uff1a<\/li>\n<\/ol>\n\n\n\n
            \n
          • RabbitMQ\uff1a\u81ea\u5e26\u7ba1\u7406\u754c\u9762<\/li>\n\n\n\n
          • Redis\uff1aredis-cli monitor<\/code> \u6216 RedisInsight<\/li>\n<\/ul>\n\n\n\n
              \n
            1. \u5e8f\u5217\u5316<\/strong>\uff1a\u63a8\u8350\u4f7f\u7528 JSON\uff08\u53ef\u8bfb\u6027\u597d\uff09\u6216 MsgPack\uff08\u9ad8\u6548\uff09<\/li>\n<\/ol>\n\n\n\n
              \n\n\n\n
              \u65b9\u6848\u5bf9\u6bd4<\/h3>\n\n\n\n
              \u65b9\u6848<\/th>\u4f18\u70b9<\/th>\u7f3a\u70b9<\/th>\u9002\u7528\u573a\u666f<\/th><\/tr><\/thead>
              \u6570\u636e\u5e93\u961f\u5217<\/td>\u65e0\u9700\u65b0\u7ec4\u4ef6\uff0c\u4e8b\u52a1\u652f\u6301<\/td>\u6027\u80fd\u5dee\uff0c\u9501\u7ade\u4e89\u4e25\u91cd<\/td>\u4f4e\u541e\u5410\u91cf\uff08<10TPS\uff09<\/td><\/tr>
              Redis<\/td>\u9ad8\u6027\u80fd\uff0c\u7b80\u5355\u6613\u7528<\/td>\u65e0\u6301\u4e45\u5316\uff08\u9700\u914d\u7f6eAOF\uff09<\/td>\u4e2d\u5c0f\u9879\u76ee\uff0c\u5b9e\u65f6\u6027\u8981\u6c42\u9ad8<\/td><\/tr>
              RabbitMQ<\/td>\u529f\u80fd\u5b8c\u6574\uff0c\u4f01\u4e1a\u7ea7\u7279\u6027<\/td>\u9700\u8981\u7ef4\u62a4Erlang\u73af\u5883<\/td>\u590d\u6742\u4e1a\u52a1\uff0c\u751f\u4ea7\u73af\u5883<\/td><\/tr>
              Beanstalkd<\/td>\u8f7b\u91cf\u7ea7\uff0c\u534f\u8bae\u7b80\u5355<\/td>\u793e\u533a\u8f83\u5c0f<\/td>\u5feb\u901f\u90e8\u7f72\uff0cPHP\u53cb\u597d<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n
              \n
              \u751f\u4ea7\u73af\u5883\u5efa\u8bae\uff1a\u4f18\u5148\u9009\u62e9 RabbitMQ \u6216 Redis\uff08\u5f00\u542f\u6301\u4e45\u5316\uff09\uff0c\u914d\u5408 Supervisor \u505a\u8fdb\u7a0b\u7ba1\u7406\u3002<\/p>\n<\/blockquote>\n\n\n\n
              <\/p>\n","protected":false},"excerpt":{"rendered":"
              \u5728PHP\u4e2d\u5b9e\u73b0\u6d88\u606f\u961f\u5217\uff08Message Queue\uff09\u6709\u591a\u79cd\u65b9\u5f0f\uff0c\u4e0b\u9762\u4ecb\u7ecd\u51e0\u79cd\u5e38\u89c1\u7684\u5b9e\u73b0\u65b9\u6848\uff0c\u4ece\u7b80\u5355\u5230\u590d\u6742\u9010\u6b65\u5c55\u5f00\uff1a \u4e00\u3001\u57fa\u4e8e\u6570\u636e\u5e93\u7684\u7b80\u6613\u961f\u5217\uff08\u9002\u5408\u8f7b\u91cf\u7ea7\u573a\u666f\uff09 \u539f\u7406\uff1a\u5229\u7528\u6570\u636e\u5e93\u8868\u4f5c\u4e3a\u961f\u5217\u5b58\u50a8\u8868\u7ed3\u6784\u793a\u4f8b\uff1a \u751f\u4ea7\u8005\u4ee3\u7801\uff1a \u6d88\u8d39\u8005\u4ee3\u7801\uff1a \u4e8c\u3001\u4f7f\u7528 Redis \u961f\u5217\uff08\u63a8\u8350\uff0c\u9ad8\u6027\u80fd\uff09 \u4f18\u52bf\uff1a\u5185\u5b58\u64cd\u4f5c\uff0c\u652f\u6301\u963b\u585e\u5f39\u51fa\uff0c\u539f\u5b50\u6027\u64cd\u4f5c\u9700\u8981..<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[1],"tags":[],"tuisongtax":[],"class_list":["post-1166","post","type-post","status-publish","format-standard","hentry","category-jswz"],"acf":[],"_links":{"self":[{"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/posts\/1166","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/comments?post=1166"}],"version-history":[{"count":0,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/posts\/1166\/revisions"}],"wp:attachment":[{"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/media?parent=1166"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/categories?post=1166"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/tags?post=1166"},{"taxonomy":"tuisongtax","embeddable":true,"href":"https:\/\/www.zhaozhao123.cn\/php\/wp-json\/wp\/v2\/tuisongtax?post=1166"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}