网络编程之手把手教你写基于TCP的Socket长连接

x33g5p2x  于2022-01-13 转载在 其他  
字(15.5k)|赞(0)|评价(0)|浏览(480)

TCP/IP 协议简介

IP协议

首先我们看 IP(Internet Protocol)协议。IP 协议提供了主机和主机间的通信。

为了完成不同主机的通信,我们需要某种方式来唯一标识一台主机,这个标识,就是著名的IP地址。通过IP地址,IP 协议就能够帮我们把一个数据包发送给对方

TCP协议

前面我们说过,IP 协议提供了主机和主机间的通信。TCP 协议在 IP 协议提供的主机间通信功能的基础上,完成这两个主机上进程对进程的通信。

有了 IP,不同主机就能够交换数据。但是,计算机收到数据后,并不知道这个数据属于哪个进程(简单讲,进程就是一个正在运行的应用程序)。TCP 的作用就在于,让我们能够知道这个数据属于哪个进程,从而完成进程间的通信。

为了标识数据属于哪个进程,我们给需要进行 TCP 通信的进程分配一个唯一的数字来标识它。这个数字,就是我们常说的端口号。

TCP 的全称是 Transmission Control Protocol,大家对它说得最多的,大概就是面向连接的特性了。之所以说它是有连接的,是说在进行通信前,通信双方需要先经过一个三次握手的过程。三次握手完成后,连接便建立了。这时候我们才可以开始发送/接收数据。(与之相对的是 UDP,不需要经过握手,就可以直接发送数据)。

下面我们简单了解一下三次握手的过程:

  • 首先,客户向服务端发送一个 SYN,假设此时 sequence number 为 x。这个 x是由操作系统根据一定的规则生成的,不妨认为它是一个随机数;
  • 服务端收到 SYN 后,会向客户端再发送一个 SYN,此时服务器的 seq number = y。与此同时,会 ACK x+1,告诉客户端“已经收到了 SYN,可以发送数据了”;
  • 客户端收到服务器的 SYN 后,回复一个 ACK y+1,这个 ACK 则是告诉服务器,SYN 已经收到,服务器可以发送数据了。

经过这 3 步,TCP 连接就建立了,这里需要注意的有三点:

  • 连接是由客户端主动发起的;
  • 在第 3 步客户端向服务器回复 ACK 的时候,TCP 协议是允许我们携带数据的。之所以做不到,是 API 的限制导致的;
  • TCP 协议还允许 “四次握手” 的发生,同样的,由于 API 的限制,这个极端的情况并不会发生。

Socket 基本用法

Socket 是 TCP 层的封装,通过 socket,我们就能进行 TCP 通信。

在 Java 的 SDK 中,socket 的共有两个接口:用于监听客户连接的 ServerSocket 和用于通信的 Socket

使用 socket 的步骤如下:

1)创建 ServerSocket 并监听客户连接;
2)使用 Socket 连接服务端;
3)通过 Socket.getInputStream()/getOutputStream() 获取输入输出流进行通信。

下面,我们通过实现一个简单的 echo 服务来学习 socket 的使用。所谓的 echo 服务,就是客户端向服务端写入任意数据,服务器都将数据原封不动地写回给客户端。

第一步:创建 ServerSocket 并监听客户连接

  1. public class EchoServer {
  2. private final ServerSocket mServerSocket;
  3. public EchoServer(int port) throws IOException {
  4. // 1. 创建一个 ServerSocket 并监听端口 port
  5. mServerSocket = new ServerSocket(port);
  6. }
  7. public void run() throws IOException {
  8. // 2. 开始接受客户连接
  9. Socket client = mServerSocket.accept();
  10. handleClient(client);
  11. }
  12. private void handleClient(Socket socket) {
  13. // 3. 使用 socket 进行通信 ...
  14. }
  15. public static void main(String[] argv) {
  16. try {
  17. EchoServer server = new EchoServer(9877);
  18. server.run();
  19. } catch (IOException e) {
  20. e.printStackTrace();
  21. }
  22. }
  23. }

第二步:使用 Socket 连接服务端

  1. public class EchoClient {
  2. private final Socket mSocket;
  3. public EchoClient(String host, int port) throws IOException {
  4. // 创建 socket 并连接服务器
  5. mSocket = new Socket(host, port);
  6. }
  7. public void run() {
  8. // 和服务端进行通信
  9. }
  10. public static void main(String[] argv) {
  11. try {
  12. // 由于服务端运行在同一主机,这里我们使用 localhost
  13. EchoClient client = new EchoClient("localhost", 9877);
  14. client.run();
  15. } catch (IOException e) {
  16. e.printStackTrace();
  17. }
  18. }
  19. }

第三步:通过 socket.getInputStream()/getOutputStream() 获取输入/输出流进行通信

首先,我们来实现服务端:

  1. public class EchoServer {
  2. // ...
  3. private void handleClient(Socket socket) throws IOException {
  4. InputStream in = socket.getInputStream();
  5. OutputStream out = socket.getOutputStream();
  6. byte[] buffer = new byte[1024];
  7. int n;
  8. while ((n = in.read(buffer)) > 0) {
  9. out.write(buffer, 0, n);
  10. }
  11. }
  12. }

可以看到,服务端的实现其实很简单,我们不停地读取输入数据,然后写回给客户端。

下面我们看看客户端:

  1. public class EchoClient {
  2. ...
  3. public void run() {
  4. //单独开辟一个线程进行读取操作
  5. Thread readThread = new Thread(this::readResponse);
  6. readThread.start();
  7. //用户在写入数据,发送给服务器
  8. try(OutputStream out = mSocket.getOutputStream())
  9. {
  10. byte[] buffer = new byte[1024];
  11. int n;
  12. while ((n = System.in.read(buffer)) > 0) {
  13. out.write(buffer, 0, n);
  14. }
  15. } catch (IOException e) {
  16. e.printStackTrace();
  17. }
  18. }
  19. private void readResponse()
  20. {
  21. try(InputStream inputStream=mSocket.getInputStream();OutputStream outputStream = mSocket.getOutputStream();)
  22. {
  23. byte[] buffer=new byte[1024];
  24. int read=-1;
  25. while((read=inputStream.read(buffer))!=-1) {
  26. //打印输出在控制台上
  27. System.out.write(buffer, 0, read);
  28. }
  29. }
  30. catch (IOException e) {
  31. e.printStackTrace();
  32. }
  33. }
  34. ...
  35. }

客户端会稍微复杂一点点,在读取用户输入的同时,我们又想读取服务器的响应。所以,这里创建了一个线程来读服务器的响应。

不熟悉 lambda 的读者,可以把Thread readerThread = new Thread(this::readResponse) 换成下面这个代码:

  1. Thread readerThread = new Thread(new Runnable() {
  2. @Override
  3. public void run() {
  4. readResponse();
  5. }
  6. });

打开两个 terminal 分别执行如下命令:

在客户端,我们会看到,输入的所有字符都打印了出来。

Socket、ServerSocket 傻傻分不清楚

在进入这一节的主题前,读者不妨先考虑一个问题:在上一节的实例中,我们运行 echo 服务后,在客户端连接成功时,一共有多少个 socket 存在?

答案是 3 个 socket:客户端一个,服务端有两个。跟这个问题的答案直接关联的是本节的主题——Socket 和 ServerSocket 的区别是什么。

眼尖的读者,可能会注意到在上一节我是这样描述他们的:

  1. Java SDK 中,socket 的共有两个接口:用于监听客户连接的 ServerSocket 和用于通信的 Socket

注意:我只说 ServerSocket 是用于监听客户连接,而没有说它也可以用来通信。下面我们来详细了解一下他们的区别。

注:以下描述使用的是 UNIX/Linux 系统的 API。

首先,我们创建 ServerSocket 后,内核会创建一个 socket。这个 socket 既可以拿来监听客户连接,也可以连接远端的服务。由于 ServerSocket 是用来监听客户连接的,紧接着它就会对内核创建的这个 socket 调用 listen 函数。这样一来,这个 socket 就成了所谓的 listening socket,它开始监听客户的连接。

接下来,我们的客户端创建一个 Socket,同样的,内核也创建一个 socket 实例。内核创建的这个 socket 跟 ServerSocket 一开始创建的那个没有什么区别。不同的是,接下来 Socket 会对它执行 connect,发起对服务端的连接。前面我们说过,socket API 其实是 TCP 层的封装,所以 connect 后,内核会发送一个 SYN 给服务端。

现在,我们切换角色到服务端。服务端的主机在收到这个 SYN 后,会创建一个新的 socket,这个新创建的 socket 跟客户端继续执行三次握手过程。

三次握手完成后,我们执行的 serverSocket.accept() 会返回一个 Socket 实例,这个 socket 就是上一步内核自动帮我们创建的。

所以说:在一个客户端连接的情况下,其实有 3 个 socket。

关于内核自动创建的这个 socket,还有一个很有意思的地方。它的端口号跟 ServerSocket 是一毛一样的。咦!!不是说,一个端口只能绑定一个 socket 吗?其实这个说法并不够准确。

前面我说的TCP 通过端口号来区分数据属于哪个进程的说法,在 socket 的实现里需要改一改。Socket 并不仅仅使用端口号来区别不同的 socket 实例,而是使用 <peer addr:peer port, local addr:local port> 这个四元组。

在上面的例子中,我们的 ServerSocket 长这样:<:, *:9877>。意思是,可以接受任何的客户端,和本地任何 IP。

accept 返回的 Socket 则是这样:<127.0.0.1:xxxx, 127.0.0.1:9877>。其中,xxxx 是客户端的端口号。

如果数据是发送给一个已连接的 socket,内核会找到一个完全匹配的实例,所以数据准确发送给了对端。

如果是客户端要发起连接,这时候只有 <:, *:9877> 会匹配成功,所以 SYN 也准确发送给了监听套接字。

Socket “长”连接的实现

背景知识

  1. Socket 长连接,指的是在客户和服务端之间保持一个 socket 连接长时间不断开。

比较熟悉 Socket 的读者,可能知道有这样一个 API:

  1. socket.setKeepAlive(true);

嗯……keep alive,“保持活着”,这个应该就是让 TCP 不断开的意思。那么,我们要实现一个 socket 的长连接,只需要这一个调用即可。

遗憾的是,生活并不总是那么美好。对于 4.4BSD 的实现来说,Socket 的这个 keep alive 选项如果打开并且两个小时内没有通信,那么底层会发一个心跳,看看对方是不是还活着。

注意:两个小时才会发一次。也就是说,在没有实际数据通信的时候,我把网线拔了,你的应用程序要经过两个小时才会知道。

在说明如何实现长连接前,我们先来理一理我们面临的问题

假定现在有一对已经连接的 socket,在以下情况发生时候,socket 将不再可用:

  1. 某一端关闭 socket:主动关闭的一方会发送 FIN,通知对方要关闭 TCP 连接。在这种情况下,另一端如果去读socket,将会读到 EoF(End of File)。于是我们知道对方关闭了 socket;
  2. 应用程序奔溃:此时 socket 会由内核关闭,结果跟情况1一样;
  3. 系统奔溃:这时候系统是来不及发送 FIN 的,因为它已经跪了。此时对方无法得知这一情况。对方在尝试读取数据时,最后会返回 read time out。如果写数据,则是 host unreachable 之类的错误。
  4. 电缆被挖断、网线被拔:跟情况3差不多,如果没有对 socket 进行读写,两边都不知道发生了事故。跟情况3不同的是,如果我们把网线接回去,socket 依旧可以正常使用。

在上面的几种情形中,有一个共同点就是,只要去读、写 socket,只要 socket 连接不正常,我们就能够知道。基于这一点,要实现一个 socket 长连接,我们需要做的就是不断地给对方写数据,然后读取对方的数据,也就是所谓的心跳。只要心还在跳,socket 就是活的。写数据的间隔,需要根据实际的应用需求来决定。

心跳包不是实际的业务数据,根据通信协议的不同,需要做不同的处理。

比方说,我们使用 JSON 进行通信,那么,可以为协议包加一个 type 字段,表面这个 JSON 是心跳还是业务数据:

  1. {
  2. "type": 0, // 0 表示心跳
  3. // ...
  4. }

使用二进制协议的情况类似。要求就是,我们能够区别一个数据包是心跳还是真实数据。这样,我们便实现了一个 socket 长连接。

实现示例

这一小节我们一起来实现一个带长连接的 Android echo 客户端。

首先是接口部分:

  1. package dhy.com;
  2. public final class LongLiveSocket {
  3. /** * 错误回调 */
  4. public interface ErrorCallback {
  5. /** * 如果需要重连,返回 true */
  6. boolean onError();
  7. }
  8. /** * 读数据回调 */
  9. public interface DataCallback {
  10. void onData(byte[] data, int offset, int len);
  11. }
  12. /** * 写数据回调 */
  13. public interface WritingCallback {
  14. void onSuccess();
  15. void onFail(byte[] data, int offset, int len);
  16. }
  17. public LongLiveSocket(String host, int port,
  18. DataCallback dataCallback, ErrorCallback errorCallback) {
  19. }
  20. public void write(byte[] data, WritingCallback callback) {
  21. }
  22. public void write(byte[] data, int offset, int len, WritingCallback callback) {
  23. }
  24. public void close() {
  25. }
  26. }

我们这个支持长连接的类就叫 LongLiveSocket 好了。如果在 socket 断开后需要重连,只需要在对应的接口里面返回 true 即可(在真实场景里,我们还需要让客户设置重连的等待时间,还有读写、连接的 timeout等。为了简单,这里就直接不支持了。

另外需要注意的一点是,如果要做一个完整的库,需要同时提供阻塞式和回调式API。同样由于篇幅原因,这里直接省掉了。

下面我们直接看实现:

  1. public final class LongLiveSocket {
  2. private static final String TAG = "LongLiveSocket";
  3. private static final long RETRY_INTERVAL_MILLIS = 3 * 1000;
  4. private static final long HEART_BEAT_INTERVAL_MILLIS = 5 * 1000;
  5. private static final long HEART_BEAT_TIMEOUT_MILLIS = 2 * 1000;
  6. /** * 错误回调 */
  7. public interface ErrorCallback {
  8. /** * 如果需要重连,返回 true */
  9. boolean onError();
  10. }
  11. /** * 读数据回调 */
  12. public interface DataCallback {
  13. void onData(byte[] data, int offset, int len);
  14. }
  15. /** * 写数据回调 */
  16. public interface WritingCallback {
  17. void onSuccess();
  18. void onFail(byte[] data, int offset, int len);
  19. }
  20. private final String mHost;
  21. private final int mPort;
  22. private final DataCallback mDataCallback;
  23. private final ErrorCallback mErrorCallback;
  24. private final HandlerThread mWriterThread;
  25. private final Handler mWriterHandler;
  26. private final Handler mUIHandler = new Handler(Looper.getMainLooper());
  27. private final Object mLock = new Object();
  28. private Socket mSocket; // guarded by mLock
  29. private boolean mClosed; // guarded by mLock
  30. private final Runnable mHeartBeatTask = new Runnable() {
  31. private byte[] mHeartBeat = new byte[0];
  32. @Override
  33. public void run() {
  34. // 我们使用长度为 0 的数据作为 heart beat
  35. write(mHeartBeat, new WritingCallback() {
  36. @Override
  37. public void onSuccess() {
  38. // 每隔 HEART_BEAT_INTERVAL_MILLIS 发送一次
  39. mWriterHandler.postDelayed(mHeartBeatTask, HEART_BEAT_INTERVAL_MILLIS);
  40. mUIHandler.postDelayed(mHeartBeatTimeoutTask, HEART_BEAT_TIMEOUT_MILLIS);
  41. }
  42. @Override
  43. public void onFail(byte[] data, int offset, int len) {
  44. // nop
  45. // write() 方法会处理失败
  46. }
  47. });
  48. }
  49. };
  50. private final Runnable mHeartBeatTimeoutTask = () -> {
  51. Log.e(TAG, "mHeartBeatTimeoutTask#run: heart beat timeout");
  52. closeSocket();
  53. };
  54. public LongLiveSocket(String host, int port,
  55. DataCallback dataCallback, ErrorCallback errorCallback) {
  56. mHost = host;
  57. mPort = port;
  58. mDataCallback = dataCallback;
  59. mErrorCallback = errorCallback;
  60. mWriterThread = new HandlerThread("socket-writer");
  61. mWriterThread.start();
  62. mWriterHandler = new Handler(mWriterThread.getLooper());
  63. mWriterHandler.post(this::initSocket);
  64. }
  65. private void initSocket() {
  66. while (true) {
  67. if (closed()) return;
  68. try {
  69. Socket socket = new Socket(mHost, mPort);
  70. synchronized (mLock) {
  71. // 在我们创建 socket 的时候,客户可能就调用了 close()
  72. if (mClosed) {
  73. silentlyClose(socket);
  74. return;
  75. }
  76. mSocket = socket;
  77. // 每次创建新的 socket,会开一个线程来读数据
  78. Thread reader = new Thread(new ReaderTask(socket), "socket-reader");
  79. reader.start();
  80. mWriterHandler.post(mHeartBeatTask);
  81. }
  82. break;
  83. } catch (IOException e) {
  84. Log.e(TAG, "initSocket: ", e);
  85. if (closed() || !mErrorCallback.onError()) {
  86. break;
  87. }
  88. try {
  89. TimeUnit.MILLISECONDS.sleep(RETRY_INTERVAL_MILLIS);
  90. } catch (InterruptedException e1) {
  91. // interrupt writer-thread to quit
  92. break;
  93. }
  94. }
  95. }
  96. }
  97. public void write(byte[] data, WritingCallback callback) {
  98. write(data, 0, data.length, callback);
  99. }
  100. public void write(byte[] data, int offset, int len, WritingCallback callback) {
  101. mWriterHandler.post(() -> {
  102. Socket socket = getSocket();
  103. if (socket == null) {
  104. // initSocket 失败而客户说不需要重连,但客户又叫我们给他发送数据
  105. throw new IllegalStateException("Socket not initialized");
  106. }
  107. try {
  108. OutputStream outputStream = socket.getOutputStream();
  109. DataOutputStream out = new DataOutputStream(outputStream);
  110. out.writeInt(len);
  111. out.write(data, offset, len);
  112. callback.onSuccess();
  113. } catch (IOException e) {
  114. Log.e(TAG, "write: ", e);
  115. closeSocket();
  116. callback.onFail(data, offset, len);
  117. if (!closed() && mErrorCallback.onError()) {
  118. initSocket();
  119. }
  120. }
  121. });
  122. }
  123. private boolean closed() {
  124. synchronized (mLock) {
  125. return mClosed;
  126. }
  127. }
  128. private Socket getSocket() {
  129. synchronized (mLock) {
  130. return mSocket;
  131. }
  132. }
  133. private void closeSocket() {
  134. synchronized (mLock) {
  135. closeSocketLocked();
  136. }
  137. }
  138. private void closeSocketLocked() {
  139. if (mSocket == null) return;
  140. silentlyClose(mSocket);
  141. mSocket = null;
  142. mWriterHandler.removeCallbacks(mHeartBeatTask);
  143. }
  144. public void close() {
  145. if (Looper.getMainLooper() == Looper.myLooper()) {
  146. new Thread() {
  147. @Override
  148. public void run() {
  149. doClose();
  150. }
  151. }.start();
  152. } else {
  153. doClose();
  154. }
  155. }
  156. private void doClose() {
  157. synchronized (mLock) {
  158. mClosed = true;
  159. // 关闭 socket,从而使得阻塞在 socket 上的线程返回
  160. closeSocketLocked();
  161. }
  162. mWriterThread.quit();
  163. // 在重连的时候,有个 sleep
  164. mWriterThread.interrupt();
  165. }
  166. private static void silentlyClose(Closeable closeable) {
  167. if (closeable != null) {
  168. try {
  169. closeable.close();
  170. } catch (IOException e) {
  171. Log.e(TAG, "silentlyClose: ", e);
  172. // error ignored
  173. }
  174. }
  175. }
  176. private class ReaderTask implements Runnable {
  177. private final Socket mSocket;
  178. public ReaderTask(Socket socket) {
  179. mSocket = socket;
  180. }
  181. @Override
  182. public void run() {
  183. try {
  184. readResponse();
  185. } catch (IOException e) {
  186. Log.e(TAG, "ReaderTask#run: ", e);
  187. }
  188. }
  189. private void readResponse() throws IOException {
  190. // For simplicity, assume that a msg will not exceed 1024-byte
  191. byte[] buffer = new byte[1024];
  192. InputStream inputStream = mSocket.getInputStream();
  193. DataInputStream in = new DataInputStream(inputStream);
  194. while (true) {
  195. int nbyte = in.readInt();
  196. if (nbyte == 0) {
  197. Log.i(TAG, "readResponse: heart beat received");
  198. mUIHandler.removeCallbacks(mHeartBeatTimeoutTask);
  199. continue;
  200. }
  201. if (nbyte > buffer.length) {
  202. throw new IllegalStateException("Receive message with len " + nbyte +
  203. " which exceeds limit " + buffer.length);
  204. }
  205. if (readn(in, buffer, nbyte) != 0) {
  206. // Socket might be closed twice but it does no harm
  207. silentlyClose(mSocket);
  208. // Socket will be re-connected by writer-thread if you want
  209. break;
  210. }
  211. mDataCallback.onData(buffer, 0, nbyte);
  212. }
  213. }
  214. private int readn(InputStream in, byte[] buffer, int n) throws IOException {
  215. int offset = 0;
  216. while (n > 0) {
  217. int readBytes = in.read(buffer, offset, n);
  218. if (readBytes < 0) {
  219. // EoF
  220. break;
  221. }
  222. n -= readBytes;
  223. offset += readBytes;
  224. }
  225. return n;
  226. }
  227. }
  228. }

下面是我们新实现的 EchoClient:

  1. public class EchoClient {
  2. private static final String TAG = "EchoClient";
  3. private final LongLiveSocket mLongLiveSocket;
  4. public EchoClient(String host, int port) {
  5. mLongLiveSocket = new LongLiveSocket(
  6. host, port,
  7. (data, offset, len) -> Log.i(TAG, "EchoClient: received: " + new String(data, offset, len)),
  8. // 返回 true,所以只要出错,就会一直重连
  9. () -> true);
  10. }
  11. public void send(String msg) {
  12. mLongLiveSocket.write(msg.getBytes(), new LongLiveSocket.WritingCallback() {
  13. @Override
  14. public void onSuccess() {
  15. Log.d(TAG, "onSuccess: ");
  16. }
  17. @Override
  18. public void onFail(byte[] data, int offset, int len) {
  19. Log.w(TAG, "onFail: fail to write: " + new String(data, offset, len));
  20. // 连接成功后,还会发送这个消息
  21. mLongLiveSocket.write(data, offset, len, this);
  22. }
  23. });
  24. }
  25. }

下面是一些输出示例:

  1. 03:54:55.583 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  2. 03:55:00.588 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  3. 03:55:05.594 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  4. 03:55:09.638 12691-12710/com.example.echo D/EchoClient: onSuccess:
  5. 03:55:09.639 12691-12713/com.example.echo I/EchoClient: EchoClient: received: hello
  6. 03:55:10.595 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  7. 03:55:14.652 12691-12710/com.example.echo D/EchoClient: onSuccess:
  8. 03:55:14.654 12691-12713/com.example.echo I/EchoClient: EchoClient: received: echo
  9. 03:55:15.596 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  10. 03:55:20.597 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received
  11. 03:55:25.602 12691-12713/com.example.echo I/LongLiveSocket: readResponse: heart beat received

最后需要说明的是,如果想节省资源,在有客户发送数据的时候可以省略 heart beat。

我们对读出错时候的处理,可能也存在一些争议。读出错后,我们只是关闭了 socket。socket 需要等到下一次写动作发生时,才会重新连接。实际应用中,如果这是一个问题,在读出错后可以直接开始重连。这种情况下,还需要一些额外的同步,避免重复创建 socket。heart beat timeout 的情况类似。

跟 TCP/IP 学协议设计

如果仅仅是为了使用是 socket,我们大可以不去理会协议的细节。之所以推荐大家去看一看《TCP/IP 详解》,是因为它们有太多值得学习的地方。很多我们工作中遇到的问题,都可以在这里找到答案。

以下每一个小节的标题都是一个小问题,建议读者独立思考一下,再继续往下看。

协议版本如何升级?

有这么一句流行的话:这个世界唯一不变的,就是变化。当我们对协议版本进行升级的时候,正确识别不同版本的协议对软件的兼容非常重要。那么,我们如何设计协议,才能够为将来的版本升级做准备呢?

答案可以在 IP 协议找到。

IP 协议的第一个字段叫 version,目前使用的是 4 或 6,分别表示 IPv4 和 IPv6。由于这个字段在协议的开头,接收端收到数据后,只要根据第一个字段的值就能够判断这个数据包是 IPv4 还是 IPv6。

再强调一下,这个字段在两个版本的IP协议都位于第一个字段,为了做兼容处理,对应的这个字段必须位于同一位置。文本协议(如,JSON、HTML)的情况类似。

如何发送不定长数据的数据包?

举个例子,我们用微信发送一条消息。这条消息的长度是不确定的,并且每条消息都有它的边界。我们如何来处理这个边界呢?

还是一样,看看 IP。IP 的头部有个 header length 和 data length 两个字段。通过添加一个 len 域,我们就能够把数据根据应用逻辑分开。

跟这个相对的,还有另一个方案,那就是在数据的末尾放置终止符。比方说,想 C 语言的字符串那样,我们在每个数据的末尾放一个 \0 作为终止符,用以标识一条消息的尾部。这个方法带来的问题是,用户的数据也可能存在 \0。此时,我们就需要对用户的数据进行转义。比方说,把用户数据的所有 \0 都变成 \0\0。读消息的过程总,如果遇到 \0\0,那它就代表 \0,如果只有一个 \0,那就是消息尾部。

使用 len 字段的好处是,我们不需要对数据进行转义。读取数据的时候,只要根据 len 字段,一次性把数据都读进来就好,效率会更高一些。

终止符的方案虽然要求我们对数据进行扫描,但是如果我们可能从任意地方开始读取数据,就需要这个终止符来确定哪里才是消息的开头了。

当然,这两个方法不是互斥的,可以一起使用。

上传多个文件,只有所有文件都上传成功时才算成功

现在我们有一个需求,需要一次上传多个文件到服务器,只有在所有文件都上传成功的情况下,才算成功。我们该如何来实现呢?

IP 在数据报过大的时候,会把一个数据报拆分成多个,并设置一个 MF (more fragments)位,表示这个包只是被拆分后的数据的一部分。

好,我们也学一学 IP。这里,我们可以给每个文件从 0 开始编号。上传文件的同时,也携带这个编号,并额外附带一个 MF 标志。除了编号最大的文件,所有文件的 MF 标志都置位。因为 MF 没有置位的是最后一个文件,服务器就可以根据这个得出总共有多少个文件。

另一种不使用 MF 标志的方法是,我们在上传文件前,就告诉服务器总共有多少个文件。

如果读者对数据库比较熟悉,学数据库用事务来处理,也是可以的。这里就不展开讨论了。

如何保证数据的有序性?

这里讲一个我曾经遇到过的面试题。现在有一个任务队列,多个工作线程从中取出任务并执行,执行结果放到一个结果队列中。先要求,放入结果队列的时候,顺序顺序需要跟从工作队列取出时的一样(也就是说,先取出的任务,执行结果需要先放入结果队列)。

我们看看 TCP/IP 是怎么处理的。IP 在发送数据的时候,不同数据报到达对端的时间是不确定的,后面发送的数据有可能较先到达。TCP 为了解决这个问题,给所发送数据的每个字节都赋了一个序列号,通过这个序列号,TCP 就能够把数据按原顺序重新组装。

一样,我们也给每个任务赋一个值,根据进入工作队列的顺序依次递增。工作线程完成任务后,在将结果放入结果队列前,先检查要放入对象的写一个序列号是不是跟自己的任务相同,如果不同,这个结果就不能放进去。此时,最简单的做法是等待,知道下一个可以放入队列的结果是自己所执行的那一个。但是,这个线程就没办法继续处理任务了。

更好的方法是,我们维护多一个结果队列的缓冲,这个缓冲里面的数据按序列号从小到大排序。

工作线程要将结果放入,有两种可能:

1)刚刚完成的任务刚好是下一个,将这个结果放入队列。然后从缓冲的头部开始,将所有可以放入结果队列的数据都放进去;

2)所完成的任务不能放入结果队列,这个时候就插入结果队列。然后,跟上一种情况一样,需要检查缓冲。

如果测试表明,这个结果缓冲的数据不多,那么使用普通的链表就可以。如果数据比较多,可以使用一个最小堆。

如何保证对方收到了消息?

我们说,TCP 提供了可靠的传输。这样不就能够保证对方收到消息了吗?

很遗憾,其实不能。在我们往 socket 写入的数据,只要对端的内核收到后,就会返回 ACK,此时,socket 就认为数据已经写入成功。然而要注意的是,这里只是对方所运行的系统的内核成功收到了数据,并不表示应用程序已经成功处理了数据。

解决办法还是一样,我们学 TCP,添加一个应用层的 APP ACK。应用接收到消息并处理成功后,发送一个 APP ACK 给对方。

有了 APP ACK,我们需要处理的另一个问题是,如果对方真的没有收到,需要怎么做?

TCP 发送数据的时候,消息一样可能丢失。TCP 发送数据后,如果长时间没有收到对方的 ACK,就假设数据已经丢失,并重新发送。

我们也一样,如果长时间没有收到 APP ACK,就假设数据丢失,重新发送一个。

相关文章