基于Java Socket的自定义协议,实现Android与服务器的长连接(二)-阿里云开发者社区

开发者社区> 玄学酱> 正文

基于Java Socket的自定义协议,实现Android与服务器的长连接(二)

简介:
+关注继续查看

在阅读本文前需要对socket以及自定义协议有一个基本的了解,可以先查看上一篇文章《基于Java Socket的自定义协议,实现Android与服务器的长连接(一)》学习相关的基础知识点。

一、协议定义

上一篇文章中,我们对socket编程和自定义协议做了一个简单的了解,本文将在此基础上加以深入,来实现Android和服务器之间的长连接,现定义协议如下:

  • 数据类协议(Data)
    • 长度(length,32bit)
    • 版本号(version,8bit,前3位预留,后5位用于表示真正的版本号)
    • 数据类型(type,8bit,0表示数据)
    • 业务类型(pattion,8bit,0表示push,其他暂未定)
    • 数据格式(dtype,8bit,0表示json,其他暂未定)
    • 消息id(msgId,32bit)
    • 正文数据(data)
  • 数据ack类协议(DataAck)
    • 长度(length,32bit)
    • 版本号(version,8bit,前3位预留,后5位用于表示真正的版本号)
    • 数据类型(type,8bit,1表示数据ack)
    • ack消息id(ackMsgId,32bit)
    • 预留信息(unused)
  • 心跳类协议(ping)
    • 长度(length,32bit)
    • 版本号(version,8bit,前3位预留,后5位用于表示真正的版本号)
    • 数据类型(type,8bit,2表示心跳)
    • 心跳id(pingId,32bit,client上报取奇数,即1,3,5...,server下发取偶数,即0,2,4...)
    • 预留信息(unused)
  • 心跳ack类协议(pingAck)
    • 长度(length,32bit)
    • 版本号(version,8bit,前3位预留,后5位用于表示真正的版本号)
    • 数据类型(type,8bit,3表示心跳ack)
    • ack心跳id(pingId,32bit,client上报取奇数,即1,3,5...,server下发取偶数,即0,2,4...)
    • 预留信息(unused)

二、协议实现

从上述的协议定义中,我们可以看出,四种协议有共同的3个要素,分别是:长度、版本号、数据类型,那么我们可以先抽象出一个基本的协议,如下:

1. BasicProtocol


  1. import android.util.Log; 
  2.  
  3. import com.shandiangou.sdgprotocol.lib.Config; 
  4. import com.shandiangou.sdgprotocol.lib.ProtocolException; 
  5. import com.shandiangou.sdgprotocol.lib.SocketUtil; 
  6.  
  7. import java.io.ByteArrayOutputStream; 
  8.  
  9. /** 
  10.  * Created by meishan on 16/12/1. 
  11.  * <p> 
  12.  * 协议类型: 0表示数据,1表示数据Ack,2表示ping,3表示pingAck 
  13.  */ 
  14. public abstract class BasicProtocol { 
  15.  
  16.     // 长度均以字节(byte)为单位 
  17.     public static final int LENGTH_LEN = 4;       //记录整条数据长度数值的长度 
  18.     protected static final int VER_LEN = 1;       //协议的版本长度(其中前3位作为预留位,后5位作为版本号) 
  19.     protected static final int TYPE_LEN = 1;      //协议的数据类型长度 
  20.  
  21.     private int reserved = 0;                     //预留信息 
  22.     private int version = Config.VERSION;         //版本号 
  23.  
  24.     /** 
  25.      * 获取整条数据长度 
  26.      * 单位:字节(byte) 
  27.      * 
  28.      * @return 
  29.      */ 
  30.     protected int getLength() { 
  31.         return LENGTH_LEN + VER_LEN + TYPE_LEN; 
  32.     } 
  33.  
  34.     public int getReserved() { 
  35.         return reserved; 
  36.     } 
  37.  
  38.     public void setReserved(int reserved) { 
  39.         this.reserved = reserved; 
  40.     } 
  41.  
  42.     public int getVersion() { 
  43.         return version; 
  44.     } 
  45.  
  46.     public void setVersion(int version) { 
  47.         this.version = version; 
  48.     } 
  49.  
  50.     /** 
  51.      * 获取协议类型,由子类实现 
  52.      * 
  53.      * @return 
  54.      */ 
  55.     public abstract int getProtocolType(); 
  56.      
  57.     /** 
  58.      * 由预留值和版本号计算完整版本号的byte[]值 
  59.      * 
  60.      * @return 
  61.      */ 
  62.     private int getVer(byte r, byte v, int vLen) { 
  63.         int num = 0; 
  64.         int rLen = 8 - vLen; 
  65.         for (int i = 0; i < rLen; i++) { 
  66.             num += (((r >> (rLen - 1 - i)) & 0x1) << (7 - i)); 
  67.         } 
  68.         return num + v; 
  69.     } 
  70.  
  71.     /** 
  72.      * 拼接发送数据,此处拼接了协议版本、协议类型和数据长度,具体内容子类中再拼接 
  73.      * 按顺序拼接 
  74.      * 
  75.      * @return 
  76.      */ 
  77.     public byte[] genContentData() { 
  78.         byte[] length = SocketUtil.int2ByteArrays(getLength()); 
  79.         byte reserved = (byte) getReserved(); 
  80.         byte version = (byte) getVersion(); 
  81.         byte[] ver = {(byte) getVer(reserved, version, 5)}; 
  82.         byte[] type = {(byte) getProtocolType()}; 
  83.  
  84.         ByteArrayOutputStream baos = new ByteArrayOutputStream(LENGTH_LEN + VER_LEN + TYPE_LEN); 
  85.         baos.write(length, 0, LENGTH_LEN); 
  86.         baos.write(ver, 0, VER_LEN); 
  87.         baos.write(type, 0, TYPE_LEN); 
  88.         return baos.toByteArray(); 
  89.     } 
  90.  
  91.     /** 
  92.      * 解析出整条数据长度 
  93.      * 
  94.      * @param data 
  95.      * @return 
  96.      */ 
  97.     protected int parseLength(byte[] data) { 
  98.         return SocketUtil.byteArrayToInt(data, 0, LENGTH_LEN); 
  99.     } 
  100.  
  101.     /** 
  102.      * 解析出预留位 
  103.      * 
  104.      * @param data 
  105.      * @return 
  106.      */ 
  107.     protected int parseReserved(byte[] data) { 
  108.         byte r = data[LENGTH_LEN];//前4个字节(0,1,2,3)为数据长度的int值,与版本号组成一个字节 
  109.         return (r >> 5) & 0xFF; 
  110.     } 
  111.  
  112.     /** 
  113.      * 解析出版本号 
  114.      * 
  115.      * @param data 
  116.      * @return 
  117.      */ 
  118.     protected int parseVersion(byte[] data) { 
  119.         byte v = data[LENGTH_LEN]; //与预留位组成一个字节 
  120.         return ((v << 3) & 0xFF) >> 3; 
  121.     } 
  122.  
  123.     /** 
  124.      * 解析出协议类型 
  125.      * 
  126.      * @param data 
  127.      * @return 
  128.      */ 
  129.     public static int parseType(byte[] data) { 
  130.         byte t = data[LENGTH_LEN + VER_LEN];//前4个字节(0,1,2,3)为数据长度的int值,以及ver占一个字节 
  131.         return t & 0xFF; 
  132.     } 
  133.  
  134.     /** 
  135.      * 解析接收数据,此处解析了协议版本、协议类型和数据长度,具体内容子类中再解析 
  136.      * 
  137.      * @param data 
  138.      * @return 
  139.      * @throws ProtocolException 协议版本不一致,抛出异常 
  140.      */ 
  141.     public int parseContentData(byte[] data) throws ProtocolException { 
  142.         int reserved = parseReserved(data); 
  143.         int version = parseVersion(data); 
  144.         int protocolType = parseType(data); 
  145.         if (version != getVersion()) { 
  146.             throw new ProtocolException("input version is error: " + version); 
  147.         } 
  148.         return LENGTH_LEN + VER_LEN + TYPE_LEN; 
  149.     } 
  150.  
  151.     @Override 
  152.     public String toString() { 
  153.         return "Version: " + getVersion() + ", Type: " + getProtocolType(); 
  154.     } 
  155.  

上述涉及到的Config类和SocketUtil类如下:


  1. /** 
  2.  * Created by meishan on 16/12/2. 
  3.  */ 
  4. public class Config { 
  5.  
  6.     public static final int VERSION = 1;                 //协议版本号 
  7.     public static final String ADDRESS = "10.17.64.237"; //服务器地址 
  8.     public static final int PORT = 9013;                 //服务器端口号 
  9.      
  10.  

  1. import java.io.BufferedInputStream; 
  2. import java.io.BufferedOutputStream; 
  3. import java.io.IOException; 
  4. import java.io.InputStream; 
  5. import java.io.OutputStream; 
  6. import java.nio.ByteBuffer; 
  7. import java.util.HashMap; 
  8. import java.util.Map; 
  9.  
  10. /** 
  11.  * Created by meishan on 16/12/1. 
  12.  */ 
  13. public class SocketUtil { 
  14.  
  15.     private static Map<Integer, String> msgImp = new HashMap<>(); 
  16.  
  17.     static { 
  18.         msgImp.put(DataProtocol.PROTOCOL_TYPE, "com.shandiangou.sdgprotocol.lib.protocol.DataProtocol");       //0 
  19.         msgImp.put(DataAckProtocol.PROTOCOL_TYPE, "com.shandiangou.sdgprotocol.lib.protocol.DataAckProtocol"); //1 
  20.         msgImp.put(PingProtocol.PROTOCOL_TYPE, "com.shandiangou.sdgprotocol.lib.protocol.PingProtocol");       //2 
  21.         msgImp.put(PingAckProtocol.PROTOCOL_TYPE, "com.shandiangou.sdgprotocol.lib.protocol.PingAckProtocol"); //3 
  22.     } 
  23.  
  24.     /** 
  25.      * 解析数据内容 
  26.      * 
  27.      * @param data 
  28.      * @return 
  29.      */ 
  30.     public static BasicProtocol parseContentMsg(byte[] data) { 
  31.         int protocolType = BasicProtocol.parseType(data); 
  32.         String className = msgImp.get(protocolType); 
  33.         BasicProtocol basicProtocol; 
  34.         try { 
  35.             basicProtocol = (BasicProtocol) Class.forName(className).newInstance(); 
  36.             basicProtocol.parseContentData(data); 
  37.         } catch (Exception e) { 
  38.             basicProtocol = null
  39.             e.printStackTrace(); 
  40.         } 
  41.         return basicProtocol; 
  42.     } 
  43.  
  44.     /** 
  45.      * 读数据 
  46.      * 
  47.      * @param inputStream 
  48.      * @return 
  49.      * @throws SocketExceptions 
  50.      */ 
  51.     public static BasicProtocol readFromStream(InputStream inputStream) { 
  52.         BasicProtocol protocol; 
  53.         BufferedInputStream bis; 
  54.          
  55.         //header中保存的是整个数据的长度值,4个字节表示。在下述write2Stream方法中,会先写入header 
  56.         byte[] header = new byte[BasicProtocol.LENGTH_LEN]; 
  57.  
  58.         try { 
  59.             bis = new BufferedInputStream(inputStream); 
  60.  
  61.             int temp
  62.             int len = 0; 
  63.             while (len < header.length) { 
  64.                 temp = bis.read(header, len, header.length - len); 
  65.                 if (temp > 0) { 
  66.                     len += temp
  67.                 } else if (temp == -1) { 
  68.                     bis.close(); 
  69.                     return null
  70.                 } 
  71.             } 
  72.  
  73.             len = 0; 
  74.             int length = byteArrayToInt(header);//数据的长度值 
  75.             byte[] content = new byte[length]; 
  76.             while (len < length) { 
  77.                 temp = bis.read(content, len, length - len); 
  78.  
  79.                 if (temp > 0) { 
  80.                     len += temp
  81.                 } 
  82.             } 
  83.  
  84.             protocol = parseContentMsg(content); 
  85.         } catch (IOException e) { 
  86.             e.printStackTrace(); 
  87.             return null
  88.         } 
  89.  
  90.         return protocol; 
  91.     } 
  92.  
  93.     /** 
  94.      * 写数据 
  95.      * 
  96.      * @param protocol 
  97.      * @param outputStream 
  98.      */ 
  99.     public static void write2Stream(BasicProtocol protocol, OutputStream outputStream) { 
  100.         BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(outputStream); 
  101.         byte[] buffData = protocol.genContentData(); 
  102.         byte[] header = int2ByteArrays(buffData.length); 
  103.         try { 
  104.             bufferedOutputStream.write(header); 
  105.             bufferedOutputStream.write(buffData); 
  106.             bufferedOutputStream.flush(); 
  107.         } catch (IOException e) { 
  108.             e.printStackTrace(); 
  109.         } 
  110.     } 
  111.  
  112.     /** 
  113.      * 关闭输入流 
  114.      * 
  115.      * @param is 
  116.      */ 
  117.     public static void closeInputStream(InputStream is) { 
  118.         try { 
  119.             if (is != null) { 
  120.                 is.close(); 
  121.             } 
  122.         } catch (IOException e) { 
  123.             e.printStackTrace(); 
  124.         } 
  125.     } 
  126.  
  127.     /** 
  128.      * 关闭输出流 
  129.      * 
  130.      * @param os 
  131.      */ 
  132.     public static void closeOutputStream(OutputStream os) { 
  133.         try { 
  134.             if (os != null) { 
  135.                 os.close(); 
  136.             } 
  137.         } catch (IOException e) { 
  138.             e.printStackTrace(); 
  139.         } 
  140.     } 
  141.  
  142.     public static byte[] int2ByteArrays(int i) { 
  143.         byte[] result = new byte[4]; 
  144.         result[0] = (byte) ((i >> 24) & 0xFF); 
  145.         result[1] = (byte) ((i >> 16) & 0xFF); 
  146.         result[2] = (byte) ((i >> 8) & 0xFF); 
  147.         result[3] = (byte) (i & 0xFF); 
  148.         return result; 
  149.     } 
  150.  
  151.     public static int byteArrayToInt(byte[] b) { 
  152.         int intValue = 0; 
  153.         for (int i = 0; i < b.length; i++) { 
  154.             intValue += (b[i] & 0xFF) << (8 * (3 - i)); //int占4个字节(0,1,2,3) 
  155.         } 
  156.         return intValue; 
  157.     } 
  158.  
  159.     public static int byteArrayToInt(byte[] b, int byteOffset, int byteCount) { 
  160.         int intValue = 0; 
  161.         for (int i = byteOffset; i < (byteOffset + byteCount); i++) { 
  162.             intValue += (b[i] & 0xFF) << (8 * (3 - (i - byteOffset))); 
  163.         } 
  164.         return intValue; 
  165.     } 
  166.  
  167.     public static int bytes2Int(byte[] b, int byteOffset) { 
  168.         ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.SIZE / Byte.SIZE); 
  169.         byteBuffer.put(b, byteOffset, 4); //占4个字节 
  170.         byteBuffer.flip(); 
  171.         return byteBuffer.getInt(); 
  172.     } 
  173.  

接下来我们实现具体的协议。

2. DataProtocol


  1. import android.util.Log; 
  2.  
  3. import com.shandiangou.sdgprotocol.lib.ProtocolException; 
  4. import com.shandiangou.sdgprotocol.lib.SocketUtil; 
  5.  
  6. import java.io.ByteArrayOutputStream; 
  7. import java.io.Serializable
  8. import java.io.UnsupportedEncodingException; 
  9.  
  10. /** 
  11.  * Created by meishan on 16/12/1. 
  12.  */ 
  13. public class DataProtocol extends BasicProtocol implements Serializable { 
  14.  
  15.     public static final int PROTOCOL_TYPE = 0; 
  16.  
  17.     private static final int PATTION_LEN = 1; 
  18.     private static final int DTYPE_LEN = 1; 
  19.     private static final int MSGID_LEN = 4; 
  20.  
  21.     private int pattion; 
  22.     private int dtype; 
  23.     private int msgId; 
  24.  
  25.     private String data; 
  26.  
  27.     @Override 
  28.     public int getLength() { 
  29.         return super.getLength() + PATTION_LEN + DTYPE_LEN + MSGID_LEN + data.getBytes().length; 
  30.     } 
  31.  
  32.     @Override 
  33.     public int getProtocolType() { 
  34.         return PROTOCOL_TYPE; 
  35.     } 
  36.  
  37.     public int getPattion() { 
  38.         return pattion; 
  39.     } 
  40.  
  41.     public void setPattion(int pattion) { 
  42.         this.pattion = pattion; 
  43.     } 
  44.  
  45.     public int getDtype() { 
  46.         return dtype; 
  47.     } 
  48.  
  49.     public void setDtype(int dtype) { 
  50.         this.dtype = dtype; 
  51.     } 
  52.  
  53.     public void setMsgId(int msgId) { 
  54.         this.msgId = msgId; 
  55.     } 
  56.  
  57.     public int getMsgId() { 
  58.         return msgId; 
  59.     } 
  60.  
  61.     public String getData() { 
  62.         return data; 
  63.     } 
  64.  
  65.     public void setData(String data) { 
  66.         this.data = data; 
  67.     } 
  68.  
  69.     /** 
  70.      * 拼接发送数据 
  71.      * 
  72.      * @return 
  73.      */ 
  74.     @Override 
  75.     public byte[] genContentData() { 
  76.         byte[] base = super.genContentData(); 
  77.         byte[] pattion = {(byte) this.pattion}; 
  78.         byte[] dtype = {(byte) this.dtype}; 
  79.         byte[] msgid = SocketUtil.int2ByteArrays(this.msgId); 
  80.         byte[] data = this.data.getBytes(); 
  81.  
  82.         ByteArrayOutputStream baos = new ByteArrayOutputStream(getLength()); 
  83.         baos.write(base, 0, base.length);          //协议版本+数据类型+数据长度+消息id 
  84.         baos.write(pattion, 0, PATTION_LEN);       //业务类型 
  85.         baos.write(dtype, 0, DTYPE_LEN);           //业务数据格式 
  86.         baos.write(msgid, 0, MSGID_LEN);           //消息id 
  87.         baos.write(data, 0, data.length);          //业务数据 
  88.         return baos.toByteArray(); 
  89.     } 
  90.  
  91.     /** 
  92.      * 解析接收数据,按顺序解析 
  93.      * 
  94.      * @param data 
  95.      * @return 
  96.      * @throws ProtocolException 
  97.      */ 
  98.     @Override 
  99.     public int parseContentData(byte[] data) throws ProtocolException { 
  100.         int pos = super.parseContentData(data); 
  101.  
  102.         //解析pattion 
  103.         pattion = data[pos] & 0xFF; 
  104.         pos += PATTION_LEN; 
  105.  
  106.         //解析dtype 
  107.         dtype = data[pos] & 0xFF; 
  108.         pos += DTYPE_LEN; 
  109.  
  110.         //解析msgId 
  111.         msgId = SocketUtil.byteArrayToInt(data, pos, MSGID_LEN); 
  112.         pos += MSGID_LEN; 
  113.  
  114.         //解析data 
  115.         try { 
  116.             this.data = new String(data, pos, data.length - pos, "utf-8"); 
  117.         } catch (UnsupportedEncodingException e) { 
  118.             e.printStackTrace(); 
  119.         } 
  120.  
  121.         return pos; 
  122.     } 
  123.  
  124.     @Override 
  125.     public String toString() { 
  126.         return "data: " + data; 
  127.     } 
  128.  

3. DataAckProtocol


  1. import com.shandiangou.sdgprotocol.lib.ProtocolException; 
  2. import com.shandiangou.sdgprotocol.lib.SocketUtil; 
  3.  
  4. import java.io.ByteArrayOutputStream; 
  5. import java.io.UnsupportedEncodingException; 
  6.  
  7. /** 
  8.  * Created by meishan on 16/12/1. 
  9.  */ 
  10. public class DataAckProtocol extends BasicProtocol { 
  11.  
  12.     public static final int PROTOCOL_TYPE = 1; 
  13.  
  14.     private static final int ACKMSGID_LEN = 4; 
  15.  
  16.     private int ackMsgId; 
  17.  
  18.     private String unused; 
  19.  
  20.     @Override 
  21.     public int getLength() { 
  22.         return super.getLength() + ACKMSGID_LEN + unused.getBytes().length; 
  23.     } 
  24.  
  25.     @Override 
  26.     public int getProtocolType() { 
  27.         return PROTOCOL_TYPE; 
  28.     } 
  29.  
  30.     public int getAckMsgId() { 
  31.         return ackMsgId; 
  32.     } 
  33.  
  34.     public void setAckMsgId(int ackMsgId) { 
  35.         this.ackMsgId = ackMsgId; 
  36.     } 
  37.  
  38.     public String getUnused() { 
  39.         return unused; 
  40.     } 
  41.  
  42.     public void setUnused(String unused) { 
  43.         this.unused = unused; 
  44.     } 
  45.  
  46.     /** 
  47.      * 拼接发送数据 
  48.      * 
  49.      * @return 
  50.      */ 
  51.     @Override 
  52.     public byte[] genContentData() { 
  53.         byte[] base = super.genContentData(); 
  54.         byte[] ackMsgId = SocketUtil.int2ByteArrays(this.ackMsgId); 
  55.         byte[] unused = this.unused.getBytes(); 
  56.  
  57.         ByteArrayOutputStream baos = new ByteArrayOutputStream(getLength()); 
  58.         baos.write(base, 0, base.length);              //协议版本+数据类型+数据长度+消息id 
  59.         baos.write(ackMsgId, 0, ACKMSGID_LEN);         //消息id 
  60.         baos.write(unused, 0, unused.length);          //unused 
  61.         return baos.toByteArray(); 
  62.     } 
  63.  
  64.     @Override 
  65.     public int parseContentData(byte[] data) throws ProtocolException { 
  66.         int pos = super.parseContentData(data); 
  67.  
  68.         //解析ackMsgId 
  69.         ackMsgId = SocketUtil.byteArrayToInt(data, pos, ACKMSGID_LEN); 
  70.         pos += ACKMSGID_LEN; 
  71.  
  72.         //解析unused 
  73.         try { 
  74.             unused = new String(data, pos, data.length - pos, "utf-8"); 
  75.         } catch (UnsupportedEncodingException e) { 
  76.             e.printStackTrace(); 
  77.         } 
  78.  
  79.         return pos; 
  80.     } 
  81.  
  82.  

4. PingProtocol


  1. import com.shandiangou.sdgprotocol.lib.ProtocolException; 
  2. import com.shandiangou.sdgprotocol.lib.SocketUtil; 
  3.  
  4. import java.io.ByteArrayOutputStream; 
  5. import java.io.UnsupportedEncodingException; 
  6.  
  7. /** 
  8.  * Created by meishan on 16/12/1. 
  9.  */ 
  10. public class PingProtocol extends BasicProtocol { 
  11.  
  12.     public static final int PROTOCOL_TYPE = 2; 
  13.  
  14.     private static final int PINGID_LEN = 4; 
  15.  
  16.     private int pingId; 
  17.  
  18.     private String unused; 
  19.  
  20.     @Override 
  21.     public int getLength() { 
  22.         return super.getLength() + PINGID_LEN + unused.getBytes().length; 
  23.     } 
  24.  
  25.     @Override 
  26.     public int getProtocolType() { 
  27.         return PROTOCOL_TYPE; 
  28.     } 
  29.  
  30.     public int getPingId() { 
  31.         return pingId; 
  32.     } 
  33.  
  34.     public void setPingId(int pingId) { 
  35.         this.pingId = pingId; 
  36.     } 
  37.  
  38.     public String getUnused() { 
  39.         return unused; 
  40.     } 
  41.  
  42.     public void setUnused(String unused) { 
  43.         this.unused = unused; 
  44.     } 
  45.  
  46.     /** 
  47.      * 拼接发送数据 
  48.      * 
  49.      * @return 
  50.      */ 
  51.     @Override 
  52.     public byte[] genContentData() { 
  53.         byte[] base = super.genContentData(); 
  54.         byte[] pingId = SocketUtil.int2ByteArrays(this.pingId); 
  55.         byte[] unused = this.unused.getBytes(); 
  56.  
  57.         ByteArrayOutputStream baos = new ByteArrayOutputStream(getLength()); 
  58.         baos.write(base, 0, base.length);          //协议版本+数据类型+数据长度+消息id 
  59.         baos.write(pingId, 0, PINGID_LEN);         //消息id 
  60.         baos.write(unused, 0, unused.length);            //unused 
  61.         return baos.toByteArray(); 
  62.     } 
  63.  
  64.     @Override 
  65.     public int parseContentData(byte[] data) throws ProtocolException { 
  66.         int pos = super.parseContentData(data); 
  67.  
  68.         //解析pingId 
  69.         pingId = SocketUtil.byteArrayToInt(data, pos, PINGID_LEN); 
  70.         pos += PINGID_LEN; 
  71.  
  72.         try { 
  73.             unused = new String(data, pos, data.length - pos, "utf-8"); 
  74.         } catch (UnsupportedEncodingException e) { 
  75.             e.printStackTrace(); 
  76.         } 
  77.  
  78.         return pos; 
  79.     } 
  80.  
  81.  

5. PingAckProtocol


  1. import com.shandiangou.sdgprotocol.lib.ProtocolException; 
  2. import com.shandiangou.sdgprotocol.lib.SocketUtil; 
  3.  
  4. import java.io.ByteArrayOutputStream; 
  5. import java.io.UnsupportedEncodingException; 
  6.  
  7. /** 
  8.  * Created by meishan on 16/12/1. 
  9.  */ 
  10. public class PingAckProtocol extends BasicProtocol { 
  11.  
  12.     public static final int PROTOCOL_TYPE = 3; 
  13.  
  14.     private static final int ACKPINGID_LEN = 4; 
  15.  
  16.     private int ackPingId; 
  17.  
  18.     private String unused; 
  19.  
  20.     @Override 
  21.     public int getLength() { 
  22.         return super.getLength() + ACKPINGID_LEN + unused.getBytes().length; 
  23.     } 
  24.  
  25.     @Override 
  26.     public int getProtocolType() { 
  27.         return PROTOCOL_TYPE; 
  28.     } 
  29.  
  30.     public int getAckPingId() { 
  31.         return ackPingId; 
  32.     } 
  33.  
  34.     public void setAckPingId(int ackPingId) { 
  35.         this.ackPingId = ackPingId; 
  36.     } 
  37.  
  38.     public String getUnused() { 
  39.         return unused; 
  40.     } 
  41.  
  42.     public void setUnused(String unused) { 
  43.         this.unused = unused; 
  44.     } 
  45.  
  46.     /** 
  47.      * 拼接发送数据 
  48.      * 
  49.      * @return 
  50.      */ 
  51.     @Override 
  52.     public byte[] genContentData() { 
  53.         byte[] base = super.genContentData(); 
  54.         byte[] ackPingId = SocketUtil.int2ByteArrays(this.ackPingId); 
  55.         byte[] unused = this.unused.getBytes(); 
  56.  
  57.         ByteArrayOutputStream baos = new ByteArrayOutputStream(getLength()); 
  58.         baos.write(base, 0, base.length);                //协议版本+数据类型+数据长度+消息id 
  59.         baos.write(ackPingId, 0, ACKPINGID_LEN);         //消息id 
  60.         baos.write(unused, 0, unused.length);            //unused 
  61.         return baos.toByteArray(); 
  62.     } 
  63.  
  64.     @Override 
  65.     public int parseContentData(byte[] data) throws ProtocolException { 
  66.         int pos = super.parseContentData(data); 
  67.  
  68.         //解析ackPingId 
  69.         ackPingId = SocketUtil.byteArrayToInt(data, pos, ACKPINGID_LEN); 
  70.         pos += ACKPINGID_LEN; 
  71.  
  72.         //解析unused 
  73.         try { 
  74.             unused = new String(data, pos, data.length - pos, "utf-8"); 
  75.         } catch (UnsupportedEncodingException e) { 
  76.             e.printStackTrace(); 
  77.         } 
  78.  
  79.         return pos; 
  80.     } 
  81.  
  82.  

三、任务调度

上述已经给出了四种协议的实现,接下来我们将使用它们来实现app和服务端之间的通信,这里我们把数据的发送、接收和心跳分别用一个线程去实现,具体如下:

1. 客户端


  1. import android.os.Handler; 
  2. import android.os.Looper; 
  3. import android.os.Message; 
  4. import android.util.Log; 
  5.  
  6. import com.shandiangou.sdgprotocol.lib.protocol.BasicProtocol; 
  7. import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol; 
  8. import com.shandiangou.sdgprotocol.lib.protocol.PingProtocol; 
  9.  
  10. import java.io.IOException; 
  11. import java.io.InputStream; 
  12. import java.io.OutputStream; 
  13. import java.net.ConnectException; 
  14. import java.net.Socket; 
  15. import java.util.concurrent.ConcurrentLinkedQueue; 
  16.  
  17. import javax.net.SocketFactory; 
  18.  
  19. /** 
  20.  * 写数据采用死循环,没有数据时wait,有新消息时notify 
  21.  * <p> 
  22.  * Created by meishan on 16/12/1. 
  23.  */ 
  24. public class ClientRequestTask implements Runnable { 
  25.  
  26.     private static final int SUCCESS = 100; 
  27.     private static final int FAILED = -1; 
  28.  
  29.     private boolean isLongConnection = true
  30.     private Handler mHandler; 
  31.     private SendTask mSendTask; 
  32.     private ReciveTask mReciveTask; 
  33.     private HeartBeatTask mHeartBeatTask; 
  34.     private Socket mSocket; 
  35.  
  36.     private boolean isSocketAvailable; 
  37.     private boolean closeSendTask; 
  38.  
  39.     protected volatile ConcurrentLinkedQueue<BasicProtocol> dataQueue = new ConcurrentLinkedQueue<>(); 
  40.  
  41.     public ClientRequestTask(RequestCallBack requestCallBacks) { 
  42.         mHandler = new MyHandler(requestCallBacks); 
  43.     } 
  44.  
  45.     @Override 
  46.     public void run() { 
  47.         try { 
  48.             try { 
  49.                 mSocket = SocketFactory.getDefault().createSocket(Config.ADDRESS, Config.PORT); 
  50. //                mSocket.setSoTimeout(10); 
  51.             } catch (ConnectException e) { 
  52.                 failedMessage(-1, "服务器连接异常,请检查网络"); 
  53.                 return
  54.             } 
  55.  
  56.             isSocketAvailable = true
  57.  
  58.             //开启接收线程 
  59.             mReciveTask = new ReciveTask(); 
  60.             mReciveTask.inputStream = mSocket.getInputStream(); 
  61.             mReciveTask.start(); 
  62.  
  63.             //开启发送线程 
  64.             mSendTask = new SendTask(); 
  65.             mSendTask.outputStream = mSocket.getOutputStream(); 
  66.             mSendTask.start(); 
  67.  
  68.             //开启心跳线程 
  69.             if (isLongConnection) { 
  70.                 mHeartBeatTask = new HeartBeatTask(); 
  71.                 mHeartBeatTask.outputStream = mSocket.getOutputStream(); 
  72.                 mHeartBeatTask.start(); 
  73.             } 
  74.         } catch (IOException e) { 
  75.             failedMessage(-1, "网络发生异常,请稍后重试"); 
  76.             e.printStackTrace(); 
  77.         } 
  78.     } 
  79.  
  80.     public void addRequest(DataProtocol data) { 
  81.         dataQueue.add(data); 
  82.         toNotifyAll(dataQueue);//有新增待发送数据,则唤醒发送线程 
  83.     } 
  84.  
  85.     public synchronized void stop() { 
  86.  
  87.         //关闭接收线程 
  88.         closeReciveTask(); 
  89.  
  90.         //关闭发送线程 
  91.         closeSendTask = true
  92.         toNotifyAll(dataQueue); 
  93.  
  94.         //关闭心跳线程 
  95.         closeHeartBeatTask(); 
  96.  
  97.         //关闭socket 
  98.         closeSocket(); 
  99.  
  100.         //清除数据 
  101.         clearData(); 
  102.  
  103.         failedMessage(-1, "断开连接"); 
  104.     } 
  105.  
  106.     /** 
  107.      * 关闭接收线程 
  108.      */ 
  109.     private void closeReciveTask() { 
  110.         if (mReciveTask != null) { 
  111.             mReciveTask.interrupt(); 
  112.             mReciveTask.isCancle = true
  113.             if (mReciveTask.inputStream != null) { 
  114.                 try { 
  115.                     if (isSocketAvailable && !mSocket.isClosed() && mSocket.isConnected()) { 
  116.                         mSocket.shutdownInput();//解决java.net.SocketException问题,需要先shutdownInput 
  117.                     } 
  118.                 } catch (IOException e) { 
  119.                     e.printStackTrace(); 
  120.                 } 
  121.                 SocketUtil.closeInputStream(mReciveTask.inputStream); 
  122.                 mReciveTask.inputStream = null
  123.             } 
  124.             mReciveTask = null
  125.         } 
  126.     } 
  127.  
  128.     /** 
  129.      * 关闭发送线程 
  130.      */ 
  131.     private void closeSendTask() { 
  132.         if (mSendTask != null) { 
  133.             mSendTask.isCancle = true
  134.             mSendTask.interrupt(); 
  135.             if (mSendTask.outputStream != null) { 
  136.                 synchronized (mSendTask.outputStream) {//防止写数据时停止,写完再停 
  137.                     SocketUtil.closeOutputStream(mSendTask.outputStream); 
  138.                     mSendTask.outputStream = null
  139.                 } 
  140.             } 
  141.             mSendTask = null
  142.         } 
  143.     } 
  144.  
  145.     /** 
  146.      * 关闭心跳线程 
  147.      */ 
  148.     private void closeHeartBeatTask() { 
  149.         if (mHeartBeatTask != null) { 
  150.             mHeartBeatTask.isCancle = true
  151.             if (mHeartBeatTask.outputStream != null) { 
  152.                 SocketUtil.closeOutputStream(mHeartBeatTask.outputStream); 
  153.                 mHeartBeatTask.outputStream = null
  154.             } 
  155.             mHeartBeatTask = null
  156.         } 
  157.     } 
  158.  
  159.     /** 
  160.      * 关闭socket 
  161.      */ 
  162.     private void closeSocket() { 
  163.         if (mSocket != null) { 
  164.             try { 
  165.                 mSocket.close(); 
  166.                 isSocketAvailable = false
  167.             } catch (IOException e) { 
  168.                 e.printStackTrace(); 
  169.             } 
  170.         } 
  171.     } 
  172.  
  173.     /** 
  174.      * 清除数据 
  175.      */ 
  176.     private void clearData() { 
  177.         dataQueue.clear(); 
  178.         isLongConnection = false
  179.     } 
  180.  
  181.     private void toWait(Object o) { 
  182.         synchronized (o) { 
  183.             try { 
  184.                 o.wait(); 
  185.             } catch (InterruptedException e) { 
  186.                 e.printStackTrace(); 
  187.             } 
  188.         } 
  189.     } 
  190.  
  191.     /** 
  192.      * notify()调用后,并不是马上就释放对象锁的,而是在相应的synchronized(){}语句块执行结束,自动释放锁后 
  193.      * 
  194.      * @param o 
  195.      */ 
  196.     protected void toNotifyAll(Object o) { 
  197.         synchronized (o) { 
  198.             o.notifyAll(); 
  199.         } 
  200.     } 
  201.  
  202.     private void failedMessage(int code, String msg) { 
  203.         Message message = mHandler.obtainMessage(FAILED); 
  204.         message.what = FAILED; 
  205.         message.arg1 = code; 
  206.         message.obj = msg; 
  207.         mHandler.sendMessage(message); 
  208.     } 
  209.  
  210.     private void successMessage(BasicProtocol protocol) { 
  211.         Message message = mHandler.obtainMessage(SUCCESS); 
  212.         message.what = SUCCESS; 
  213.         message.obj = protocol; 
  214.         mHandler.sendMessage(message); 
  215.     } 
  216.  
  217.     private boolean isConnected() { 
  218.         if (mSocket.isClosed() || !mSocket.isConnected()) { 
  219.             ClientRequestTask.this.stop(); 
  220.             return false
  221.         } 
  222.         return true
  223.     } 
  224.  
  225.     /** 
  226.      * 服务器返回处理,主线程运行 
  227.      */ 
  228.     public class MyHandler extends Handler { 
  229.  
  230.         private RequestCallBack mRequestCallBack; 
  231.  
  232.         public MyHandler(RequestCallBack callBack) { 
  233.             super(Looper.getMainLooper()); 
  234.             this.mRequestCallBack = callBack; 
  235.         } 
  236.  
  237.         @Override 
  238.         public void handleMessage(Message msg) { 
  239.             super.handleMessage(msg); 
  240.             switch (msg.what) { 
  241.                 case SUCCESS: 
  242.                     mRequestCallBack.onSuccess((BasicProtocol) msg.obj); 
  243.                     break; 
  244.                 case FAILED: 
  245.                     mRequestCallBack.onFailed(msg.arg1, (String) msg.obj); 
  246.                     break; 
  247.                 default
  248.                     break; 
  249.             } 
  250.         } 
  251.     } 
  252.  
  253.     /** 
  254.      * 数据接收线程 
  255.      */ 
  256.     public class ReciveTask extends Thread { 
  257.  
  258.         private boolean isCancle = false
  259.         private InputStream inputStream; 
  260.  
  261.         @Override 
  262.         public void run() { 
  263.             while (!isCancle) { 
  264.                 if (!isConnected()) { 
  265.                     break; 
  266.                 } 
  267.  
  268.                 if (inputStream != null) { 
  269.                     BasicProtocol reciverData = SocketUtil.readFromStream(inputStream); 
  270.                     if (reciverData != null) { 
  271.                         if (reciverData.getProtocolType() == 1 || reciverData.getProtocolType() == 3) { 
  272.                             successMessage(reciverData); 
  273.                         } 
  274.                     } else { 
  275.                         break; 
  276.                     } 
  277.                 } 
  278.             } 
  279.  
  280.             SocketUtil.closeInputStream(inputStream);//循环结束则退出输入流 
  281.         } 
  282.     } 
  283.  
  284.     /** 
  285.      * 数据发送线程 
  286.      * 当没有发送数据时让线程等待 
  287.      */ 
  288.     public class SendTask extends Thread { 
  289.  
  290.         private boolean isCancle = false
  291.         private OutputStream outputStream; 
  292.  
  293.         @Override 
  294.         public void run() { 
  295.             while (!isCancle) { 
  296.                 if (!isConnected()) { 
  297.                     break; 
  298.                 } 
  299.  
  300.                 BasicProtocol dataContent = dataQueue.poll(); 
  301.                 if (dataContent == null) { 
  302.                     toWait(dataQueue);//没有发送数据则等待 
  303.                     if (closeSendTask) { 
  304.                         closeSendTask();//notify()调用后,并不是马上就释放对象锁的,所以在此处中断发送线程 
  305.                     } 
  306.                 } else if (outputStream != null) { 
  307.                     synchronized (outputStream) { 
  308.                         SocketUtil.write2Stream(dataContent, outputStream); 
  309.                     } 
  310.                 } 
  311.             } 
  312.  
  313.             SocketUtil.closeOutputStream(outputStream);//循环结束则退出输出流 
  314.         } 
  315.     } 
  316.  
  317.     /** 
  318.      * 心跳实现,频率5秒 
  319.      * Created by meishan on 16/12/1. 
  320.      */ 
  321.     public class HeartBeatTask extends Thread { 
  322.  
  323.         private static final int REPEATTIME = 5000; 
  324.         private boolean isCancle = false
  325.         private OutputStream outputStream; 
  326.         private int pingId; 
  327.  
  328.         @Override 
  329.         public void run() { 
  330.             pingId = 1; 
  331.             while (!isCancle) { 
  332.                 if (!isConnected()) { 
  333.                     break; 
  334.                 } 
  335.  
  336.                 try { 
  337.                     mSocket.sendUrgentData(0xFF); 
  338.                 } catch (IOException e) { 
  339.                     isSocketAvailable = false
  340.                     ClientRequestTask.this.stop(); 
  341.                     break; 
  342.                 } 
  343.  
  344.                 if (outputStream != null) { 
  345.                     PingProtocol pingProtocol = new PingProtocol(); 
  346.                     pingProtocol.setPingId(pingId); 
  347.                     pingProtocol.setUnused("ping..."); 
  348.                     SocketUtil.write2Stream(pingProtocol, outputStream); 
  349.                     pingId = pingId + 2; 
  350.                 } 
  351.  
  352.                 try { 
  353.                     Thread.sleep(REPEATTIME); 
  354.                 } catch (InterruptedException e) { 
  355.                     e.printStackTrace(); 
  356.                 } 
  357.             } 
  358.  
  359.             SocketUtil.closeOutputStream(outputStream); 
  360.         } 
  361.     } 
  362.  

其中涉及到的RequestCallBack接口如下:


  1. /** 
  2.  * Created by meishan on 16/12/1. 
  3.  */ 
  4. public interface RequestCallBack { 
  5.  
  6.     void onSuccess(BasicProtocol msg); 
  7.  
  8.     void onFailed(int errorCode, String msg); 
  9.  

2. 服务端


  1. import java.io.DataInputStream; 
  2. import java.io.DataOutputStream; 
  3. import java.net.Socket; 
  4. import java.util.Iterator; 
  5. import java.util.concurrent.ConcurrentHashMap; 
  6. import java.util.concurrent.ConcurrentLinkedQueue; 
  7.  
  8. /** 
  9.  * Created by meishan on 16/12/1. 
  10.  */ 
  11. public class ServerResponseTask implements Runnable { 
  12.  
  13.     private ReciveTask reciveTask; 
  14.     private SendTask sendTask; 
  15.     private Socket socket; 
  16.     private ResponseCallback tBack; 
  17.  
  18.     private volatile ConcurrentLinkedQueue<BasicProtocol> dataQueue = new ConcurrentLinkedQueue<>(); 
  19.     private static ConcurrentHashMap<String, Socket> onLineClient = new ConcurrentHashMap<>(); 
  20.  
  21.     private String userIP; 
  22.  
  23.     public String getUserIP() { 
  24.         return userIP; 
  25.     } 
  26.  
  27.     public ServerResponseTask(Socket socket, ResponseCallback tBack) { 
  28.         this.socket = socket; 
  29.         this.tBack = tBack; 
  30.         this.userIP = socket.getInetAddress().getHostAddress(); 
  31.         System.out.println("用户IP地址:" + userIP); 
  32.     } 
  33.  
  34.     @Override 
  35.     public void run() { 
  36.         try { 
  37.             //开启接收线程 
  38.             reciveTask = new ReciveTask(); 
  39.             reciveTask.inputStream = new DataInputStream(socket.getInputStream()); 
  40.             reciveTask.start(); 
  41.  
  42.             //开启发送线程 
  43.             sendTask = new SendTask(); 
  44.             sendTask.outputStream = new DataOutputStream(socket.getOutputStream()); 
  45.             sendTask.start(); 
  46.         } catch (Exception e) { 
  47.             e.printStackTrace(); 
  48.         } 
  49.     } 
  50.  
  51.     public void stop() { 
  52.         if (reciveTask != null) { 
  53.             reciveTask.isCancle = true
  54.             reciveTask.interrupt(); 
  55.             if (reciveTask.inputStream != null) { 
  56.                 SocketUtil.closeInputStream(reciveTask.inputStream); 
  57.                 reciveTask.inputStream = null
  58.             } 
  59.             reciveTask = null
  60.         } 
  61.  
  62.         if (sendTask != null) { 
  63.             sendTask.isCancle = true
  64.             sendTask.interrupt(); 
  65.             if (sendTask.outputStream != null) { 
  66.                 synchronized (sendTask.outputStream) {//防止写数据时停止,写完再停 
  67.                     sendTask.outputStream = null
  68.                 } 
  69.             } 
  70.             sendTask = null
  71.         } 
  72.     } 
  73.  
  74.     public void addMessage(BasicProtocol data) { 
  75.         if (!isConnected()) { 
  76.             return
  77.         } 
  78.  
  79.         dataQueue.offer(data); 
  80.         toNotifyAll(dataQueue);//有新增待发送数据,则唤醒发送线程 
  81.     } 
  82.  
  83.     public Socket getConnectdClient(String clientID) { 
  84.         return onLineClient.get(clientID); 
  85.     } 
  86.  
  87.     /** 
  88.      * 打印已经链接的客户端 
  89.      */ 
  90.     public static void printAllClient() { 
  91.         if (onLineClient == null) { 
  92.             return
  93.         } 
  94.         Iterator<String> inter = onLineClient.keySet().iterator(); 
  95.         while (inter.hasNext()) { 
  96.             System.out.println("client:" + inter.next()); 
  97.         } 
  98.     } 
  99.  
  100.     public void toWaitAll(Object o) { 
  101.         synchronized (o) { 
  102.             try { 
  103.                 o.wait(); 
  104.             } catch (InterruptedException e) { 
  105.                 e.printStackTrace(); 
  106.             } 
  107.         } 
  108.     } 
  109.  
  110.     public void toNotifyAll(Object obj) { 
  111.         synchronized (obj) { 
  112.             obj.notifyAll(); 
  113.         } 
  114.     } 
  115.  
  116.     private boolean isConnected() { 
  117.         if (socket.isClosed() || !socket.isConnected()) { 
  118.             onLineClient.remove(userIP); 
  119.             ServerResponseTask.this.stop(); 
  120.             System.out.println("socket closed..."); 
  121.             return false
  122.         } 
  123.         return true
  124.     } 
  125.  
  126.     public class ReciveTask extends Thread { 
  127.  
  128.         private DataInputStream inputStream; 
  129.         private boolean isCancle; 
  130.  
  131.         @Override 
  132.         public void run() { 
  133.             while (!isCancle) { 
  134.                 if (!isConnected()) { 
  135.                     isCancle = true
  136.                     break; 
  137.                 } 
  138.  
  139.                 BasicProtocol clientData = SocketUtil.readFromStream(inputStream); 
  140.  
  141.                 if (clientData != null) { 
  142.                     if (clientData.getProtocolType() == 0) { 
  143.                         System.out.println("dtype: " + ((DataProtocol) clientData).getDtype() + ", pattion: " + ((DataProtocol) clientData).getPattion() + ", msgId: " + ((DataProtocol) clientData).getMsgId() + ", data: " + ((DataProtocol) clientData).getData()); 
  144.  
  145.                         DataAckProtocol dataAck = new DataAckProtocol(); 
  146.                         dataAck.setUnused("收到消息:" + ((DataProtocol) clientData).getData()); 
  147.                         dataQueue.offer(dataAck); 
  148.                         toNotifyAll(dataQueue); //唤醒发送线程 
  149.  
  150.                         tBack.targetIsOnline(userIP); 
  151.                     } else if (clientData.getProtocolType() == 2) { 
  152.                         System.out.println("pingId: " + ((PingProtocol) clientData).getPingId()); 
  153.  
  154.                         PingAckProtocol pingAck = new PingAckProtocol(); 
  155.                         pingAck.setUnused("收到心跳"); 
  156.                         dataQueue.offer(pingAck); 
  157.                         toNotifyAll(dataQueue); //唤醒发送线程 
  158.  
  159.                         tBack.targetIsOnline(userIP); 
  160.                     } 
  161.                 } else { 
  162.                     System.out.println("client is offline..."); 
  163.                     break; 
  164.                 } 
  165.             } 
  166.  
  167.             SocketUtil.closeInputStream(inputStream); 
  168.         } 
  169.     } 
  170.  
  171.     public class SendTask extends Thread { 
  172.  
  173.         private DataOutputStream outputStream; 
  174.         private boolean isCancle; 
  175.  
  176.         @Override 
  177.         public void run() { 
  178.             while (!isCancle) { 
  179.                 if (!isConnected()) { 
  180.                     isCancle = true
  181.                     break; 
  182.                 } 
  183.  
  184.                 BasicProtocol procotol = dataQueue.poll(); 
  185.                 if (procotol == null) { 
  186.                     toWaitAll(dataQueue); 
  187.                 } else if (outputStream != null) { 
  188.                     synchronized (outputStream) { 
  189.                         SocketUtil.write2Stream(procotol, outputStream); 
  190.                     } 
  191.                 } 
  192.             } 
  193.  
  194.             SocketUtil.closeOutputStream(outputStream); 
  195.         } 
  196.     } 

其中涉及到的ResponseCallback接口如下:


  1. /** 
  2.  * Created by meishan on 16/12/1. 
  3.  */ 
  4. public interface ResponseCallback { 
  5.  
  6.     void targetIsOffline(DataProtocol reciveMsg); 
  7.  
  8.     void targetIsOnline(String clientIp); 
  9.  

上述代码中处理了几种情况下的异常,比如,建立连接后,服务端停止运行,此时客户端的输入流还在阻塞状态,怎么保证客户端不抛出异常,这些处理可以结合SocketUtil类来看。

四、调用封装

1. 客户端

import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol;


  1. import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol; 
  2.  
  3. /** 
  4.  * Created by meishan on 16/12/1. 
  5.  */ 
  6. public class ConnectionClient { 
  7.  
  8.     private boolean isClosed; 
  9.  
  10.     private ClientRequestTask mClientRequestTask; 
  11.  
  12.     public ConnectionClient(RequestCallBack requestCallBack) { 
  13.         mClientRequestTask = new ClientRequestTask(requestCallBack); 
  14.         new Thread(mClientRequestTask).start(); 
  15.     } 
  16.  
  17.     public void addNewRequest(DataProtocol data) { 
  18.         if (mClientRequestTask != null && !isClosed) 
  19.             mClientRequestTask.addRequest(data); 
  20.     } 
  21.  
  22.     public void closeConnect() { 
  23.         isClosed = true
  24.         mClientRequestTask.stop(); 
  25.     } 
  26.  

2. 服务端


  1. import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol; 
  2.  
  3. import java.io.IOException; 
  4. import java.net.ServerSocket; 
  5. import java.net.Socket; 
  6. import java.util.concurrent.ExecutorService; 
  7. import java.util.concurrent.Executors; 
  8.  
  9. /** 
  10.  * Created by meishan on 16/12/1. 
  11.  */ 
  12. public class ConnectionServer { 
  13.  
  14.     private static boolean isStart = true
  15.     private static ServerResponseTask serverResponseTask; 
  16.  
  17.     public ConnectionServer() { 
  18.  
  19.     } 
  20.  
  21.     public static void main(String[] args) { 
  22.  
  23.         ServerSocket serverSocket = null
  24.         ExecutorService executorService = Executors.newCachedThreadPool(); 
  25.         try { 
  26.             serverSocket = new ServerSocket(Config.PORT); 
  27.             while (isStart) { 
  28.                 Socket socket = serverSocket.accept(); 
  29.                 serverResponseTask = new ServerResponseTask(socket, 
  30.                         new ResponseCallback() { 
  31.  
  32.                             @Override 
  33.                             public void targetIsOffline(DataProtocol reciveMsg) {// 对方不在线 
  34.                                 if (reciveMsg != null) { 
  35.                                     System.out.println(reciveMsg.getData()); 
  36.                                 } 
  37.                             } 
  38.  
  39.                             @Override 
  40.                             public void targetIsOnline(String clientIp) { 
  41.                                 System.out.println(clientIp + " is onLine"); 
  42.                                 System.out.println("-----------------------------------------"); 
  43.                             } 
  44.                         }); 
  45.  
  46.                 if (socket.isConnected()) { 
  47.                     executorService.execute(serverResponseTask); 
  48.                 } 
  49.             } 
  50.  
  51.             serverSocket.close(); 
  52.  
  53.         } catch (IOException e) { 
  54.             e.printStackTrace(); 
  55.         } finally { 
  56.             if (serverSocket != null) { 
  57.                 try { 
  58.                     isStart = false
  59.                     serverSocket.close(); 
  60.                     if (serverSocket != null
  61.                         serverResponseTask.stop(); 
  62.                 } catch (IOException e) { 
  63.                     e.printStackTrace(); 
  64.                 } 
  65.             } 
  66.         } 
  67.     } 
  68.  

总结

实现自定义协议的关键在于协议的拼装和解析,上述已给出了关键的代码,如果需要查看完整的代码以及demo,可以下载源码

注意:先运行服务端demo的main函数,再查看本机的ip地址,然后修改客户端(android)代码中Config.java里面的ip地址,当然,要确保android手机和服务端在同一个局域里面,最后再打开客户端。




作者:枚杉
来源:51CTO

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
网络协议之:WebSocket的消息格式
我们知道WebSocket是建立在TCP协议基础上的一种网络协议,用来进行客户端和服务器端的实时通信。非常的好用。最简单的使用WebSocket的办法就是直接使用浏览器的API和服务器端进行通信。 本文将会深入分析WebSocket的消息交互格式,让大家得以明白,websocket到底是怎么工作的。
90 0
spring boot redis 自定义序列化方式(fastJson)
spring boot redis 自定义序列化方式(fastJson)
3899 0
浅谈WebSocket协议、WS协议和WSS协议原理及关系
现如今,一些游戏,网站,APP,支付行业的网络经常会收到DDOS和CC,因此这个事情也变得不足为奇了。但在防护方案中有多种不同的方案。如,今天有几个同事在针对高防CDN问题讨论的时,谈到WebSocket协议,那么一些非技术人员不知道WebSocket协议,WS协议和WSS协议原理及关系,今天就简单分享下。
2659 0
Java编程架构深入解析-RPC 消息协议设计
本节我们开始讲解 RPC 的消息协议设计背后的基本原理,了解 RPC 的协议开发背后有哪些需要考虑的基本点。在通晓原理之后,我们就可以自己设计一套协议来开发属于自己的 RPC 系统。
1285 0
Android与服务器端数据交互(http协议整合struts2+android)
android中有时候我们不需要用到本机的SQLite数据库提供数据,更多的时候是从网络上获取数据,那么Android怎么从服务器端获取数据呢?有很多种,归纳起来有 一:基于Http协议获取数据方法。
659 0
+关注
玄学酱
这个时候,玄酱是不是应该说点什么...
20710
文章
438
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载