Android TCP封装工具类

TCP通信的封装，我们可以从以下几个方面进行改进：

线程池优化：使用更高效的线程池配置，避免频繁创建和销毁线程。

连接重试机制：在网络不稳定时，自动重试连接。

心跳机制：保持长连接，避免因超时断开。

数据缓冲区优化：动态调整缓冲区大小，适应不同数据量。

异常处理增强：区分不同类型的异常，提供更详细的错误信息。

代码简洁性：减少冗余代码，提高可读性和可维护性。

TCP客户端封装（Java）

import android.util.Log;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class TcpClient {private static final String TAG = "TcpClient";private static final int HEARTBEAT_INTERVAL = 10; // 心跳间隔（秒）private static final int CONNECT_TIMEOUT = 5000; // 连接超时时间（毫秒）private static final int RECONNECT_DELAY = 3000; // 重连延迟时间（毫秒）private Socket socket;private InputStream inputStream;private OutputStream outputStream;private ExecutorService executorService;private ScheduledExecutorService heartbeatExecutor;private boolean isConnected = false;private TcpListener listener;private String serverIp;private int serverPort;public TcpClient(TcpListener listener) {this.listener = listener;executorService = Executors.newCachedThreadPool();heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();}/*** 连接到服务器** @param ip   服务器IP地址* @param port 服务器端口*/public void connect(String ip, int port) {this.serverIp = ip;this.serverPort = port;executorService.execute(this::connectInternal);}private void connectInternal() {try {// 创建Socket并连接服务器socket = new Socket();socket.connect(new InetSocketAddress(serverIp, serverPort), CONNECT_TIMEOUT);inputStream = socket.getInputStream();outputStream = socket.getOutputStream();isConnected = true;// 通知连接成功if (listener != null) {listener.onConnected();}// 开始接收数据receiveData();// 启动心跳机制startHeartbeat();} catch (IOException e) {Log.e(TAG, "Connection failed: " + e.getMessage());if (listener != null) {listener.onError("Connection failed: " + e.getMessage());}scheduleReconnect();}}/*** 断开连接*/public void disconnect() {executorService.execute(() -> {try {if (socket != null) {socket.close();}if (inputStream != null) {inputStream.close();}if (outputStream != null) {outputStream.close();}isConnected = false;// 通知断开连接if (listener != null) {listener.onDisconnected();}} catch (IOException e) {Log.e(TAG, "Disconnect error: " + e.getMessage());} finally {stopHeartbeat();}});}/*** 发送数据** @param data 要发送的数据*/public void sendData(byte[] data) {if (!isConnected || outputStream == null) {Log.e(TAG, "Not connected to server");return;}executorService.execute(() -> {try {outputStream.write(data);outputStream.flush();Log.d(TAG, "Data sent successfully");} catch (IOException e) {Log.e(TAG, "Failed to send data: " + e.getMessage());if (listener != null) {listener.onError("Failed to send data: " + e.getMessage());}disconnect();}});}/*** 接收数据*/private void receiveData() {executorService.execute(() -> {byte[] buffer = new byte[1024];int bytesRead;while (isConnected) {try {bytesRead = inputStream.read(buffer);if (bytesRead == -1) {// 服务器关闭连接disconnect();break;}// 处理接收到的数据byte[] receivedData = new byte[bytesRead];System.arraycopy(buffer, 0, receivedData, 0, bytesRead);// 通知数据接收if (listener != null) {listener.onDataReceived(receivedData);}} catch (IOException e) {Log.e(TAG, "Failed to receive data: " + e.getMessage());if (listener != null) {listener.onError("Failed to receive data: " + e.getMessage());}disconnect();break;}}});}/*** 启动心跳机制*/private void startHeartbeat() {heartbeatExecutor.scheduleAtFixedRate(() -> {if (isConnected) {sendData("HEARTBEAT".getBytes()); // 发送心跳包}}, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);}/*** 停止心跳机制*/private void stopHeartbeat() {heartbeatExecutor.shutdown();}/*** 安排重连*/private void scheduleReconnect() {executorService.schedule(this::connectInternal, RECONNECT_DELAY, TimeUnit.MILLISECONDS);}/*** 是否已连接*/public boolean isConnected() {return isConnected;}/*** 关闭线程池*/public void shutdown() {executorService.shutdown();heartbeatExecutor.shutdown();}/*** TCP事件监听器*/public interface TcpListener {void onConnected(); // 连接成功void onDisconnected(); // 断开连接void onDataReceived(byte[] data); // 接收到数据void onError(String error); // 发生错误}
}

2. 在Activity中使用

import android.os.Bundle;
import android.util.Log;
import androidx.appcompat.app.AppCompatActivity;public class MainActivity extends AppCompatActivity implements TcpClient.TcpListener {private static final String SERVER_IP = "192.168.1.100"; // 服务器IPprivate static final int SERVER_PORT = 8080; // 服务器端口private TcpClient tcpClient;@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_main);// 初始化TCP客户端tcpClient = new TcpClient(this);// 连接到服务器tcpClient.connect(SERVER_IP, SERVER_PORT);// 发送数据String message = "Hello, Server!";tcpClient.sendData(message.getBytes());}@Overridepublic void onConnected() {Log.d("TcpClient", "Connected to server");}@Overridepublic void onDisconnected() {Log.d("TcpClient", "Disconnected from server");}@Overridepublic void onDataReceived(byte[] data) {String message = new String(data);Log.d("TcpClient", "Received data: " + message);}@Overridepublic void onError(String error) {Log.e("TcpClient", "Error: " + error);}@Overrideprotected void onDestroy() {super.onDestroy();// 断开连接并释放资源if (tcpClient != null) {tcpClient.disconnect();tcpClient.shutdown();}}
}

进一步优化（Kotlin版本）

import android.os.Bundle
import android.util.Log
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.*
import java.io.IOException
import java.io.InputStream
import java.io.OutputStream
import java.net.InetSocketAddress
import java.net.Socketclass MainActivity : AppCompatActivity(), TcpClient.TcpListener {private val serverIp = "192.168.1.100" // 服务器IPprivate val serverPort = 8080 // 服务器端口private lateinit var tcpClient: TcpClientoverride fun onCreate(savedInstanceState: Bundle?) {super.onCreate(savedInstanceState)setContentView(R.layout.activity_main)// 初始化TCP客户端tcpClient = TcpClient(this)// 连接到服务器tcpClient.connect(serverIp, serverPort)// 发送数据val message = "Hello, Server!"tcpClient.sendData(message.toByteArray())}override fun onConnected() {Log.d("TcpClient", "Connected to server")}override fun onDisconnected() {Log.d("TcpClient", "Disconnected from server")}override fun onDataReceived(data: ByteArray) {val message = String(data)Log.d("TcpClient", "Received data: $message")}override fun onError(error: String) {Log.e("TcpClient", "Error: $error")}override fun onDestroy() {super.onDestroy()// 断开连接并释放资源tcpClient.disconnect()tcpClient.shutdown()}
}class TcpClient(private val listener: TcpListener) {private var socket: Socket? = nullprivate var inputStream: InputStream? = nullprivate var outputStream: OutputStream? = nullprivate var isConnected = falseprivate val scope = CoroutineScope(Dispatchers.IO)private var heartbeatJob: Job? = nullfun connect(ip: String, port: Int) {scope.launch {try {socket = Socket().apply {connect(InetSocketAddress(ip, port), 5000) // 5秒超时}inputStream = socket?.getInputStream()outputStream = socket?.getOutputStream()isConnected = truewithContext(Dispatchers.Main) {listener.onConnected()}receiveData()startHeartbeat()} catch (e: IOException) {withContext(Dispatchers.Main) {listener.onError("Connection failed: ${e.message}")}scheduleReconnect()}}}fun disconnect() {scope.launch {try {socket?.close()inputStream?.close()outputStream?.close()isConnected = falsewithContext(Dispatchers.Main) {listener.onDisconnected()}} catch (e: IOException) {withContext(Dispatchers.Main) {listener.onError("Disconnect error: ${e.message}")}} finally {stopHeartbeat()}}}fun sendData(data: ByteArray) {if (!isConnected || outputStream == null) {Log.e("TcpClient", "Not connected to server")return}scope.launch {try {outputStream?.write(data)outputStream?.flush()Log.d("TcpClient", "Data sent successfully")} catch (e: IOException) {withContext(Dispatchers.Main) {listener.onError("Failed to send data: ${e.message}")}disconnect()}}}private fun receiveData() {scope.launch {val buffer = ByteArray(1024)var bytesRead: Intwhile (isConnected) {try {bytesRead = inputStream?.read(buffer) ?: -1if (bytesRead == -1) {disconnect()break}val receivedData = buffer.copyOf(bytesRead)withContext(Dispatchers.Main) {listener.onDataReceived(receivedData)}} catch (e: IOException) {withContext(Dispatchers.Main) {listener.onError("Failed to receive data: ${e.message}")}disconnect()break}}}}private fun startHeartbeat() {heartbeatJob = scope.launch {while (isConnected) {sendData("HEARTBEAT".toByteArray())delay(10000) // 10秒间隔}}}private fun stopHeartbeat() {heartbeatJob?.cancel()}private fun scheduleReconnect() {scope.launch {delay(3000) // 3秒后重连connect(socket?.inetAddress?.hostAddress ?: "", socket?.port ?: 0)}}fun shutdown() {scope.cancel()}interface TcpListener {fun onConnected()fun onDisconnected()fun onDataReceived(data: ByteArray)fun onError(error: String)}
}