PHP 與 Java Socket 通信實戰：TCP 長連接、協議設計與生產級客戶端封裝

前言

Socket 是跨語言進程通信的基石。PHP 作為客戶端，Java 作為服務端，通過 TCP 長連接交換自定義協議數據，適用於：高性能內部 RPC、推送服務、網關轉發、舊系統集成。本文提供完整雙端代碼、協議設計模式、斷線重連、心跳保活、線程池模型與生產級 PHP 客戶端封裝。

協議設計：TLV (Type-Length-Value) 二進制協議

比文本協議（JSON/HTTP）更緊湊、解析更快、無歧義。

+--------+--------+--------+------------------+
| Magic  | Type   | Length | Body             |
| (2B)   | (1B)   | (4B)   | (Length Bytes)   |
+--------+--------+--------+------------------+
Magic: 0xAB 0xCD (固定頭，識別協議)
Type:  0x01=請求, 0x02=響應, 0x03=心跳, 0x04=推送
Length: Big-Endian 無符號整數 (網絡序)
Body:  業務負載 (JSON/Protobuf/MessagePack)

Java 服務端實現 (Netty 4.x - 生產級推薦)

// 依賴: io.netty:netty-all:4.1.x
public class SocketServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup boss = new NioEventLoopGroup(1);
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(boss, worker)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(
                         new LengthFieldBasedFrameDecoder(65536, 3, 4, 0, 7), // Magic(2)+Type(1)+Length(4)
                         new TlvDecoder(),      // 解碼為 TlvMessage
                         new TlvEncoder(),      // 編碼 TlvMessage
                         new BusinessHandler()  // 業務邏輯
                     );
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)
             .childOption(ChannelOption.SO_KEEPALIVE, true);
            
            ChannelFuture f = b.bind(5678).sync();
            System.out.println("Socket Server started on port 5678");
            f.channel().closeFuture().sync();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

// 業務處理器
@ChannelHandler.Sharable
class BusinessHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlvMessage msg) {
        if (msg.getType() == 0x01) { // 請求
            String body = new String(msg.getBody(), StandardCharsets.UTF_8);
            System.out.println("收到請求: " + body);
            
            // 業務處理 (示例: 原樣返回)
            TlvMessage resp = new TlvMessage((byte)0x02, body.getBytes(StandardCharsets.UTF_8));
            ctx.writeAndFlush(resp);
        } else if (msg.getType() == 0x03) { // 心跳
            ctx.writeAndFlush(new TlvMessage((byte)0x03, new byte[0])); // 回心跳
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

PHP 客戶端封裝 (支持長連接、重連、心跳、異步回調)

<?php
class TcpClient {
    private $host;
    private $port;
    private $socket;
    private $connected = false;
    private $buffer = '';
    private $callbacks = []; // requestId => callback
    private $requestId = 0;
    private $heartbeatInterval = 30; // 秒
    private $lastHeartbeat = 0;
    
    // TLV 常量
    const MAGIC = "\xAB\xCD";
    const TYPE_REQUEST = 0x01;
    const TYPE_RESPONSE = 0x02;
    const TYPE_HEARTBEAT = 0x03;
    
    public function __construct(string $host, int $port) {
        $this->host = $host;
        $this->port = $port;
    }
    
    // 連接
    public function connect(float $timeout = 5.0): bool {
        $this->socket = @stream_socket_client(
            "tcp://{$this->host}:{$this->port}",
            $errno, $errstr, $timeout,
            STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT // 長連接
        );
        if (!$this->socket) {
            throw new RuntimeException("Connect failed: $errstr ($errno)");
        }
        stream_set_blocking($this->socket, 0); // 非阻塞
        stream_set_timeout($this->socket, 5);  // 讀寫超時
        $this->connected = true;
        $this->lastHeartbeat = time();
        return true;
    }
    
    // 發送 TLV 數據
    public function send(int $type, string $body): int {
        if (!$this->connected) $this->connect();
        
        $length = strlen($body);
        $header = self::MAGIC . chr($type) . pack('N', $length); // N = Big-Endian 32bit
        $packet = $header . $body;
        
        $written = @fwrite($this->socket, $packet);
        if ($written === false || $written !== strlen($packet)) {
            $this->connected = false;
            throw new RuntimeException('Send failed');
        }
        return $written;
    }
    
    // 發送請求並註冊回調 (異步模式)
    public function request(string $body, callable $callback): int {
        $requestId = ++$this->requestId;
        $this->callbacks[$requestId] = $callback;
        // 協議擴展：在 Body 前加入 RequestId (4字節)
        $payload = pack('N', $requestId) . $body;
        $this->send(self::TYPE_REQUEST, $payload);
        return $requestId;
    }
    
    // 發送心跳
    public function heartbeat(): void {
        $this->send(self::TYPE_HEARTBEAT, '');
        $this->lastHeartbeat = time();
    }
    
    // 事件循環 (需外部調用，如 Swoole/Workerman/定時任務中)
    public function loop(): void {
        if (!$this->connected) return;
        
        // 1. 讀取數據
        $read = [$this->socket];
        $write = $except = null;
        if (stream_select($read, $write, $except, 0, 100000) > 0) { // 100ms
            $data = fread($this->socket, 65536);
            if ($data === false || $data === '') {
                $this->connected = false;
                return;
            }
            $this->buffer .= $data;
            $this->parseBuffer();
        }
        
        // 2. 心跳檢查
        if (time() - $this->lastHeartbeat > $this->heartbeatInterval) {
            $this->heartbeat();
        }
    }
    
    // 解析緩衝區完整包
    private function parseBuffer(): void {
        while (strlen($this->buffer) >= 7) { // 最小包頭長度
            // 檢查 Magic
            if (substr($this->buffer, 0, 2) !== self::MAGIC) {
                // 協議不同步，丟棄 1 字節嘗試重新同步 (簡單策略)
                $this->buffer = substr($this->buffer, 1);
                continue;
            }
            
            $type = ord($this->buffer[2]);
            $length = unpack('N', substr($this->buffer, 3, 4))[1];
            $totalLen = 7 + $length;
            
            if (strlen($this->buffer) < $totalLen) break; // 包不完整
            
            $body = substr($this->buffer, 7, $length);
            $this->buffer = substr($this->buffer, $totalLen);
            
            $this->handleMessage($type, $body);
        }
    }
    
    private function handleMessage(int $type, string $body): void {
        switch ($type) {
            case self::TYPE_RESPONSE:
                // 前 4 字節為 RequestId
                if (strlen($body) >= 4) {
                    $requestId = unpack('N', substr($body, 0, 4))[1];
                    $payload = substr($body, 4);
                    if (isset($this->callbacks[$requestId])) {
                        $cb = $this->callbacks[$requestId];
                        unset($this->callbacks[$requestId]);
                        $cb($payload, null);
                    }
                }
                break;
            case self::TYPE_HEARTBEAT:
                // 收到心跳響應，更新時間
                $this->lastHeartbeat = time();
                break;
            case 0x04: // 推送
                // 觸發全局推送回調
                if (isset($this->callbacks['push'])) {
                    $this->callbacks['push']($body);
                }
                break;
        }
    }
    
    public function onPush(callable $callback): void {
        $this->callbacks['push'] = $callback;
    }
    
    public function close(): void {
        if ($this->socket) {
            fclose($this->socket);
            $this->socket = null;
            $this->connected = false;
        }
    }
    
    public function __destruct() {
        $this->close();
    }
}

PHP 使用示例 (CLI 長運行腳本 / Swoole Task / Workerman)

<?php
require 'TcpClient.php';

$client = new TcpClient('127.0.0.1', 5678);
$client->connect();

// 同步請求示例 (簡化：發送後循環等待響應)
$client->request(json_encode(['action' => 'getUser', 'uid' => 1001]), function($data, $err) {
    echo "異步響應: $data\n";
    $user = json_decode($data, true);
    var_dump($user);
});

// 推送回調
$client->onPush(function($data) {
    echo "收到推送: $data\n";
});

// 事件循環 (在 Swoole/Workerman 中放入 Timer/EventLoop)
// 這裡簡單演示：運行 10 秒
$start = time();
while (time() - $start < 10) {
    $client->loop();
    usleep(10000); // 10ms 降低 CPU
}

$client->close();

生產級增強清單

替代方案：gRPC (強烈推薦新項目)

手寫 Socket 協議屬於「造輪子」。新項目請直接使用：

• gRPC + Protobuf：IDL 定義接口，自動生成 PHP/Java 代碼，支持流式、攔截器、負載均衡、TLS、健康檢查。
• PHP 端：composer require grpc/grpc + pecl install grpc (或純 PHP spiral/roadrunner-grpc)。
• Java 端：Spring Boot + grpc-spring-boot-starter。

總結

• Socket 通信核心：**協議設計 (TLV/Protobuf) + 幂等業務邏輯 + 可靠傳輸層 (重連/心跳/超時)**。
• PHP 客戶端**非阻塞 IO + 事件循環**是高併發前提，Swoole/Workerman 原生支持，原生 PHP 需 stream_select 輪詢。
• Java 服務端**Netty** 是事實標準，無需手寫 NIO。
• **新項目首選 gRPC**，避開協議設計、序列化、治理等坑，專注業務。
• **舊系統集成**才考慮自定義 Socket，務必做好監控、告警、演練。