如何用PHP Ratchet实现关联手机号的WebSocket服务器管理Socket？

架构老林

2026-7-7

解决方案：用Ratchet实现手机号与Amazon Connect Socket的绑定转发

核心架构思路

放弃Webhook直接创建Amazon Connect Socket的方式，改用独立运行的Ratchet服务器作为中间层：

Ratchet维护全局连接池，每个手机号对应一个持久的Amazon Connect Socket连接
Webhook仅负责接收WhatsApp消息，通过HTTP接口将消息转发给Ratchet
Ratchet根据手机号匹配对应Socket，将消息转发给Amazon Connect；同时监听Amazon Connect的返回消息，再转发回WhatsApp

实现步骤与代码示例

1. 创建Ratchet连接管理类

这个类负责维护手机号与Amazon Connect连接的映射，处理双向消息转发：

use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
use React\Socket\Connector;
use React\Promise\PromiseInterface;

class AmazonConnectGateway implements MessageComponentInterface {
    // 存储手机号到Amazon Connect连接的映射
    private array $phoneToAcConn = [];
    // 连接互斥锁，避免并发创建重复连接
    private \React\Mutex\Mutex $connMutex;

    public function __construct() {
        $this->connMutex = new \React\Mutex\Mutex();
    }

    // 获取或创建对应手机号的Amazon Connect连接
    public function getOrCreateAcConnection(string $phoneNumber): PromiseInterface|null {
        if (isset($this->phoneToAcConn[$phoneNumber])) {
            return \React\Promise\resolve($this->phoneToAcConn[$phoneNumber]);
        }

        return $this->connMutex->lock()->then(function () use ($phoneNumber) {
            // 再次检查，避免锁等待期间已创建连接
            if (isset($this->phoneToAcConn[$phoneNumber])) {
                return $this->phoneToAcConn[$phoneNumber];
            }

            $connector = new Connector();
            return $connector->connect('tcp://your-amazon-connect-socket:port')->then(function ($acConn) use ($phoneNumber) {
                $this->phoneToAcConn[$phoneNumber] = $acConn;

                // 监听Amazon Connect消息，转发回WhatsApp
                $acConn->on('data', function ($data) use ($phoneNumber) {
                    $this->sendToWhatsApp($phoneNumber, $data);
                });

                // 连接关闭时清理映射
                $acConn->on('close', function () use ($phoneNumber) {
                    unset($this->phoneToAcConn[$phoneNumber]);
                });

                // 心跳保活（根据Amazon Connect要求调整间隔）
                $heartbeatInterval = \React\EventLoop\Loop::get()->addPeriodicTimer(30, function () use ($acConn, $phoneNumber) {
                    if (!$acConn->isReadable()) {
                        unset($this->phoneToAcConn[$phoneNumber]);
                        \React\EventLoop\Loop::get()->cancelTimer($this);
                        return;
                    }
                    $acConn->write('PING');
                });

                return $acConn;
            });
        })->finally(function () {
            $this->connMutex->unlock();
        });
    }

    // 转发WhatsApp消息到Amazon Connect
    public function forwardToAc(string $phoneNumber, string $message): void {
        $this->getOrCreateAcConnection($phoneNumber)->then(function ($acConn) use ($message) {
            $acConn->write($message);
        }, function ($error) {
            error_log("转发失败: {$error->getMessage()}");
        });
    }

    // 调用WhatsApp API发送消息
    private function sendToWhatsApp(string $phoneNumber, string $message): void {
        // 替换为实际WhatsApp API调用逻辑
        $payload = json_encode([
            'to' => $phoneNumber,
            'text' => ['body' => $message]
        ]);

        $ch = curl_init('https://graph.facebook.com/v18.0/your-whatsapp-number/messages');
        curl_setopt($ch, CURLOPT_HTTPHEADER, [
            'Content-Type: application/json',
            'Authorization: Bearer your-access-token'
        ]);
        curl_setopt($ch, CURLOPT_POST, true);
        curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
        curl_exec($ch);
        curl_close($ch);
    }

    // MessageComponentInterface 空实现（若无需外部WebSocket客户端可忽略）
    public function onOpen(ConnectionInterface $conn) {}
    public function onMessage(ConnectionInterface $from, $msg) {}
    public function onClose(ConnectionInterface $conn) {}
    public function onError(ConnectionInterface $conn, \Exception $e) {}
}

2. 启动Ratchet服务器与HTTP接口

创建启动脚本，同时运行WebSocket服务（可选）和HTTP接口供Webhook调用：

use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use React\Http\HttpServer as ReactHttpServer;
use React\Socket\SocketServer;

require __DIR__ . '/vendor/autoload.php';

$gateway = new AmazonConnectGateway();

// 启动HTTP服务器，接收Webhook转发请求
$httpServer = new ReactHttpServer(function ($request, $response) use ($gateway) {
    if ($request->getMethod() !== 'POST' || $request->getPath() !== '/forward') {
        return $response->withStatus(404)->write('Not Found');
    }

    $body = (string)$request->getBody();
    $data = json_decode($body, true);

    if (!isset($data['phone'], $data['message'])) {
        return $response->withStatus(400)->write('Missing phone or message');
    }

    $gateway->forwardToAc($data['phone'], $data['message']);
    return $response->withStatus(200)->write('Forwarded');
});

$httpSocket = new SocketServer('0.0.0.0:8081');
$httpServer->listen($httpSocket);

// 启动WebSocket服务器（若需要外部客户端连接可保留）
$wsServer = IoServer::factory(
    new HttpServer(new WsServer($gateway)),
    8080
);

// 启动事件循环
\React\EventLoop\Loop::get()->run();

3. 修改WhatsApp Webhook脚本

Webhook仅需接收消息，然后调用Ratchet的HTTP接口转发：

// WhatsApp Webhook接收脚本
$input = file_get_contents('php://input');
$data = json_decode($input, true);

// 提取发送方手机号和消息内容（根据WhatsApp Webhook格式调整）
$phone = $data['entry'][0]['changes'][0]['value']['messages'][0]['from'];
$message = $data['entry'][0]['changes'][0]['value']['messages'][0]['text']['body'];

// 调用Ratchet HTTP接口转发
$ch = curl_init('http://localhost:8081/forward');
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode([
    'phone' => $phone,
    'message' => $message
]));
curl_setopt($ch, CURLOPT_HTTPHEADER, ['Content-Type: application/json']);
curl_exec($ch);
curl_close($ch);

echo 'OK';

关键注意事项

连接保活：添加心跳机制避免Amazon Connect Socket超时断开
并发安全：用互斥锁防止同一手机号重复创建连接
错误重试：当Amazon Connect连接断开时，下次消息会自动重建连接
日志监控：添加详细日志，方便排查连接或转发失败问题

内容的提问来源于stack exchange，提问作者Marco