原文地址：http://blog.csdn.net/u010818425/article/details/53448817
在阅读本文前需要对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
import android.util.Log;
import com.shandiangou.sdgprotocol.lib.Config;
import com.shandiangou.sdgprotocol.lib.ProtocolException;
import com.shandiangou.sdgprotocol.lib.SocketUtil;
import java.io.ByteArrayOutputStream;
/**
* Created by meishan on 16/12/1.
* <p>
* 协议类型: 0表示数据，1表示数据Ack，2表示ping，3表示pingAck
*/
public abstract class BasicProtocol {
public static final int LENGTH_LEN =
4;
protected static final int VER_LEN =
1;
protected static final int TYPE_LEN =
1;
private int reserved =
0;
private int version = Config.VERSION;
/**
* 获取整条数据长度
* 单位：字节（byte）
*
* @return
*/
protected int getLength() {
return LENGTH_LEN + VER_LEN + TYPE_LEN;
}
public int getReserved() {
return reserved;
}
public void setReserved(
int reserved) {
this.reserved = reserved;
}
public int getVersion() {
return version;
}
public void setVersion(
int version) {
this.version = version;
}
/**
* 获取协议类型，由子类实现
*
* @return
*/
public abstract int getProtocolType();
/**
* 由预留值和版本号计算完整版本号的byte[]值
*
* @return
*/
private int getVer(
byte r,
byte v,
int vLen) {
int num =
0;
int rLen =
8 - vLen;
for (
int i =
0; i < rLen; i++) {
num += (((r >> (rLen -
1 - i)) &
0x1) << (
7 - i));
}
return num + v;
}
/**
* 拼接发送数据，此处拼接了协议版本、协议类型和数据长度，具体内容子类中再拼接
* 按顺序拼接
*
* @return
*/
public byte[]
genContentData() {
byte[] length = SocketUtil.int2ByteArrays(getLength());
byte reserved = (
byte) getReserved();
byte version = (
byte) getVersion();
byte[] ver = {(
byte) getVer(reserved, version,
5)};
byte[] type = {(
byte) getProtocolType()};
ByteArrayOutputStream baos =
new ByteArrayOutputStream(LENGTH_LEN + VER_LEN + TYPE_LEN);
baos.write(length,
0, LENGTH_LEN);
baos.write(ver,
0, VER_LEN);
baos.write(type,
0, TYPE_LEN);
return baos.toByteArray();
}
/**
* 解析出整条数据长度
*
* @param data
* @return
*/
protected int parseLength(
byte[] data) {
return SocketUtil.byteArrayToInt(data,
0, LENGTH_LEN);
}
/**
* 解析出预留位
*
* @param data
* @return
*/
protected int parseReserved(
byte[] data) {
byte r = data[LENGTH_LEN];
return (r >>
5) &
0xFF;
}
/**
* 解析出版本号
*
* @param data
* @return
*/
protected int parseVersion(
byte[] data) {
byte v = data[LENGTH_LEN];
return ((v <<
3) &
0xFF) >>
3;
}
/**
* 解析出协议类型
*
* @param data
* @return
*/
public static int parseType(
byte[] data) {
byte t = data[LENGTH_LEN + VER_LEN];
return t &
0xFF;
}
/**
* 解析接收数据，此处解析了协议版本、协议类型和数据长度，具体内容子类中再解析
*
* @param data
* @return
* @throws ProtocolException 协议版本不一致，抛出异常
*/
public int parseContentData(
byte[] data)
throws ProtocolException {
int reserved = parseReserved(data);
int version = parseVersion(data);
int protocolType = parseType(data);
if (version != getVersion()) {
throw new ProtocolException(
"input version is error: " + version);
}
return LENGTH_LEN + VER_LEN + TYPE_LEN;
}
@Override
public String
toString() {
return "Version: " + getVersion() +
", Type: " + getProtocolType();
}
}
上述涉及到的Config类和SocketUtil类如下：
/**
* Created by meishan on 16/12/2.
*/
public class Config {
public static final int VERSION =
1;
public static final String ADDRESS =
"10.17.64.237";
public static final int PORT =
9013;
}
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
/**
* Created by meishan on 16/12/1.
*/
public class SocketUtil {
private static Map<Integer, String> msgImp =
new HashMap<>();
static {
msgImp.put(DataProtocol.PROTOCOL_TYPE,
"com.shandiangou.sdgprotocol.lib.protocol.DataProtocol");
msgImp.put(DataAckProtocol.PROTOCOL_TYPE,
"com.shandiangou.sdgprotocol.lib.protocol.DataAckProtocol");
msgImp.put(PingProtocol.PROTOCOL_TYPE,
"com.shandiangou.sdgprotocol.lib.protocol.PingProtocol");
msgImp.put(PingAckProtocol.PROTOCOL_TYPE,
"com.shandiangou.sdgprotocol.lib.protocol.PingAckProtocol");
}
/**
* 解析数据内容
*
* @param data
* @return
*/
public static BasicProtocol
parseContentMsg(
byte[] data) {
int protocolType = BasicProtocol.parseType(data);
String className = msgImp.get(protocolType);
BasicProtocol basicProtocol;
try {
basicProtocol = (BasicProtocol) Class.forName(className).newInstance();
basicProtocol.parseContentData(data);
}
catch (Exception e) {
basicProtocol =
null;
e.printStackTrace();
}
return basicProtocol;
}
/**
* 读数据
*
* @param inputStream
* @return
* @throws SocketExceptions
*/
public static BasicProtocol
readFromStream(InputStream inputStream) {
BasicProtocol protocol;
BufferedInputStream bis;
byte[] header =
new byte[BasicProtocol.LENGTH_LEN];
try {
bis =
new BufferedInputStream(inputStream);
int temp;
int len =
0;
while (len < header.length) {
temp = bis.read(header, len, header.length - len);
if (temp >
0) {
len += temp;
}
else if (temp == -
1) {
bis.close();
return null;
}
}
len =
0;
int length = byteArrayToInt(header);
byte[] content =
new byte[length];
while (len < length) {
temp = bis.read(content, len, length - len);
if (temp >
0) {
len += temp;
}
}
protocol = parseContentMsg(content);
}
catch (IOException e) {
e.printStackTrace();
return null;
}
return protocol;
}
/**
* 写数据
*
* @param protocol
* @param outputStream
*/
public static void write2Stream(BasicProtocol protocol, OutputStream outputStream) {
BufferedOutputStream bufferedOutputStream =
new BufferedOutputStream(outputStream);
byte[] buffData = protocol.genContentData();
byte[] header = int2ByteArrays(buffData.length);
try {
bufferedOutputStream.write(header);
bufferedOutputStream.write(buffData);
bufferedOutputStream.flush();
}
catch (IOException e) {
e.printStackTrace();
}
}
/**
* 关闭输入流
*
* @param is
*/
public static void closeInputStream(InputStream is) {
try {
if (is !=
null) {
is.close();
}
}
catch (IOException e) {
e.printStackTrace();
}
}
/**
* 关闭输出流
*
* @param os
*/
public static void closeOutputStream(OutputStream os) {
try {
if (os !=
null) {
os.close();
}
}
catch (IOException e) {
e.printStackTrace();
}
}
public static byte[]
int2ByteArrays(
int i) {
byte[] result =
new byte[
4];
result[
0] = (
byte) ((i >>
24) &
0xFF);
result[
1] = (
byte) ((i >>
16) &
0xFF);
result[
2] = (
byte) ((i >>
8) &
0xFF);
result[
3] = (
byte) (i &
0xFF);
return result;
}
public static int byteArrayToInt(
byte[] b) {
int intValue =
0;
for (
int i =
0; i < b.length; i++) {
intValue += (b[i] &
0xFF) << (
8 * (
3 - i));
}
return intValue;
}
public static int byteArrayToInt(
byte[] b,
int byteOffset,
int byteCount) {
int intValue =
0;
for (
int i = byteOffset; i < (byteOffset + byteCount); i++) {
intValue += (b[i] &
0xFF) << (
8 * (
3 - (i - byteOffset)));
}
return intValue;
}
public static int bytes2Int(
byte[] b,
int byteOffset) {
ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.SIZE / Byte.SIZE);
byteBuffer.put(b, byteOffset,
4);
byteBuffer.flip();
return byteBuffer.getInt();
}
}
接下来我们实现具体的协议。
2. DataProtocol
import android.util.Log;
import com.shandiangou.sdgprotocol.lib.ProtocolException;
import com.shandiangou.sdgprotocol.lib.SocketUtil;
import java.io.ByteArrayOutputStream;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
/**
* Created by meishan on 16/12/1.
*/
public class DataProtocol extends BasicProtocol implements Serializable {
public static final int PROTOCOL_TYPE =
0;
private static final int PATTION_LEN =
1;
private static final int DTYPE_LEN =
1;
private static final int MSGID_LEN =
4;
private int pattion;
private int dtype;
private int msgId;
private String data;
@Override
public int getLength() {
return super.getLength() + PATTION_LEN + DTYPE_LEN + MSGID_LEN + data.getBytes().length;
}
@Override
public int getProtocolType() {
return PROTOCOL_TYPE;
}
public int getPattion() {
return pattion;
}
public void setPattion(
int pattion) {
this.pattion = pattion;
}
public int getDtype() {
return dtype;
}
public void setDtype(
int dtype) {
this.dtype = dtype;
}
public void setMsgId(
int msgId) {
this.msgId = msgId;
}
public int getMsgId() {
return msgId;
}
public String
getData() {
return data;
}
public void setData(String data) {
this.data = data;
}
/**
* 拼接发送数据
*
* @return
*/
@Override
public byte[]
genContentData() {
byte[] base =
super.genContentData();
byte[] pattion = {(
byte)
this.pattion};
byte[] dtype = {(
byte)
this.dtype};
byte[] msgid = SocketUtil.int2ByteArrays(
this.msgId);
byte[] data =
this.data.getBytes();
ByteArrayOutputStream baos =
new ByteArrayOutputStream(getLength());
baos.write(base,
0, base.length);
baos.write(pattion,
0, PATTION_LEN);
baos.write(dtype,
0, DTYPE_LEN);
baos.write(msgid,
0, MSGID_LEN);
baos.write(data,
0, data.length);
return baos.toByteArray();
}
/**
* 解析接收数据，按顺序解析
*
* @param data
* @return
* @throws ProtocolException
*/
@Override
public int parseContentData(
byte[] data)
throws ProtocolException {
int pos =
super.parseContentData(data);
pattion = data[pos] &
0xFF;
pos += PATTION_LEN;
dtype = data[pos] &
0xFF;
pos += DTYPE_LEN;
msgId = SocketUtil.byteArrayToInt(data, pos, MSGID_LEN);
pos += MSGID_LEN;
try {
this.data =
new String(data, pos, data.length - pos,
"utf-8");
}
catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return pos;
}
@Override
public String
toString() {
return "data: " + data;
}
}
3. DataAckProtocol
import com.shandiangou.sdgprotocol.lib.ProtocolException;
import com.shandiangou.sdgprotocol.lib.SocketUtil;
import java.io.ByteArrayOutputStream;
import java.io.UnsupportedEncodingException;
/**
* Created by meishan on 16/12/1.
*/
public class DataAckProtocol extends BasicProtocol {
public static final int PROTOCOL_TYPE =
1;
private static final int ACKMSGID_LEN =
4;
private int ackMsgId;
private String unused;
@Override
public int getLength() {
return super.getLength() + ACKMSGID_LEN + unused.getBytes().length;
}
@Override
public int getProtocolType() {
return PROTOCOL_TYPE;
}
public int getAckMsgId() {
return ackMsgId;
}
public void setAckMsgId(
int ackMsgId) {
this.ackMsgId = ackMsgId;
}
public String
getUnused() {
return unused;
}
public void setUnused(String unused) {
this.unused = unused;
}
/**
* 拼接发送数据
*
* @return
*/
@Override
public byte[]
genContentData() {
byte[] base =
super.genContentData();
byte[] ackMsgId = SocketUtil.int2ByteArrays(
this.ackMsgId);
byte[] unused =
this.unused.getBytes();
ByteArrayOutputStream baos =
new ByteArrayOutputStream(getLength());
baos.write(base,
0, base.length);
baos.write(ackMsgId,
0, ACKMSGID_LEN);
baos.write(unused,
0, unused.length);
return baos.toByteArray();
}
@Override
public int parseContentData(
byte[] data)
throws ProtocolException {
int pos =
super.parseContentData(data);
ackMsgId = SocketUtil.byteArrayToInt(data, pos, ACKMSGID_LEN);
pos += ACKMSGID_LEN;
try {
unused =
new String(data, pos, data.length - pos,
"utf-8");
}
catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return pos;
}
}
4. PingProtocol
import com.shandiangou.sdgprotocol.lib.ProtocolException;
import com.shandiangou.sdgprotocol.lib.SocketUtil;
import java.io.ByteArrayOutputStream;
import java.io.UnsupportedEncodingException;
/**
* Created by meishan on 16/12/1.
*/
public class PingProtocol extends BasicProtocol {
public static final int PROTOCOL_TYPE =
2;
private static final int PINGID_LEN =
4;
private int pingId;
private String unused;
@Override
public int getLength() {
return super.getLength() + PINGID_LEN + unused.getBytes().length;
}
@Override
public int getProtocolType() {
return PROTOCOL_TYPE;
}
public int getPingId() {
return pingId;
}
public void setPingId(
int pingId) {
this.pingId = pingId;
}
public String
getUnused() {
return unused;
}
public void setUnused(String unused) {
this.unused = unused;
}
/**
* 拼接发送数据
*
* @return
*/
@Override
public byte[]
genContentData() {
byte[] base =
super.genContentData();
byte[] pingId = SocketUtil.int2ByteArrays(
this.pingId);
byte[] unused =
this.unused.getBytes();
ByteArrayOutputStream baos =
new ByteArrayOutputStream(getLength());
baos.write(base,
0, base.length);
baos.write(pingId,
0, PINGID_LEN);
baos.write(unused,
0, unused.length);
return baos.toByteArray();
}
@Override
public int parseContentData(
byte[] data)
throws ProtocolException {
int pos =
super.parseContentData(data);
pingId = SocketUtil.byteArrayToInt(data, pos, PINGID_LEN);
pos += PINGID_LEN;
try {
unused =
new String(data, pos, data.length - pos,
"utf-8");
}
catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return pos;
}
}
5. PingAckProtocol
import com.shandiangou.sdgprotocol.lib.ProtocolException;
import com.shandiangou.sdgprotocol.lib.SocketUtil;
import java.io.ByteArrayOutputStream;
import java.io.UnsupportedEncodingException;
/**
* Created by meishan on 16/12/1.
*/
public class PingAckProtocol extends BasicProtocol {
public static final int PROTOCOL_TYPE =
3;
private static final int ACKPINGID_LEN =
4;
private int ackPingId;
private String unused;
@Override
public int getLength() {
return super.getLength() + ACKPINGID_LEN + unused.getBytes().length;
}
@Override
public int getProtocolType() {
return PROTOCOL_TYPE;
}
public int getAckPingId() {
return ackPingId;
}
public void setAckPingId(
int ackPingId) {
this.ackPingId = ackPingId;
}
public String
getUnused() {
return unused;
}
public void setUnused(String unused) {
this.unused = unused;
}
/**
* 拼接发送数据
*
* @return
*/
@Override
public byte[]
genContentData() {
byte[] base =
super.genContentData();
byte[] ackPingId = SocketUtil.int2ByteArrays(
this.ackPingId);
byte[] unused =
this.unused.getBytes();
ByteArrayOutputStream baos =
new ByteArrayOutputStream(getLength());
baos.write(base,
0, base.length);
baos.write(ackPingId,
0, ACKPINGID_LEN);
baos.write(unused,
0, unused.length);
return baos.toByteArray();
}
@Override
public int parseContentData(
byte[] data)
throws ProtocolException {
int pos =
super.parseContentData(data);
ackPingId = SocketUtil.byteArrayToInt(data, pos, ACKPINGID_LEN);
pos += ACKPINGID_LEN;
try {
unused =
new String(data, pos, data.length - pos,
"utf-8");
}
catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return pos;
}
}
三、任务调度
上述已经给出了四种协议的实现，接下来我们将使用它们来实现app和服务端之间的通信，这里我们把数据的发送、接收和心跳分别用一个线程去实现，具体如下：
1. 客户端
import android.os.Handler;
import android.os.Looper;
import android.os.Message;
import android.util.Log;
import com.shandiangou.sdgprotocol.lib.protocol.BasicProtocol;
import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol;
import com.shandiangou.sdgprotocol.lib.protocol.PingProtocol;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.Socket;
import java.util.concurrent.ConcurrentLinkedQueue;
import javax.net.SocketFactory;
/**
* 写数据采用死循环，没有数据时wait，有新消息时notify
* <p>
* Created by meishan on 16/12/1.
*/
public class ClientRequestTask implements Runnable {
private static final int SUCCESS =
100;
private static final int FAILED = -
1;
private boolean isLongConnection =
true;
private Handler mHandler;
private SendTask mSendTask;
private ReciveTask mReciveTask;
private HeartBeatTask mHeartBeatTask;
private Socket mSocket;
private boolean isSocketAvailable;
private boolean closeSendTask;
protected volatile ConcurrentLinkedQueue<BasicProtocol> dataQueue =
new ConcurrentLinkedQueue<>();
public ClientRequestTask(RequestCallBack requestCallBacks) {
mHandler =
new MyHandler(requestCallBacks);
}
@Override
public void run() {
try {
try {
mSocket = SocketFactory.getDefault().createSocket(Config.ADDRESS, Config.PORT);
}
catch (ConnectException e) {
failedMessage(-
1,
"服务器连接异常，请检查网络");
return;
}
isSocketAvailable =
true;
mReciveTask =
new ReciveTask();
mReciveTask.inputStream = mSocket.getInputStream();
mReciveTask.start();
mSendTask =
new SendTask();
mSendTask.outputStream = mSocket.getOutputStream();
mSendTask.start();
if (isLongConnection) {
mHeartBeatTask =
new HeartBeatTask();
mHeartBeatTask.outputStream = mSocket.getOutputStream();
mHeartBeatTask.start();
}
}
catch (IOException e) {
failedMessage(-
1,
"网络发生异常，请稍后重试");
e.printStackTrace();
}
}
public void addRequest(DataProtocol data) {
dataQueue.add(data);
toNotifyAll(dataQueue);
}
public synchronized void stop() {
closeReciveTask();
closeSendTask =
true;
toNotifyAll(dataQueue);
closeHeartBeatTask();
closeSocket();
clearData();
failedMessage(-
1,
"断开连接");
}
/**
* 关闭接收线程
*/
private void closeReciveTask() {
if (mReciveTask !=
null) {
mReciveTask.interrupt();
mReciveTask.isCancle =
true;
if (mReciveTask.inputStream !=
null) {
try {
if (isSocketAvailable && !mSocket.isClosed() && mSocket.isConnected()) {
mSocket.shutdownInput();
}
}
catch (IOException e) {
e.printStackTrace();
}
SocketUtil.closeInputStream(mReciveTask.inputStream);
mReciveTask.inputStream =
null;
}
mReciveTask =
null;
}
}
/**
* 关闭发送线程
*/
private void closeSendTask() {
if (mSendTask !=
null) {
mSendTask.isCancle =
true;
mSendTask.interrupt();
if (mSendTask.outputStream !=
null) {
synchronized (mSendTask.outputStream) {
SocketUtil.closeOutputStream(mSendTask.outputStream);
mSendTask.outputStream =
null;
}
}
mSendTask =
null;
}
}
/**
* 关闭心跳线程
*/
private void closeHeartBeatTask() {
if (mHeartBeatTask !=
null) {
mHeartBeatTask.isCancle =
true;
if (mHeartBeatTask.outputStream !=
null) {
SocketUtil.closeOutputStream(mHeartBeatTask.outputStream);
mHeartBeatTask.outputStream =
null;
}
mHeartBeatTask =
null;
}
}
/**
* 关闭socket
*/
private void closeSocket() {
if (mSocket !=
null) {
try {
mSocket.close();
isSocketAvailable =
false;
}
catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 清除数据
*/
private void clearData() {
dataQueue.clear();
isLongConnection =
false;
}
private void toWait(Object o) {
synchronized (o) {
try {
o.wait();
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* notify()调用后，并不是马上就释放对象锁的，而是在相应的synchronized(){}语句块执行结束，自动释放锁后
*
* @param o
*/
protected void toNotifyAll(Object o) {
synchronized (o) {
o.notifyAll();
}
}
private void failedMessage(
int code, String msg) {
Message message = mHandler.obtainMessage(FAILED);
message.what = FAILED;
message.arg1 = code;
message.obj = msg;
mHandler.sendMessage(message);
}
private void successMessage(BasicProtocol protocol) {
Message message = mHandler.obtainMessage(SUCCESS);
message.what = SUCCESS;
message.obj = protocol;
mHandler.sendMessage(message);
}
private boolean isConnected() {
if (mSocket.isClosed() || !mSocket.isConnected()) {
ClientRequestTask.
this.stop();
return false;
}
return true;
}
/**
* 服务器返回处理，主线程运行
*/
public class MyHandler extends Handler {
private RequestCallBack mRequestCallBack;
public MyHandler(RequestCallBack callBack) {
super(Looper.getMainLooper());
this.mRequestCallBack = callBack;
}
@Override
public void handleMessage(Message msg) {
super.handleMessage(msg);
switch (msg.what) {
case SUCCESS:
mRequestCallBack.onSuccess((BasicProtocol) msg.obj);
break;
case FAILED:
mRequestCallBack.onFailed(msg.arg1, (String) msg.obj);
break;
default:
break;
}
}
}
/**
* 数据接收线程
*/
public class ReciveTask extends Thread {
private boolean isCancle =
false;
private InputStream inputStream;
@Override
public void run() {
while (!isCancle) {
if (!isConnected()) {
break;
}
if (inputStream !=
null) {
BasicProtocol reciverData = SocketUtil.readFromStream(inputStream);
if (reciverData !=
null) {
if (reciverData.getProtocolType() ==
1 || reciverData.getProtocolType() ==
3) {
successMessage(reciverData);
}
}
else {
break;
}
}
}
SocketUtil.closeInputStream(inputStream);
}
}
/**
* 数据发送线程
* 当没有发送数据时让线程等待
*/
public class SendTask extends Thread {
private boolean isCancle =
false;
private OutputStream outputStream;
@Override
public void run() {
while (!isCancle) {
if (!isConnected()) {
break;
}
BasicProtocol dataContent = dataQueue.poll();
if (dataContent ==
null) {
toWait(dataQueue);
if (closeSendTask) {
closeSendTask();
}
}
else if (outputStream !=
null) {
synchronized (outputStream) {
SocketUtil.write2Stream(dataContent, outputStream);
}
}
}
SocketUtil.closeOutputStream(outputStream);
}
}
/**
* 心跳实现，频率5秒
* Created by meishan on 16/12/1.
*/
public class HeartBeatTask extends Thread {
private static final int REPEATTIME =
5000;
private boolean isCancle =
false;
private OutputStream outputStream;
private int pingId;
@Override
public void run() {
pingId =
1;
while (!isCancle) {
if (!isConnected()) {
break;
}
try {
mSocket.sendUrgentData(
0xFF);
}
catch (IOException e) {
isSocketAvailable =
false;
ClientRequestTask.
this.stop();
break;
}
if (outputStream !=
null) {
PingProtocol pingProtocol =
new PingProtocol();
pingProtocol.setPingId(pingId);
pingProtocol.setUnused(
"ping...");
SocketUtil.write2Stream(pingProtocol, outputStream);
pingId = pingId +
2;
}
try {
Thread.sleep(REPEATTIME);
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
SocketUtil.closeOutputStream(outputStream);
}
}
}
其中涉及到的RequestCallBack接口如下：
/**
* Created by meishan on 16/12/1.
*/
public interface RequestCallBack {
void onSuccess(BasicProtocol msg);
void onFailed(
int errorCode, String msg);
}
2. 服务端
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.net.Socket;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* Created by meishan on 16/12/1.
*/
public class ServerResponseTask implements Runnable {
private ReciveTask reciveTask;
private SendTask sendTask;
private Socket socket;
private ResponseCallback tBack;
private volatile ConcurrentLinkedQueue<BasicProtocol> dataQueue =
new ConcurrentLinkedQueue<>();
private static ConcurrentHashMap<String, Socket> onLineClient =
new ConcurrentHashMap<>();
private String userIP;
public String
getUserIP() {
return userIP;
}
public ServerResponseTask(Socket socket, ResponseCallback tBack) {
this.socket = socket;
this.tBack = tBack;
this.userIP = socket.getInetAddress().getHostAddress();
System.out.println(
"用户IP地址：" + userIP);
}
@Override
public void run() {
try {
reciveTask =
new ReciveTask();
reciveTask.inputStream =
new DataInputStream(socket.getInputStream());
reciveTask.start();
sendTask =
new SendTask();
sendTask.outputStream =
new DataOutputStream(socket.getOutputStream());
sendTask.start();
}
catch (Exception e) {
e.printStackTrace();
}
}
public void stop() {
if (reciveTask !=
null) {
reciveTask.isCancle =
true;
reciveTask.interrupt();
if (reciveTask.inputStream !=
null) {
SocketUtil.closeInputStream(reciveTask.inputStream);
reciveTask.inputStream =
null;
}
reciveTask =
null;
}
if (sendTask !=
null) {
sendTask.isCancle =
true;
sendTask.interrupt();
if (sendTask.outputStream !=
null) {
synchronized (sendTask.outputStream) {
sendTask.outputStream =
null;
}
}
sendTask =
null;
}
}
public void addMessage(BasicProtocol data) {
if (!isConnected()) {
return;
}
dataQueue.offer(data);
toNotifyAll(dataQueue);
}
public Socket
getConnectdClient(String clientID) {
return onLineClient.get(clientID);
}
/**
* 打印已经链接的客户端
*/
public static void printAllClient() {
if (onLineClient ==
null) {
return;
}
Iterator<String> inter = onLineClient.keySet().iterator();
while (inter.hasNext()) {
System.out.println(
"client:" + inter.next());
}
}
public void toWaitAll(Object o) {
synchronized (o) {
try {
o.wait();
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public void toNotifyAll(Object obj) {
synchronized (obj) {
obj.notifyAll();
}
}
private boolean isConnected() {
if (socket.isClosed() || !socket.isConnected()) {
onLineClient.remove(userIP);
ServerResponseTask.
this.stop();
System.out.println(
"socket closed...");
return false;
}
return true;
}
public class ReciveTask extends Thread {
private DataInputStream inputStream;
private boolean isCancle;
@Override
public void run() {
while (!isCancle) {
if (!isConnected()) {
isCancle =
true;
break;
}
BasicProtocol clientData = SocketUtil.readFromStream(inputStream);
if (clientData !=
null) {
if (clientData.getProtocolType() ==
0) {
System.out.println(
"dtype: " + ((DataProtocol) clientData).getDtype() +
", pattion: " + ((DataProtocol) clientData).getPattion() +
", msgId: " + ((DataProtocol) clientData).getMsgId() +
", data: " + ((DataProtocol) clientData).getData());
DataAckProtocol dataAck =
new DataAckProtocol();
dataAck.setUnused(
"收到消息：" + ((DataProtocol) clientData).getData());
dataQueue.offer(dataAck);
toNotifyAll(dataQueue);
tBack.targetIsOnline(userIP);
}
else if (clientData.getProtocolType() ==
2) {
System.out.println(
"pingId: " + ((PingProtocol) clientData).getPingId());
PingAckProtocol pingAck =
new PingAckProtocol();
pingAck.setUnused(
"收到心跳");
dataQueue.offer(pingAck);
toNotifyAll(dataQueue);
tBack.targetIsOnline(userIP);
}
}
else {
System.out.println(
"client is offline...");
break;
}
}
SocketUtil.closeInputStream(inputStream);
}
}
public class SendTask extends Thread {
private DataOutputStream outputStream;
private boolean isCancle;
@Override
public void run() {
while (!isCancle) {
if (!isConnected()) {
isCancle =
true;
break;
}
BasicProtocol procotol = dataQueue.poll();
if (procotol ==
null) {
toWaitAll(dataQueue);
}
else if (outputStream !=
null) {
synchronized (outputStream) {
SocketUtil.write2Stream(procotol, outputStream);
}
}
}
SocketUtil.closeOutputStream(outputStream);
}
}
其中涉及到的ResponseCallback接口如下：
/**
* Created by meishan on 16/12/1.
*/
public interface ResponseCallback {
void targetIsOffline(DataProtocol reciveMsg);
void targetIsOnline(String clientIp);
}
上述代码中处理了几种情况下的异常，比如，建立连接后，服务端停止运行，此时客户端的输入流还在阻塞状态，怎么保证客户端不抛出异常，这些处理可以结合SocketUtil类来看。
四、调用封装
1. 客户端
import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol;
/**
* Created by meishan on 16/12/1.
*/
public class ConnectionClient {
private boolean isClosed;
private ClientRequestTask mClientRequestTask;
public ConnectionClient(RequestCallBack requestCallBack) {
mClientRequestTask =
new ClientRequestTask(requestCallBack);
new Thread(mClientRequestTask).start();
}
public void addNewRequest(DataProtocol data) {
if (mClientRequestTask !=
null && !isClosed)
mClientRequestTask.addRequest(data);
}
public void closeConnect() {
isClosed =
true;
mClientRequestTask.stop();
}
}
2. 服务端
import com.shandiangou.sdgprotocol.lib.protocol.DataProtocol;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* Created by meishan on 16/12/1.
*/
public class ConnectionServer {
private static boolean isStart =
true;
private static ServerResponseTask serverResponseTask;
public ConnectionServer() {
}
public static void main(String[] args) {
ServerSocket serverSocket =
null;
ExecutorService executorService = Executors.newCachedThreadPool();
try {
serverSocket =
new ServerSocket(Config.PORT);
while (isStart) {
Socket socket = serverSocket.accept();
serverResponseTask =
new ServerResponseTask(socket,
new ResponseCallback() {
@Override
public void targetIsOffline(DataProtocol reciveMsg) {
if (reciveMsg !=
null) {
System.out.println(reciveMsg.getData());
}
}
@Override
public void targetIsOnline(String clientIp) {
System.out.println(clientIp +
" is onLine");
System.out.println(
"-----------------------------------------");
}
});
if (socket.isConnected()) {
executorService.execute(serverResponseTask);
}
}
serverSocket.close();
}
catch (IOException e) {
e.printStackTrace();
}
finally {
if (serverSocket !=
null) {
try {
isStart =
false;
serverSocket.close();
if (serverSocket !=
null)
serverResponseTask.stop();
}
catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
参考
http://blog.csdn.net/mr_oorange/article/details/52353626 http://vtrtbb.iteye.com/blog/849336 http://blog.csdn.net/a8128230/article/details/46676149 http://www.tuicool.com/articles/6j2ARrB http://weixiaolu.iteye.com/blog/1479656
