PHP之 Socket实践

一 Socket简介

1.1 Socket(套接宇)，用来描述IP地址和端口,是通信链的句柄，应用程序可以通过Socket向网络发送请求或者应答网络请求。

1.2 Socket是支持TCP/IP协议的网络通信的基本操作单元，是对网络通信过程中端点的抽象表示，包含了进行网络通信所必须的五种信息

连接所使用的协议
本地主机的IP地址
本地远程的协议端口
远地主机的IP地址以
远地进程的协议端口

二 PHPSocket服务器端示例

2.1 首先要在PHP安装目录，开启php.ini配置文件里面的socket扩展，extension=sockets；

2.2 本地调试，ip为127.0.0.1或者本机P地址，端口找个未被使用的就可以比如1000

2.3 服务器端源码：server.php

<?php
$host = '127.0.0.1';
$port = '1000';
$null = NULL;

//创建tcp socket
$socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1);
socket_bind($socket, 0, $port);

//监听端口
socket_listen($socket);

//连接的client socket 列表
$clients = array($socket);

//设置一个死循环,用来监听连接 ,状态
while (true) {

    $changed = $clients;
    socket_select($changed, $null, $null, 0, 10);

    //如果有新的连接
    if (in_array($socket, $changed)) {
        //接受并加入新的socket连接
        $socket_new = socket_accept($socket);
        $clients[] = $socket_new;

        //通过socket获取数据执行handshake
        $header = socket_read($socket_new, 1024);
        perform_handshaking($header, $socket_new, $host, $port);

        //获取client ip 编码json数据,并发送通知
        socket_getpeername($socket_new, $ip);
        $response = mask(json_encode(array('type' => 'system', 'message' => $ip . ' connected')));
        send_message($response);
        $found_socket = array_search($socket, $changed);
        unset($changed[$found_socket]);
    }

    //轮询 每个client socket 连接
    foreach ($changed as $changed_socket) {

        //如果有client数据发送过来
        while (@socket_recv($changed_socket, $buf, 1024, 0) >= 1) {
            //解码发送过来的数据
            $received_text = unmask($buf);
            $tst_msg = json_decode($received_text);
			if($tst_msg!=null){
				$user_name = $tst_msg->id;
				$user_message = $tst_msg->title;
				//把消息发送回所有连接的 client 上去
				$response_text = mask(json_encode(array('type' => 'usermsg', 'name' => $user_name, 'message' => $user_message)));
				send_message($response_text);
				echo $response_text.PHP_EOL;
			}
        
            break 2;
        }

        //检查offline的client
        $buf = @socket_read($changed_socket, 1024, PHP_NORMAL_READ);
        if ($buf === false) {
            $found_socket = array_search($changed_socket, $clients);
            socket_getpeername($changed_socket, $ip);
            unset($clients[$found_socket]);
            $response = mask(json_encode(array('type' => 'system', 'message' => $ip . ' disconnected')));
            send_message($response);
        }
    }
}
// 关闭监听的socket
socket_close($sock);

//发送消息的方法
function send_message($msg)
{
    global $clients;
    foreach ($clients as $changed_socket) {
        @socket_write($changed_socket, $msg, strlen($msg));
    }
    return true;
}


//解码数据
function unmask($text)
{
    $length = ord($text[1]) & 127;
    if ($length == 126) {
        $masks = substr($text, 4, 4);
        $data = substr($text, 8);
    } elseif ($length == 127) {
        $masks = substr($text, 10, 4);
        $data = substr($text, 14);
    } else {
        $masks = substr($text, 2, 4);
        $data = substr($text, 6);
    }
    $text = "";
    for ($i = 0; $i < strlen($data); ++$i) {
        $text .= $data[$i] ^ $masks[$i % 4];
    }
    return $text;
}

//编码数据
function mask($text)
{
    $b1 = 0x80 | (0x1 & 0x0f);
    $length = strlen($text);

    if ($length <= 125)
        $header = pack('CC', $b1, $length);
    elseif ($length > 125 && $length < 65536)
        $header = pack('CCn', $b1, 126, $length);
    elseif ($length >= 65536)
        $header = pack('CCNN', $b1, 127, $length);
    return $header . $text;
}

//握手的逻辑
function perform_handshaking($receved_header, $client_conn, $host, $port)
{
    $headers = array();
    $lines = preg_split("/\r\n/", $receved_header);
    foreach ($lines as $line) {
        $line = chop($line);
        if (preg_match('/\A(\S+): (.*)\z/', $line, $matches)) {
            $headers[$matches[1]] = $matches[2];
        }
    }

    $secKey = $headers['Sec-WebSocket-Key'];
    $secAccept = base64_encode(pack('H*', sha1($secKey . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));
    $upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" .
        "Upgrade: websocket\r\n" .
        "Connection: Upgrade\r\n" .
        "WebSocket-Origin: $host\r\n" .
        "WebSocket-Location: ws://$host:$port/demo/shout.php\r\n" .
        "Sec-WebSocket-Accept:$secAccept\r\n\r\n";
    socket_write($client_conn, $upgrade, strlen($upgrade));
}

2.4 运行php指令开启该socket服务，光标闪烁说明服务开启成功

三 HTML+JS客户端示例

3.1 新建html脚本文件client.html，填写服务器的ip和端口，连接服务器socket

<!DOCTYPE html>
<html>
<head>
<title>Test - PHP Push WebSocket</title>
<meta charset="utf-8" />

<script type="text/javascript">
var ws, url = 'ws://127.0.0.1:1000';

window.onbeforeunload = function() {
	ws.send('quit');
};
window.onload = function() {
	try {
		ws = new WebSocket(url);
		write('Connecting... (readyState '+ws.readyState+')');
		ws.onopen = function(msg) {
			write('Connection successfully opened (readyState ' + this.readyState+')');
		
		   var message = {
	           id: 1,
	           title: 'hahahahahahah'
           }
           ws.send(JSON.stringify(message)); 
		
		};
		ws.onmessage = function(msg) {
			write('Server says: '+msg.data);
		};
		ws.onclose = function(msg) {
			if(this.readyState == 2)
				write('Closing... The connection is going throught the closing handshake (readyState '+this.readyState+')');
			else if(this.readyState == 3)
				write('Connection closed... The connection has been closed or could not be opened (readyState '+this.readyState+')');
			else
				write('Connection closed... (unhandled readyState '+this.readyState+')');
		};
		ws.onerror = function(event) {
			terminal.innerHTML = '<li style="color: red;">'+event.data+'</li>'+terminal.innerHTML;
		};
	}
	catch(exception) {
		write(exception);
	}
};

function write(text) {
	var date = new Date();
	var dateText = '['+date.getFullYear()+'-'+(date.getMonth()+1 > 9 ? date.getMonth()+1 : '0'+date.getMonth()+1)+'-'+(date.getDate() > 9 ? date.getDate() : '0'+date.getDate())+' '+(date.getHours() > 9 ? date.getHours() : '0'+date.getHours())+':'+(date.getMinutes() > 9 ? date.getMinutes() : '0'+date.getMinutes())+':'+(date.getSeconds() > 9 ? date.getSeconds() : '0'+date.getSeconds())+']';
	var terminal = document.getElementById('terminal');
	terminal.innerHTML = '<li>'+dateText+' '+text+'</li>'+terminal.innerHTML;
}
</script>

</head>
<body>
	<a href="client.html" target="_blank">Add another client</a>
	<ul id="terminal"></ul>
</body>
</html>

3.2 如下浏览器地址栏输入 http://localhost/client.html，新建客户端，连接成功会收到服务器发送过来的消息

3.3 同时有新客户端连接，服务器端也会收到客户端的发来的消息

四 Android端Socket客户端示例

4.1 直接用OkHttp里面的Websocket来连接socket，同时添加心跳机制

public class MainActivity extends AppCompatActivity {
    private WebSocket mWebSocket;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

    }



    public void openSocket(View view) {
        initSocket();
    }


    private void initSocket() {
        String sw_url = "ws://192.168.81.1:1000";
        OkHttpClient client = new OkHttpClient.Builder().readTimeout(5000, TimeUnit.MILLISECONDS).build();
        Request request = new Request.Builder().url(sw_url).build();
        client.newWebSocket(request, new WebSocketListener() {
            @Override
            public void onOpen(WebSocket webSocket, Response response) {//开启长连接成功的回调
                super.onOpen(webSocket, response);
                mWebSocket = webSocket;
            }

            @Override
            public void onMessage(WebSocket webSocket, String text) {//接收消息的回调
                super.onMessage(webSocket, text);
                //收到服务器端传过来的消息text
                Log.e("TAG", "接收消息的回调--------------" + text);
                try {
                    //这个是解析你的回调数据
                    JSONObject jsonObject = new JSONObject(text);
                    //通过事件总线修改
                    String type = jsonObject.getString("type");
                    String message = jsonObject.getString("message");
                    Log.e("TAG", "message：" + message);
                } catch (JSONException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onMessage(WebSocket webSocket, ByteString bytes) {
                super.onMessage(webSocket, bytes);
                Log.e("TAG", "onMessage:" + bytes.toString());
            }

            @Override
            public void onClosing(WebSocket webSocket, int code, String reason) {
                super.onClosing(webSocket, code, reason);
                Log.e("TAG", "onClosing:" + reason);

            }

            @Override
            public void onClosed(WebSocket webSocket, int code, String reason) {
                super.onClosed(webSocket, code, reason);
                Log.e("TAG", "onClosed:" + reason);
            }

            @Override
            public void onFailure(WebSocket webSocket, Throwable t, @Nullable Response response) {//长连接连接失败的回调
                super.onFailure(webSocket, t, response);
                Log.e("TAG", "onFailure:" + response);

            }
        });
        client.dispatcher().executorService().shutdown();
        mHandler.postDelayed(heartBeatRunnable, HEART_BEAT_RATE);//开启心跳检测
    }



    class InitSocketThread extends Thread {
        @Override
        public void run() {
            super.run();
            try {
                initSocket();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 心跳检测时间
     */
    private static final long HEART_BEAT_RATE = 10 * 1000;//每隔10秒进行一次对长连接的心跳检测

    private long sendTime = 0L;
    // 发送心跳包
    private Handler mHandler = new Handler();
    private Runnable heartBeatRunnable = new Runnable() {
        @Override
        public void run() {
            if (System.currentTimeMillis() - sendTime >= HEART_BEAT_RATE) {
                if (mWebSocket != null) {
                    String msg="android-心跳";
                    try {
                        JSONObject jsonObject=new JSONObject();
                        jsonObject.put("id","111");
                        jsonObject.put("title","EEEEEEEEEEEEEEEE");
                        msg=jsonObject.toString();
                    }catch (Exception e){
                        e.printStackTrace();
                    }

                    boolean isSuccess = mWebSocket.send(msg);//发送一个消息给服务器，通过发送消息的成功失败来判断长连接的连接状态
                    if (!isSuccess) {//长连接已断开，
                        Log.e("TAG", "发送心跳包-------------长连接已断开");
                        mHandler.removeCallbacks(heartBeatRunnable);
                        mWebSocket.cancel();//取消掉以前的长连接
                        new InitSocketThread().start();//创建一个新的连接
                    } else {//长连接处于连接状态---
                        Log.e("TAG", "发送心跳包-------------长连接处于连接状态");
                    }
                }

                sendTime = System.currentTimeMillis();
            }
            mHandler.postDelayed(this, HEART_BEAT_RATE);//每隔一定的时间，对长连接进行一次心跳检测
        }
    };

    //关闭长连接
    @Override
    public void onDestroy() {
        super.onDestroy();
        if (mWebSocket != null) {
            mWebSocket.close(1000, null);
        }
    }

}

4.2 可以看到控制台连接成功会收到服务器端消息，同时心跳打印