前提
之前写过两篇关于socket的文章，但是，只是简单的介绍了一下关于socket Tcp和Udp的简单使用。如果没有看过的朋友可以去看看Android Socket编程（tcp）初探和Android Socket编程（udp）初探。相信很多朋友在公司使用socket开发的时候都会自定义协议来传递信息。一方面是为了安全排除脏数据，另一个方面是为了更加高效的处理自己所需要的数据。今天就来介绍一下关于socket自定义协议和使用Protocol Buffer解析数据。
首先
既然说到了Protocol Buffer，那么我们就简单介绍一下Protocol Buffer是什么？并且使用为什么要使用Protocol Buffer？
- 1、什么是Protocol Buffer
一种 结构化数据 的数据存储格式（类似于 XML、Json ），其作用是通过将 结构化的数据 进行 串行化（序列化），从而实现 数据存储 / RPC 数据交换的功能。至于更详细的用法和介绍请移步Protocol Buffer 序列化原理大揭秘 - 为什么Protocol Buffer性能这么好？
2、为什么要使用Protocol Buffer
在回答这个问题之前，我们还是先给出一个在实际开发中经常会遇到的系统场景。比如：我们的客户端程序是使用Java开发的，可能运行自不同的平台，如：Linux、Windows或者是Android，而我们的服务器程序通常是基于Linux平台并使用C++或者Python开发完成的。在这两种程序之间进行数据通讯时存在多种方式用于设计消息格式，如：
1、 直接传递C/C++/Python语言中一字节对齐的结构体数据，只要结构体的声明为定长格式，那么该方式对于C/C++/Python程序而言就非常方便了，仅需将接收到的数据按照结构体类型强行转换即可。事实上对于变长结构体也不会非常麻烦。在发送数据时，也只需定义一个结构体变量并设置各个成员变量的值之后，再以char*的方式将该二进制数据发送到远端。反之，该方式对于Java开发者而言就会非常繁琐，首先需要将接收到的数据存于ByteBuffer之中，再根据约定的字节序逐个读取每个字段，并将读取后的值再赋值给另外一个值对象中的域变量，以便于程序中其他代码逻辑的编写。对于该类型程序而言，联调的基准是必须客户端和服务器双方均完成了消息报文构建程序的编写后才能展开，而该设计方式将会直接导致Java程序开发的进度过慢。即便是Debug阶段，也会经常遇到Java程序中出现各种域字段拼接的小错误。 2、 使用SOAP协议(WebService)作为消息报文的格式载体，由该方式生成的报文是基于文本格式的，同时还存在大量的XML描述信息，因此将会大大增加网络IO的负担。又由于XML解析的复杂性，这也会大幅降低报文解析的性能。总之，使用该设计方式将会使系统的整体运行性能明显下降。 对于以上两种方式所产生的问题，Protocol Buffer均可以很好的解决，不仅如此，Protocol Buffer还有一个非常重要的优点就是可以保证同一消息报文新旧版本之间的兼容性。__对于Protocol Buffer具体的用法请移步[Protocol Buffer技术详解(语言规范)](https://www.cnblogs.com/stephen-liu74/archive/2013/01/02/2841485.html)今天主要讲解的是socket自定义协议这块__
其次
说了那么多，我们来看看我们今天的主要内容— 自定义socket协议
先看一张心跳返回的图
- 1、Protobuf协议
- 假设客户端请求包体数据协议如下
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性能这么好？