前提
之前写过两篇关于socket的文章，但是，只是简单的介绍了一下关于socket Tcp和Udp的简单使用。如果没有看过的朋友可以去看看Android Socket编程（tcp）初探和Android Socket编程（udp）初探。相信很多朋友在公司使用socket开发的时候都会自定义协议来传递信息。一方面是为了安全排除脏数据，另一个方面是为了更加高效的处理自己所需要的数据。今天就来介绍一下关于socket自定义协议和使用Protocol Buffer解析数据。
首先
既然说到了Protocol Buffer，那么我们就简单介绍一下Protocol Buffer是什么？并且使用为什么要使用Protocol Buffer？
一种 结构化数据 的数据存储格式（类似于 XML、Json ），其作用是通过将 结构化的数据 进行 串行化（序列化），从而实现 数据存储 / RPC 数据交换的功能。至于更详细的用法和介绍请移步Protocol Buffer 序列化原理大揭秘 - 为什么Protocol Buffer性能这么好？
其次
说了那么多，我们来看看我们今天的主要内容— 自定义socket协议
先看一张心跳返回的图
request.proto
syntax = "proto3";
// 登录的包体数据
message Request {
int32 uid = 0;
string api_token = 1;
}
发送的格式：
{包头}{命令}{包体}
{包头} -> 包体转成protubuf的长度
{命令} -> 对应功能的命令字参数
{包体} -> 对应的protubuf数据
response.proto
syntax = "proto3";
// 登录成功后服务器返回的包体数据
message Response {
int32 login = 1;
}
服务器返回的格式:
{包头}{命令}{状态码}{包体}
{包头} -> 包体转成protubuf的长度
{命令} -> 对应功能的命令字参数
{状态码} -> 对应状态的状态码
{包体} -> 对应的protubuf数据
2、客户端socket写法
分析：试想一下，要socket不会因为手机屏幕的熄灭或者其他什么的而断开，我们应该把socket放到哪里去写，又要怎么保证socket的连接状态呢？对于Android来说放到 service里面去是最合适的，并且为了保证连接状态。那么，就要发送一个心跳包保证连接状态。既然这样，那么我们来写service和socket。
-
3、service写法
public class SocketService extends Service {
Thread mSocketThread;
Socket mSocket;
InetSocketAddress mSocketAddress;
//心跳线程
Thread mHeartThread;
//接收线程
Thread mReceiveThread;
//登录线程
Thread mLoginThread;
boolean isHeart = false;
boolean isReceive = false;
SocketBinder mBinder = new SocketBinder(this);
public SocketService() {
}
@Override
public void onCreate() {
super.onCreate();
createConnection();
receiveMsg();
isHeart = true;
isReceive = true;
}
@Override
public IBinder onBind(Intent intent) {
return mBinder;
}
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
startGps();
sendHeart();
if (!TextUtils.isEmpty(intent.getStringExtra(AppConfig.SERVICE_TAG))) {
String TAG = intent.getStringExtra(AppConfig.SERVICE_TAG);
switch (TAG) {
case AppConfig.STOP_SERVICE_VALUE: {//停止服务
ClientSocket.getsInstance().shutDownConnection(mSocket);
stopSelf();
mSocket = null;
mHeartThread = null;
mReceiveThread = null;
mLoginThread = null;
mSocketThread = null;
isHeart = false;
isReceive = false;
break;
}
default:
break;
}
}
return super.onStartCommand(intent, flags, startId);
}
/**
* 发送心跳包
*/
private void sendHeart() {
mHeartThread = new Thread(new Runnable() {
@Override
public void run() {
while (isHeart) {
ClientSocket.getsInstance().sendHeart(mSocket, SocketStatus.HEART_CODE);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
mHeartThread.start();
}
/**
* 登录
*/
private void login(final double mLatitude, final double mLongitude) {
mLoginThread = new Thread(new Runnable() {
@Override
public void run() {
if (PreferencesUtils.getInt(SocketService.this, Constants.USER_ID) != 0 &&
!TextUtils.isEmpty(PreferencesUtils.getString(SocketService.this,
Constants.USER_TOKEN))) {
Request.Request requestLogin =
Request.Request.newBuilder()
.setUid(PreferencesUtils.getInt(SocketService.this,
Constants.USER_ID))
.setApiToken(PreferencesUtils.getString(SocketService.this,
Constants.USER_TOKEN).trim())
.build();
ClientSocket.getsInstance().sendLogin(mSocket, requestLogin, SocketStatus.LOGIN_CODE);
}
}
});
mLoginThread.start();
}
/**
* 创建连接
*
* @return
*/
public void createConnection() {
mSocketThread = new Thread(new Runnable() {
@Override
public void run() {
try {
mSocket = new Socket();
mSocketAddress = new InetSocketAddress(AppConfig.TCP_IP, AppConfig.TCP_PORT);
mSocket.connect(mSocketAddress, 20 * 1000);
// 设置 socket 读取数据流的超时时间
mSocket.setSoTimeout(20 * 1000);
// 发送数据包，默认为 false，即客户端发送数据采用 Nagle 算法；
// 但是对于实时交互性高的程序，建议其改为 true，即关闭 Nagle
// 算法，客户端每发送一次数据，无论数据包大小都会将这些数据发送出去
mSocket.setTcpNoDelay(true);
// 设置客户端 socket 关闭时，close() 方法起作用时延迟 30 秒关闭，如果 30 秒内尽量将未发送的数据包发送出去
// socket.setSoLinger(true, 30);
// 设置输出流的发送缓冲区大小，默认是4KB，即4096字节
mSocket.setSendBufferSize(10 * 1024);
// 设置输入流的接收缓冲区大小，默认是4KB，即4096字节
mSocket.setReceiveBufferSize(10 * 1024);
// 作用：每隔一段时间检查服务器是否处于活动状态，如果服务器端长时间没响应，自动关闭客户端socket
// 防止服务器端无效时，客户端长时间处于连接状态
mSocket.setKeepAlive(true);
} catch (UnknownHostException e) {
Logger.e(e.getMessage() + "========+UnknownHostException");
e.printStackTrace();
} catch (IOException e) {
createConnection();
Logger.e(e.getMessage() + "========IOException");
e.printStackTrace();
} catch (NetworkOnMainThreadException e) {
Logger.e(e.getMessage() + "========NetworkOnMainThreadException");
e.printStackTrace();
}
}
});
mSocketThread.start();
}
/**
* 接收
*/
private void receiveMsg() {
mReceiveThread = new Thread(new Runnable() {
@Override
public void run() {
while (isReceive) {
try {
if (mSocket != null && mSocket.isConnected()) {
DataInputStream dis = ClientSocket.getsInstance().getMessageStream(mSocket);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
if (dis != null) {
int length = 0;
int head = 0;
int buffer_size = 4;
byte[] headBuffer = new byte[4];
byte[] cmdBuffer = new byte[4];
byte[] stateBuffer = new byte[4];
length = dis.read(headBuffer, 0, buffer_size);
if (length == 4) {
bos.write(headBuffer, 0, length);
System.arraycopy(bos.toByteArray(), 0, headBuffer, 0, buffer_size);
head = ByteUtil.bytesToInt(headBuffer, 0);
length = dis.read(cmdBuffer, 0, buffer_size);
bos.write(cmdBuffer, 0, length);
System.arraycopy(bos.toByteArray(), 4, cmdBuffer, 0, buffer_size);
int cmd = ByteUtil.hexStringToAlgorism(ByteUtil.str2HexStr(ByteUtil.byte2hex(cmdBuffer)));
int heartNumber = ByteUtil.hexStringToAlgorism(ByteUtil.str2HexStr(SocketStatus.HEART));
String discover = Integer.toHexString(0x0101);
int discoverNumber = ByteUtil.hexStringToAlgorism(ByteUtil.str2HexStr(SocketStatus.DISCOVER));
int giftNumber = ByteUtil.hexStringToAlgorism(ByteUtil.str2HexStr(SocketStatus.GIFT));
if (cmd == heartNumber) {
length = dis.read(stateBuffer, 0, buffer_size);
bos.write(stateBuffer, 0, length);
System.arraycopy(bos.toByteArray(), 8, stateBuffer, 0, buffer_size);
switch (ByteUtil.bytesToInt(stateBuffer, 0)) {
case SocketStatus.LOGIN_SUCCESS: {//登录成功
Logger.e("登录成功");
mLoginValue = "1";
EventUtils.sendEvent(new Event<>(Constants.MSG_LOGIN_SUCCESS));
break;
}
case SocketStatus.HEART_SUCCESS: {//心跳返回
if (ByteUtil.bytesToInt(stateBuffer, 0) == 200
&& Integer.toHexString(ByteUtil.bytesToInt(cmdBuffer, 0))
.equals(discover)) {
byte[] buffer = new byte[head];
length = dis.read(buffer, 0, head);
bos.write(buffer, 0, length);
Response.Response response = Response.
Response.parseFrom(buffer);
Logger.e(responseExplore.getNickname() + responseExplore.getAvatar());
//发送到activity中对数据进行处理
EventUtils.sendEvent(new Event<>(Constants.MSG_START_DISCOVER_RESULT,
responseExplore));
Logger.e(responseExplore + "=======response");
} else {
Logger.e("心跳返回");
}
break;
}
default:
break;
}
}
} else {
//出错重连
ClientSocket.getsInstance().shutDownConnection(mSocket);
createConnection();
}
} else {
createConnection();
}
}
} catch (IOException ex) {
ex.printStackTrace();
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
mReceiveThread.start();
}
@Override
public void onDestroy() {
super.onDestroy();
ClientSocket.getsInstance().shutDownConnection(mSocket);
stopSelf();
mHeartThread = null;
mReceiveThread = null;
mLoginThread = null;
mSocketThread = null;
mStopDiscoverThread = null;
isHeart = false;
isReceive = false;
}
/**
* Binder
*/
public class SocketBinder extends Binder {
private SocketService mService;
public OnServiceCallBack mCallBack;
public SocketBinder(SocketService mService) {
this.mService = mService;
}
/**
* 发送方法
*
* @param object
*/
public void sendMethod(Object object) {
mService.sendMsg(object);
mCallBack.onService(object);
}
/**
* 设置回调
*
* @param callBack
*/
public void setOnServiceCallBack(OnServiceCallBack callBack) {
this.mCallBack = callBack;
}
}
}
分析
上面的service中首先创建socket，然后连接，在socket发生错误的时候（比如网络异常）重新进行创建在连接。然后，开一个接收线程一直接收，每次接收都是接收4个字节的int值进行判断是否可以进入到下一步，如果可以则继续向下。读取4个字节的包头然后读取4个字节的命令 再读取4个字节的状态码 最后读取4个字节的包体，包体就包含我们所需要返回的数据。并且，在刚开始的时候就开启了一个接收线程每隔50毫秒接收一次数据，这样不仅可以读取到心跳包还可以读取到我们需要的数据。在最后，server生命周期结束的时候停止所有的线程。
-
4、发送数据的类
public class ClientSocket {
private DataOutputStream out = null;
private DataInputStream getMessageStream;
private static ClientSocket sInstance;
private ClientSocket() {
}
/**
* 单例
*
* @return
*/
public static ClientSocket getsInstance() {
if (sInstance == null) {
synchronized (ClientSocket.class) {
if (sInstance == null) {
sInstance = new ClientSocket();
}
}
}
return sInstance;
}
/**
* 登录
*
* @return
*/
public void sendLogin(Socket socket, Request.RequestLogin requestLogin, int code) {
byte[] data = requestLogin.toByteArray();
byte[] head = ByteUtil.intToBytes(data.length);
byte[] cmd = ByteUtil.intToBytes(code);
byte[] bytes = addBytes(head, cmd, data);
if (socket != null) {
if (socket.isConnected()) {
try {
OutputStream os = socket.getOutputStream();
os.write(bytes);
os.flush();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 心跳
*
* @param code 关键字（命令）
* @return
*/
public boolean sendHeart(Socket socket, int code) {
boolean isSuccess;
byte[] head = ByteUtil.intToBytes(0);
byte[] cmd = ByteUtil.intToBytes(code);
byte[] bytes = addBytes(head, cmd);
if (socket.isConnected()) {
try {
out = new DataOutputStream(socket.getOutputStream());
out.write(bytes);
out.flush();
isSuccess = true;
} catch (IOException e) {
e.printStackTrace();
isSuccess = false;
}
} else {
isSuccess = false;
}
return isSuccess;
}
/**
* 断开连接
*/
public void shutDownConnection(Socket socket) {
try {
if (out != null) {
out.close();
}
if (getMessageStream != null) {
getMessageStream.close();
}
if (socket != null) {
socket.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 获取服务器返回的流
*
* @param socket
* @return
*/
public DataInputStream getMessageStream(Socket socket) {
if (socket == null) {
return null;
}
if (socket.isClosed()) {
return null;
}
if (!socket.isConnected()) {
return null;
}
try {
getMessageStream = new DataInputStream(new BufferedInputStream(
socket.getInputStream()));
} catch (IOException e) {
e.printStackTrace();
if (getMessageStream != null) {
try {
getMessageStream.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
return getMessageStream;
}
}
分析：
这里使用了单例模式，保证了数据的唯一性，不会重复创建，可以看到登录发送了包头、命令和数据长度，而心跳只是包头和命令，因为包体长度为空，所以不用发送，最后转成4个字节的二进制数据进行发送。这样，proto buffer的优点就体现出来了，方便客户端和服务端的解析。
-
二进制转换工具类
public class ByteUtil {
/**
* 将2个byte数组进行拼接
*/
public static byte[] addBytes(byte[] data1, byte[] data2) {
byte[] data3 = new byte[data1.length + data2.length];
System.arraycopy(data1, 0, data3, 0, data1.length);
System.arraycopy(data2, 0, data3, data1.length, data2.length);
return data3;
}
/**
* 将3个byte数组进行拼接
*/
public static byte[] addBytes(byte[] data1, byte[] data2, byte[] data3) {
byte[] data4 = new byte[data1.length + data2.length + data3.length];
System.arraycopy(data1, 0, data4, 0, data1.length);
System.arraycopy(data2, 0, data4, data1.length, data2.length);
System.arraycopy(data3, 0, data4, data1.length + data2.length, data3.length);
return data4;
}
/**
* int转byte{}
*/
public static byte[] intToBytes(int value, ByteOrder mode) {
byte[] src = new byte[4];
if (mode == ByteOrder.LITTLE_ENDIAN) {
src[3] = (byte) ((value >> 24) & 0xFF);
src[2] = (byte) ((value >> 16) & 0xFF);
src[1] = (byte) ((value >> 8) & 0xFF);
src[0] = (byte) (value & 0xFF);
} else {
src[0] = (byte) ((value >> 24) & 0xFF);
src[1] = (byte) ((value >> 16) & 0xFF);
src[2] = (byte) ((value >> 8) & 0xFF);
src[3] = (byte) (value & 0xFF);
}
return src;
}
/**
* 16进制表示的字符串转换为字节数组
*
* @param s 16进制表示的字符串
* @return byte[] 字节数组
*/
public static byte[] hexStringToByteArray(String s) {
int len = s.length();
byte[] b = new byte[len / 2];
for (int i = 0; i < len; i += 2) {
// 两位一组，表示一个字节,把这样表示的16进制字符串，还原成一个字节
b[i / 2] = (byte) ((Character.digit(s.charAt(i), 16) << 4) + Character
.digit(s.charAt(i + 1), 16));
}
return b;
}
/**
* byte数组中取int数值，本方法适用于(低位在前，高位在后)的顺序，和和intToBytes（）配套使用
*
* @param src byte数组
* @param offset 从数组的第offset位开始
* @return int数值
*/
public static int bytesToInt(byte[] src, int offset) {
int value;
value = (int) ((src[offset] & 0xFF)
| ((src[offset + 1] & 0xFF) << 8)
| ((src[offset + 2] & 0xFF) << 16)
| ((src[offset + 3] & 0xFF) << 24));
return value;
}
/**
* byte数组中取int数值，本方法适用于(低位在后，高位在前)的顺序。和intToBytes2（）配套使用
*/
public static int bytesToInt2(byte[] src, int offset) {
int value;
value = (int) (((src[offset] & 0xFF) << 24)
| ((src[offset + 1] & 0xFF) << 16)
| ((src[offset + 2] & 0xFF) << 8)
| (src[offset + 3] & 0xFF));
return value;
}
/**
* 将int数值转换为占四个字节的byte数组，本方法适用于(低位在前，高位在后)的顺序。 和
bytesToInt（）配套使用
*
* @param value 要转换的int值
* @return byte数组
*/
public static byte[] intToBytes(int value) {
byte[] src = new byte[4];
src[3] = (byte) ((value >> 24) & 0xFF);
src[2] = (byte) ((value >> 16) & 0xFF);
src[1] = (byte) ((value >> 8) & 0xFF);
src[0] = (byte) (value & 0xFF);
return src;
}
/**
* 将int数值转换为占四个字节的byte数组，本方法适用于(高位在前，低位在后)的顺序。 和
bytesToInt2（）配套使用
*/
public static byte[] intToBytes2(int value) {
byte[] src = new byte[4];
src[0] = (byte) ((value >> 24) & 0xFF);
src[1] = (byte) ((value >> 16) & 0xFF);
src[2] = (byte) ((value >> 8) & 0xFF);
src[3] = (byte) (value & 0xFF);
return src;
}
/**
* 将字节转换为二进制字符串
*
* @param bytes 字节数组
* @return 二进制字符串
*/
public static String byteToBit(byte... bytes) {
StringBuffer sb = new StringBuffer();
int z, len;
String str;
for (int w = 0; w < bytes.length; w++) {
z = bytes[w];
z |= 256;
str = Integer.toBinaryString(z);
len = str.length();
sb.append(str.substring(len - 8, len));
}
return sb.toString();
}
/**
* 字节数组转为普通字符串（ASCII对应的字符）
*
* @param bytearray byte[]
* @return String
*/
public static String byte2String(byte[] bytearray) {
String result = "";
char temp;
int length = bytearray.length;
for (int i = 0; i < length; i++) {
temp = (char) bytearray[i];
result += temp;
}
return result;
}
/**
* 二进制字符串转十进制
*
* @param binary 二进制字符串
* @return 十进制数值
*/
public static int binaryToAlgorism(String binary) {
int max = binary.length();
int result = 0;
for (int i = max; i > 0; i--) {
char c = binary.charAt(i - 1);
int algorism = c - '0';
result += Math.pow(2, max - i) * algorism;
}
return result;
}
/**
* 字节数组转换为十六进制字符串
*
* @param b byte[] 需要转换的字节数组
* @return String 十六进制字符串
*/
public static String byte2hex(byte b[]) {
if (b == null) {
throw new IllegalArgumentException(
"Argument b ( byte array ) is null! ");
}
String hs = "";
String stmp = "";
for (int n = 0; n < b.length; n++) {
stmp = Integer.toHexString(b[n] & 0xff);
if (stmp.length() == 1) {
hs = hs + "0" + stmp;
} else {
hs = hs + stmp;
}
}
return hs.toUpperCase();
}
/**
* 十六进制字符串转换十进制
*
* @param hex 十六进制字符串
* @return 十进制数值
*/
public static int hexStringToAlgorism(String hex) {
hex = hex.toUpperCase();
int max = hex.length();
int result = 0;
for (int i = max; i > 0; i--) {
char c = hex.charAt(i - 1);
int algorism = 0;
if (c >= '0' && c <= '9') {
algorism = c - '0';
} else {
algorism = c - 55;
}
result += Math.pow(16, max - i) * algorism;
}
return result;
}
/**
* 字符串转换成十六进制字符串
*
* @param str 待转换的ASCII字符串
* @return String 每个Byte之间空格分隔，如: [61 6C 6B]
*/
public static String str2HexStr(String str) {
char[] chars = "0123456789ABCDEF".toCharArray();
StringBuilder sb = new StringBuilder("");
byte[] bs = str.getBytes();
int bit;
for (int i = 0; i < bs.length; i++) {
bit = (bs[i] & 0x0f0) >> 4;
sb.append(chars[bit]);
bit = bs[i] & 0x0f;
sb.append(chars[bit]);
sb.append(' ');
}
return sb.toString().trim();
}
/**
* 16进制转换成字符串
*
* @param hexStr
* @return
*/
public static String hexStr2Str(String hexStr) {
String str = "0123456789ABCDEF";
char[] hexs = hexStr.toCharArray();
byte[] bytes = new byte[hexStr.length() / 2];
int n;
for (int i = 0; i < bytes.length; i++) {
n = str.indexOf(hexs[2 * i]) * 16;
n += str.indexOf(hexs[2 * i + 1]);
bytes[i] = (byte) (n & 0xff);
}
return new String(bytes);
}
/**
* 重写了Inpustream 中的skip(long n) 方法，
* 将数据流中起始的n 个字节跳过
*/
public static long skipBytesFromStream(InputStream inputStream, long n) {
long remaining = n;
// SKIP_BUFFER_SIZE is used to determine the size of skipBuffer
int SKIP_BUFFER_SIZE = 2048;
// skipBuffer is initialized in skip(long), if needed.
byte[] skipBuffer = null;
int nr = 0;
if (skipBuffer == null) {
skipBuffer = new byte[SKIP_BUFFER_SIZE];
}
byte[] localSkipBuffer = skipBuffer;
if (n <= 0) {
return 0;
}
while (remaining > 0) {
try {
nr = inputStream.read(localSkipBuffer, 0,
(int) Math.min(SKIP_BUFFER_SIZE, remaining));
} catch (IOException e) {
e.printStackTrace();
}
if (nr < 0) {
break;
}
remaining -= nr;
}
return n - remaining;
}
}
最后
对于socket和 proto buffer来说，能灵活运用，首先要感谢我们公司的同事，没有他们提供思路，估计很难灵活运用socket和proto buffer。其次，要感谢之前公司的大佬，还有给我提供宝贵意见的各位好友。还有要感谢自己，能静下心来，坚持不懈，克服proto buffer和socket相结合的写法。
感谢
Protocol Buffer技术详解(语言规范)
Protocol Buffer 序列化原理大揭秘 - 为什么Protocol Buffer性能这么好？
