Android-Socket

基本概念

是对 TCP/IP 协议族的一种封装，是应用层与TCP/IP协议族通信的中间软件抽象层。
从设计模式的角度看来，Socket其实就是一个门面模式，它把复杂的TCP/IP协议族隐藏在Socket接口后面，对用户来说，一组简单的接口就是全部，让Socket去组织数据，以符合指定的协议。
还可以认为是一种网络间不同计算机上的进程通信的一种方法，利用三元组（ip地址，协议，端口）就可以唯一标识网络中的进程，网络中的进程通信可以利用这个标志与其它进程进行交互。
起源于 Unix ，Unix/Linux 基本哲学之一就是“一切皆文件”，都可以用“打开(open) –> 读写(write/read) –> 关闭(close)”模式来进行操作。因此 Socket 也被处理为一种特殊的文件。

在Java的SDK中，Socket有两个接口：

• 用于监听客户连接的ServerSocket
• 用于通信的Socket

基本用法

使用Socket的步骤：

• 创建ServerSocket并监听客户连接
• 使用Socket连接服务器
• 通过Socket获取输入输出流进行通信

示例：

/**
 * 实现一个echo 服务，就是客户端向服务端写入任意数据，服务器都将数据原封不动地写回给客户端。
 * 1.创建ServerSocket并监听客户连接
 *
 * 直接右键运行此文件
 * Created by xwxwaa on 2019/5/30.
 */

public class EchoServer {

    private final ServerSocket serverSocket;

    public static void main(String[] argv){
        EchoServer echoServer = null;
        try {
            echoServer = new EchoServer(9876);
            echoServer.run();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public EchoServer(int port) throws IOException {
        // 1.创建一个ServerSocket，并监听端口 port
        // 内核会创建一个Socket，然后ServerSocket对这个Socket调用listen函数，开始监听客户的连接。
        // Socket 并不仅仅使用端口号来区别不同的 socket 实例，而是使用 <peer addr:peer port, local addr:local port> 这个四元组。
        // 关于端口号：ServerSocket 长这样：<*:*, *:9876>。意思是，可以接受任何的客户端，和本地任何 IP。
        serverSocket = new ServerSocket(port);
    }

    public void run() throws IOException {
        // 2.开始接受客户连接
        // 服务端的主机接收到SYN后，会创建一个新的Socket，这个新的Socket会跟客户端继续执行三次握手过程。
        // 三次握手完成后，执行的serverSocket.accept()会返回一个实例，这个Socket就是上一步内核创建的。
        // 关于端口号：accept 返回的 Socket 则是这样：<127.0.0.1:xxxx, 127.0.0.1:9877>，其中xxxx 是客户端的端口号。
        Socket socket = serverSocket.accept();
        handleChilent(socket);
    }

    private void handleChilent(Socket socket) throws IOException {
        // 3.使用Socket进行通信
        // 服务端，通过socket获取输入输出流进行通信
        InputStream inputStream = socket.getInputStream();
        OutputStream outputStream = socket.getOutputStream();
        byte[] buffer = new byte[1024];
        int n;
        // 不停的读取输入数据，然后写回给客户端。
        while (( n = inputStream.read(buffer))>0){
            outputStream.write(buffer,0,n);
        }
    }
}

/**
 * 实现一个echo 服务，就是客户端向服务端写入任意数据，服务器都将数据原封不动地写回给客户端。
 * 使用Socket连接服务器
 *
 * 直接右键运行此文件
 * Created by xwxwaa on 2019/5/30.
 */

public class EchoClient {

    private final Socket socket;

    public EchoClient(String host,int port) throws IOException {
        // 创建Socket并连接服务器
        // 内核会创建一个Socket，并且Socket会对它执行connect，发起对服务端的连接。
        // 因为Socket API其实是TCP层的封装，所以connect后，内核会发送一个SYN给服务端。
        socket= new Socket(host,port);
    }

    public void run() throws IOException {
        // 和服务端进行通信
        // 通过 socket 获取输入/输出流进行通信
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                // 在读取用户输入的同时，我们又想读取服务器的响应。
                // 所以，这里创建了一个线程来读服务器的响应。
                readResponse();
            }
        });
        thread.start();

        OutputStream outputStream = socket.getOutputStream();
        byte[] buffer = new byte[1024];
        int n ;
        // 从键盘读出一个字符，然后返回它的Unicode码
        // 用System.in.read()时，在键盘上按下的任何一个键都会被当做是输入值，包括Enter键也会被当做是一个值！
        while ((n = System.in.read(buffer))>0){
            outputStream.write(buffer,0,n);
        }
    }


    private void readResponse(){
        try {
            InputStream inputStream = socket.getInputStream();
            byte[] buffer = new byte[1024];
            int n ;
            while ((n=inputStream.read(buffer))>0){
                System.out.write(buffer,0,n);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] argv){
        try {
            // 由于服务端运行在同一主机，使用localhost
            EchoClient echoClient = new EchoClient("localhost",9876);
            echoClient.run();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

需要注意几点：

• 在上面的代码中，所有的异常都没有处理。实际应用中，在发生异常时，需要关闭 socket，并根据实际业务做一些错误处理工作
• 在客户端，没有停止 readThread。实际应用中，我们可以通过关闭 socket 来让线程从阻塞读中返回。推荐读者阅读《Java并发编程实战》
• 服务端只处理了一个客户连接。如果需要同时处理多个客户端，可以创建线程来处理请求。

Socket长连接

Socket长连接指的是客户和服务器之间保持一个socket连接长时间不断开。
对于 4.4BSD 的实现来说，Socket 的 keep alive 选项如果打开（socket.setKeepAlive(true);）并且“两个小时“内没有通信，那么底层会发一个心跳，看看对方是不是还活着。

实现长连接前，面临的问题。假定现在有一对已经连接的 socket，在以下情况发生时候，socket 将不再可用：

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

从上可以知道，在没有实际数据通信的时候，应用程序要经过两个小时才会知道对方的状态。而只要去读、写 socket，只要 socket 连接不正常，我们就能够知道。基于这一点，要实现一个 socket 长连接，我们需要做的就是不断地给对方写数据，然后读取对方的数据，也就是所谓的心跳。只要心还在跳，socket 就是活的。写数据的间隔，需要根据实际的应用需求来决定。

心跳包不是实际的业务数据，根据通信协议的不同，需要做不同的处理。需要能够区别一个数据包是心跳还是真实数据，比如JSON中加一个 type 字段。

示例：

/**
 * 长连接
 *
 * Created by xwxwaa on 2019/5/31.
 */
public final class LongLiveSocket {

    // 日志
    private static final String TAG = "LongLiveSocket";

    // 重试时间间隔
    private static final long RETRY_INTERVAL_MILLIS = 3 * 1000;
    // 心跳间隔
    private static final long HEART_BEAT_INTERVAL_MILLIS = 5 * 1000;
    // 心跳超时
    private static final long HEART_BEAT_TIMEOUT_MILLIS = 2 * 1000;

    /**
     * 错误回调
     */
    public interface ErrorCallback {
        /**
         * 如果需要重连，返回 true
         */
        boolean onError();
    }

    /**
     * 读数据回调
     */
    public interface DataCallback {
        void onData(byte[] data, int offset, int len);
    }

    /**
     * 写数据回调
     */
    public interface WritingCallback {
        void onSuccess();
        void onFail(byte[] data, int offset, int len);
    }

    // 主机地址
    private final String mHost;
    // 端口号
    private final int mPort;
    // 读数据回调
    private final DataCallback mDataCallback;
    // 错误回调
    private final ErrorCallback mErrorCallback;

    private final HandlerThread mWriterThread;
    private final Handler mWriterHandler;
    private final Handler mUIHandler = new Handler(Looper.getMainLooper());

    private final Object mLock = new Object();
    private Socket mSocket;  // guarded by mLock
    private boolean mClosed; // guarded by mLock

    private volatile int mSeqNumHeartBeatSent;
    private volatile int mSeqNumHeartBeatRecv;

    private final Runnable mHeartBeatTask = new Runnable() {
        private byte[] mHeartBeat = new byte[0];

        @Override
        public void run() {
            // no need to be atomic
            // noinspection NonAtomicOperationOnVolatileField
            ++mSeqNumHeartBeatSent;
            // 我们使用长度为 0 的数据作为 heart beat
            write(mHeartBeat, new WritingCallback() {
                @Override
                public void onSuccess() {
                    // 每隔 HEART_BEAT_INTERVAL_MILLIS 发送一次，心跳间隔
                    mWriterHandler.postDelayed(mHeartBeatTask, HEART_BEAT_INTERVAL_MILLIS);
                    // At this point, the heart-beat might be received and handled
                    if (mSeqNumHeartBeatRecv < mSeqNumHeartBeatSent) {
                        mUIHandler.postDelayed(mHeartBeatTimeoutTask, HEART_BEAT_TIMEOUT_MILLIS);
                        // double check
                        if (mSeqNumHeartBeatRecv == mSeqNumHeartBeatSent) {
                            mUIHandler.removeCallbacks(mHeartBeatTimeoutTask);
                        }
                    }
                }

                @Override
                public void onFail(byte[] data, int offset, int len) {
                    // nop
                    // write() 方法会处理失败
                }
            });
        }
    };

    private final Runnable mHeartBeatTimeoutTask = () -> {
        Log.e(TAG, "mHeartBeatTimeoutTask#run: heart beat timeout");
        closeSocket();
    };


    public LongLiveSocket(String host, int port,
                          DataCallback dataCallback, ErrorCallback errorCallback) {
        mHost = host;
        mPort = port;
        mDataCallback = dataCallback;
        mErrorCallback = errorCallback;

        mWriterThread = new HandlerThread("socket-writer");
        mWriterThread.start();
        mWriterHandler = new Handler(mWriterThread.getLooper());
        mWriterHandler.post(this::initSocket);
    }

    private void initSocket() {
        while (true) {
            if (closed()) return;

            try {
                Socket socket = new Socket(mHost, mPort);
                synchronized (mLock) {
                    // 在我们创建 socket 的时候，客户可能就调用了 close()
                    if (mClosed) {
                        silentlyClose(socket);
                        return;
                    }
                    mSocket = socket;
                    // 每次创建新的 socket，会开一个线程来读数据
                    Thread reader = new Thread(new ReaderTask(socket), "socket-reader");
                    reader.start();
                    mWriterHandler.post(mHeartBeatTask);
                }
                break;
            } catch (IOException e) {
                Log.e(TAG, "initSocket: ", e);
                if (closed() || !mErrorCallback.onError()) {
                    break;
                }
                try {
                    TimeUnit.MILLISECONDS.sleep(RETRY_INTERVAL_MILLIS);
                } catch (InterruptedException e1) {
                    // interrupt writer-thread to quit
                    break;
                }
            }
        }
    }

    public void write(byte[] data, WritingCallback callback) {
        write(data, 0, data.length, callback);
    }

    public void write(byte[] data, int offset, int len, WritingCallback callback) {
        mWriterHandler.post(() -> {
            Socket socket = getSocket();
            if (socket == null) {
                // 心跳超时的情况下，这里 socket 会是 null
                initSocket();
                socket = getSocket();
                if (socket == null) {
                    if (!closed()) {
                        callback.onFail(data, offset, len);
                    } /* else {
                        // silently drop the data
                    } */
                    return;
                }
            }
            try {
                OutputStream outputStream = socket.getOutputStream();
                DataOutputStream out = new DataOutputStream(outputStream);
                out.writeInt(len);
                out.write(data, offset, len);
                callback.onSuccess();
            } catch (IOException e) {
                Log.e(TAG, "write: ", e);
                closeSocket();
                callback.onFail(data, offset, len);
                if (!closed() && mErrorCallback.onError()) {
                    initSocket();
                }
            }
        });
    }

    private boolean closed() {
        synchronized (mLock) {
            return mClosed;
        }
    }

    private Socket getSocket() {
        synchronized (mLock) {
            return mSocket;
        }
    }

    private void closeSocket() {
        synchronized (mLock) {
            closeSocketLocked();
        }
    }

    private void closeSocketLocked() {
        if (mSocket == null) return;

        silentlyClose(mSocket);
        mSocket = null;
        mWriterHandler.removeCallbacks(mHeartBeatTask);
    }

    public void close() {
        if (Looper.getMainLooper() == Looper.myLooper()) {
            new Thread() {
                @Override
                public void run() {
                    doClose();
                }
            }.start();
        } else {
            doClose();
        }
    }

    private void doClose() {
        synchronized (mLock) {
            mClosed = true;
            // 关闭 socket，从而使得阻塞在 socket 上的线程返回
            closeSocketLocked();
        }
        mWriterThread.quit();
        // 在重连的时候，有个 sleep
        mWriterThread.interrupt();
    }


    private static void silentlyClose(Closeable closeable) {
        if (closeable != null) {
            try {
                closeable.close();
            } catch (IOException e) {
                Log.e(TAG, "silentlyClose: ", e);
                // error ignored
            }
        }
    }


    private class ReaderTask implements Runnable {

        private final Socket mSocket;

        public ReaderTask(Socket socket) {
            mSocket = socket;
        }

        @Override
        public void run() {
            try {
                readResponse();
            } catch (IOException e) {
                Log.e(TAG, "ReaderTask#run: ", e);
            }
        }

        private void readResponse() throws IOException {
            // For simplicity, assume that a msg will not exceed 1024-byte
            byte[] buffer = new byte[1024];
            InputStream inputStream = mSocket.getInputStream();
            DataInputStream in = new DataInputStream(inputStream);
            while (true) {
                int nbyte = in.readInt();
                if (nbyte == 0) {
                    Log.i(TAG, "readResponse: heart beat received");
                    mUIHandler.removeCallbacks(mHeartBeatTimeoutTask);
                    mSeqNumHeartBeatRecv = mSeqNumHeartBeatSent;
                    continue;
                }

                if (nbyte > buffer.length) {
                    throw new IllegalStateException("Receive message with len " + nbyte +
                            " which exceeds limit " + buffer.length);
                }

                if (readn(in, buffer, nbyte) != 0) {
                    // Socket might be closed twice but it does no harm
                    silentlyClose(mSocket);
                    // Socket will be re-connected by writer-thread if you want
                    break;
                }
                mDataCallback.onData(buffer, 0, nbyte);
            }
        }

        private int readn(InputStream in, byte[] buffer, int n) throws IOException {
            int offset = 0;
            while (n > 0) {
                int readBytes = in.read(buffer, offset, n);
                if (readBytes < 0) {
                    // EoF
                    break;
                }
                n -= readBytes;
                offset += readBytes;
            }
            return n;
        }
    }
}

/**
 * 长连接 服务端
 *
 * Created by xwxwaa on 2019/5/31.
 */

public class LongEchoServer {
    private static final String TAG = "EchoServer";

    private final int mPort;
    private final ExecutorService mExecutorService;

    public LongEchoServer(int port) {
        mPort = port;
        // 固定大小的线程池
        mExecutorService = Executors.newFixedThreadPool(4);
    }

    public void run() {
        mExecutorService.submit(() -> {
            ServerSocket serverSocket;
            try {
                serverSocket = new ServerSocket(mPort);
            } catch (IOException e) {
                Log.e(TAG, "run: ", e);
                return;
            }
            try {
                while (true) {
                    Socket client = serverSocket.accept();
                    handleClient(client);
                }
            } catch (IOException e) {
                Log.e(TAG, "run: ");
            }
        });
    }

    private void handleClient(Socket socket) {
        mExecutorService.submit(() -> {
            try {
                doHandleClient(socket);
            } catch (IOException e) {
                Log.e(TAG, "handleClient: ", e);
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    Log.e(TAG, "handleClient: ", e);
                }
            }
        });
    }

    private void doHandleClient(Socket socket) throws IOException {
        InputStream in = socket.getInputStream();
        OutputStream out = socket.getOutputStream();
        byte[] buffer = new byte[1024];
        int n;
        while ((n = in.read(buffer)) > 0) {
            out.write(buffer, 0, n);
        }
    }

}

/**
 *  长连接 客户端
 *
 * Created by xwxwaa on 2019/5/31.
 */

public class LongEchoClient {

    private static final String TAG = "EchoClient";

    private final LongLiveSocket mLongLiveSocket;

    public LongEchoClient(String host, int port) {
        // 实例化。依次传入主机地址，端口号，并实现写数据的回调，和错误回调（设置为需要重连）
        mLongLiveSocket = new LongLiveSocket(
                host, port,
                (data, offset, len) -> Log.e(TAG, "EchoClient: received: " + new String(data, offset, len)),
                () -> true);
    }

    /**
     * 客户端发送的信息
     */
    public void send(String msg) {
        mLongLiveSocket.write(msg.getBytes(), new LongLiveSocket.WritingCallback() {
            @Override
            public void onSuccess() {
                Log.e(TAG, "onSuccess: ");
            }

            @Override
            public void onFail(byte[] data, int offset, int len) {
                Log.e(TAG, "onFail: fail to write: " + new String(data, offset, len));
                // 失败，则重新发送
                mLongLiveSocket.write(data, offset, len, this);
            }
        });
    }

    public void close() {
        mLongLiveSocket.close();
    }
}

public class MainActivity extends AppCompatActivity {

    /**
     * 长连接 服务端
     */
    private LongEchoServer mEchoServer;
    /**
     * 长连接 客户端
     */
    private LongEchoClient mEchoClient;

    private EditText mMsg;

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

        final int port = 9877;
        mEchoServer = new LongEchoServer(port);
        mEchoServer.run();

        mEchoClient = new LongEchoClient("localhost", port);

        mMsg = findViewById(R.id.msg);
        findViewById(R.id.send).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                String msg = mMsg.getText().toString();
                if (TextUtils.isEmpty(msg)) {
                    return;
                }
                mEchoClient.send(msg);
                mMsg.setText("");
            }
        });
    }
}

关于协议设计

协议版本如何升级？

可通过 IP 协议实现。IP 协议的第一个字段叫 version，目前使用的是 4 或 6，分别表示 IPv4 和 IPv6。由于这个字段在两个版本的IP协议的开头，接收端收到数据后，只要根据第一个字段的值就能够判断这个数据包是 IPv4 还是 IPv6。文本协议（如，JSON、HTML）的情况类似。

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

IP 的头部有个 header length 和 data length 两个字段。通过添加一个 len 域，我们就能够把数据根据应用逻辑分开。
跟这个相对的，还有另一个方案，那就是在数据的末尾放置终止符。比方说，想 C 语言的字符串那样，我们在每个数据的末尾放一个 \0 作为终止符，用以标识一条消息的尾部。这个方法带来的问题是，用户的数据也可能存在 \0。此时，我们就需要对用户的数据进行转义。比方说，把用户数据的所有 \0 都变成 \0\0。读消息的过程总，如果遇到 \0\0，那它就代表 \0，如果只有一个 \0，那就是消息尾部。
使用 len 字段的好处是，我们不需要对数据进行转义。读取数据的时候，只要根据 len 字段，一次性把数据都读进来就好，效率会更高一些。
终止符的方案虽然要求我们对数据进行扫描，但是如果我们可能从任意地方开始读取数据，就需要这个终止符来确定哪里才是消息的开头了。
当然，这两个方法不是互斥的，可以一起使用。

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

IP 在数据报过大的时候，会把一个数据报拆分成多个，并设置一个 MF （more fragments）位，表示这个包只是被拆分后的数据的一部分。
这里，我们也可以给每个文件从 0 开始编号。上传文件的同时，也携带这个编号，并额外附带一个 MF 标志。除了编号最大的文件，所有文件的 MF 标志都置位。因为 MF 没有置位的是最后一个文件，服务器就可以根据这个得出总共有多少个文件。
另一种不使用 MF 标志的方法是，我们在上传文件前，就告诉服务器总共有多少个文件。
如果对数据库比较熟悉，学数据库用事务来处理，也可以。

如何保证数据的有序性

现在有一个任务队列，多个工作线程从中取出任务并执行，执行结果放到一个结果队列中。先要求，放入结果队列的时候，顺序顺序需要跟从工作队列取出时的一样（也就是说，先取出的任务，执行结果需要先放入结果队列）。
我们看看 TCP/IP 是怎么处理的。IP 在发送数据的时候，不同数据报到达对端的时间是不确定的，后面发送的数据有可能较先到达。TCP 为了解决这个问题，给所发送数据的每个字节都赋了一个序列号，通过这个序列号，TCP 就能够把数据按原顺序重新组装。
一样，我们也给每个任务赋一个值，根据进入工作队列的顺序依次递增。工作线程完成任务后，在将结果放入结果队列前，先检查要放入对象的写一个序列号是不是跟自己的任务相同，如果不同，这个结果就不能放进去。此时，最简单的做法是等待，知道下一个可以放入队列的结果是自己所执行的那一个。但是，这个线程就没办法继续处理任务了。
更好的方法是，我们维护多一个结果队列的缓冲，这个缓冲里面的数据按序列号从小到大排序。工作线程要将结果放入，有两种可能：
刚刚完成的任务刚好是下一个，将这个结果放入队列。然后从缓冲的头部开始，将所有可以放入结果队列的数据都放进去。
所完成的任务不能放入结果队列，这个时候就插入结果队列。然后，跟上一种情况一样，需要检查缓冲。
如果测试表明，这个结果缓冲的数据不多，那么使用普通的链表就可以。如果数据比较多，可以使用一个最小堆。

如何保证对方收到了消息

虽说 TCP 提供了可靠的传输，但我们往 socket 写入的数据，只要对端的内核收到后，就会返回 ACK，此时，socket 就认为数据已经写入成功。然而要注意的是，这里只是对方所运行的系统的内核成功收到了数据，并不表示应用程序已经成功处理了数据。
解决办法是学 TCP，添加一个应用层的 APP ACK。应用接收到消息并处理成功后，发送一个 APP ACK 给对方。
有了 APP ACK，需要处理的另一个问题是，如果对方真的没有收到，需要怎么做？
TCP 发送数据的时候，消息一样可能丢失。TCP 发送数据后，如果长时间没有收到对方的 ACK，就假设数据已经丢失，并重新发送。
我们也一样，如果长时间没有收到 APP ACK，就假设数据丢失，重新发送一个。

链接

参考资料：
Socket长连接

传送门：GitHub