【Android实战】Socket消息通信

这篇博客的内容是基于http://my.oschina.net/fengcunhan/blog/178155进行改造的。所以须要先看完这篇博客，然后再来看以下的内容。

1、须要完毕的功能是直播间的socket相关通信

包含例如以下功能：心跳包检測、创建房间、进入房间、退出房间、发言、显示发言、关闭房间、用户信息推送、用户进出房间信息推送、通道验证、道具使用消息推送、账号异地登录消息推送、用户道具信息获取推送

原来某些信息的获取是通过http每3秒一次的轮询进行获得（实时性不好且影响某些功能的体验（尤其是道具撒花的功能））

2、tcp服务接口协议

Tcp消息的结束符“\r\n”,消息字段切割以“|”竖线。client提交数据以SS开头，服务端返回数据以RC开头。

向server端提交数据是类似get提交的方式 uid=2&content=hello

3、以发送心跳包的格式为例

备注：心跳包是防止tcp连接中断，每1分钟向server发送一次数据请求

向server发送：

列	内容	是否必须	说明
信息头	SS	是
操作符	tcp.heartbeat	是	server要运行的操作
内容长度	10	是	最后数据列的长度，为以后做数据校验预留
数据	uid:int:用户id

server返回：

列	内容	说明
信息头	RC
操作符	tcp.heartbeat	server要运行的操作
时间	2015-11-04 12:00:00
数据	status:int:结果状态 msg:string:server返回消息	Json数据格式

实例：>> SS|tcp.heartbeat|7|uid=123\r\n

<< RC|tcp.heartbeat|2015-11-04 13:30:00|{

"status": 0,

"msg": "ok",

“key”:””

}\r\n

每次数据提交中，某些參数必选，某些參数可选

某些数据提交中。key可选。key是为了确认会话（没有收到响应的key。就进行消息重发），server端会返回key数据

key:string:每条消息的唯一标示，仅仅要不短时间内出现反复而且字符长度不要太长都能够，md5值，时间戳+随机数等等，为了保证server端应答的唯一性

4、遇到的几个问题：

1）、socket返回的数据流的问题（数据流返回比較密集的情况下easy出现）

1)返回一条正常的数据流信息

2)返回多条信息的拼接流

3)非完整多条信息的拼接流（有可能返回一条半，下半条下次返回。也有可能后半条丢失）

处理方法：每次收到新的消息都要和之前的消息进行拼接。然后通过\r\n进行切割，正常的消息会分发，非正常消息（不完整）会在解析的过程中被丢弃

2）、由于\r\n，会出现log打印不出来的问题（linux下 \r\n 问题）

处理方法：为了方便调试，能够通过trim()或者replace \r\n的方式

5、其他相关

1）心跳检測 socket断网重连

2）消息验证失败，重发一次（待优化）

3）相关错误码处理

4）基于不同消息类型进行数据处理展示

5）为什么要用AIDL？

6）调研的过程中（认为AndroidAsync应该也不错）

简介：AndroidAsync 是一个基于nio的异步socket ，http（clientserver端），websocket，socket.io库。AndroidAsync 是一个底层的网络协议库，假设你想要一个easy使用，高级的，http请求库，请使用Ion（它是基于AndroidAsync 的）。正常来说开发人员更倾向于使用
Ion。

假设你须要一个未被封装的Android的raw Socket。 HTTP client/server, WebSocket, and Socket.IO， AndroidAsync 正适合你。

6、相关代码

/**

 * 因为移动设备的网络的复杂性，常常会出现网络断开，假设没有心跳包的检測，

 * client仅仅会在须要发送数据的时候才知道自己已经断线，会延误，甚至丢失server发送过来的数据。

* 所以须要提供心跳检測

 */

public class SocketService extends Service {

	private static final String TAG = "SocketService";

	private static final long HEART_BEAT_RATE = 30 * 1000;//眼下心跳检測频率为30s

	private static final long RE_SEND_MSG = 5 * 1000;//重发消息间隔

	public static final String HOST = "t.ing.baofeng.com";

	public static final int PORT = 9099;

	public static final String MESSAGE_ACTION="com.storm.durian.message_ACTION";

	public static final String HEART_BEAT_ACTION="com.storm.durian.heart_beat_ACTION";

	public static final String RE_CONNECTION = "com.storm.durian.re_connection";//socket断了之后又一次连接及做之后的操作

	public Map<String,String> map = new ConcurrentHashMap<String,String>();

	private ReadThread mReadThread;

	private LocalBroadcastManager mLocalBroadcastManager;

	private WeakReference<Socket> mSocket;

	private String socketMsgStr = "";

	// For heart Beat

	private Handler mHandler = new Handler();

	private Runnable heartBeatRunnable = new Runnable() {

		@Override

		public void run() {

			if (System.currentTimeMillis() - sendTime >= HEART_BEAT_RATE) {

				Map<String, String> m = new HashMap<String, String>();

				String key = String.valueOf(System.currentTimeMillis());

				m.put("key",key);

				boolean isSuccess = sendMsg(key,AsyncSocketUtils.spliceSendMessage(getApplicationContext(), m, UrlContainer.HEART_BEAT));//就发送一个\r\n过去 假设发送失败，就又一次初始化一个socket

				if (!isSuccess) {

					reConnectSocket();

				}

				LogHelper.e(TAG, "send a heart beat "+isSuccess);

			}

			mHandler.postDelayed(this, HEART_BEAT_RATE);

		}

	};

	/**

	 * 重连socket

	 */

	private void reConnectSocket() {

		mHandler.removeCallbacks(heartBeatRunnable);

		mHandler.removeCallbacks(reSendMsgRunnable);

		mReadThread.release();

		releaseLastSocket(mSocket);

		new InitSocketThread().start();

		mHandler.postDelayed(new Runnable() {

            @Override

            public void run() {

                Intent intent = new Intent(RE_CONNECTION);

                mLocalBroadcastManager.sendBroadcast(intent);

            }

        }, 2000);

	}

	private Runnable reSendMsgRunnable = new Runnable() {

		@Override

		public void run() {

			if(map != null &&map.size()>0){

				for (String key:map.keySet()){

					sendMsg(key,map.get(key));

					//重发一次

					map.remove(key);

				}

			}

			mHandler.postDelayed(this, RE_SEND_MSG);

		}

	};

	private long sendTime = 0L;

	private ISocketService.Stub iSocketService = new ISocketService.Stub() {

		@Override

		public boolean sendMessage(String key,String message) throws RemoteException {

			return sendMsg(key,message);

		}

		@Override

		public boolean keyAuthentication(String key) throws RemoteException {

			return keyAuthen(key);

		}

	};

	@Override

	public IBinder onBind(Intent arg0) {

		return iSocketService;

	}

	@Override

	public void onCreate() {

		super.onCreate();

		new InitSocketThread().start();

		mLocalBroadcastManager=LocalBroadcastManager.getInstance(this);

	}

	@Override

	public void onDestroy() {

		// TODO Auto-generated method stub

		super.onDestroy();

		if(map !=null) {

			map.clear();

			map = null;

		}

		LogHelper.e(TAG, "onDestroy");

		mHandler.removeCallbacks(heartBeatRunnable);

		mHandler.removeCallbacks(reSendMsgRunnable);

		if(mReadThread != null){

			mReadThread.release();

		}

		if(mSocket != null){

			releaseLastSocket(mSocket);

		}

	}

	public boolean sendMsg(String key,String msg) {

		if (null == mSocket || null == mSocket.get()) {

			return false;

		}

		Socket soc = mSocket.get();

		try {

			if (!soc.isClosed() && !soc.isOutputShutdown()) {

				OutputStream os = soc.getOutputStream();

				String message = msg + "\r\n";

				os.write(message.getBytes());

				os.flush();

				sendTime = System.currentTimeMillis();//每次发送成数据，就改一下最后成功发送的时间，节省心跳间隔时间

				LogHelper.e(TAG,"SuS sendMsg------> "+msg);

			} else {

				return false;

			}

		} catch (IOException e) {

			e.printStackTrace();

			return false;

		}

		map.put(key,msg);

		return true;

	}

	public boolean keyAuthen(String key){

		if (map.containsKey(key)) {

			map.remove(key);

			return true;

		}

		return false;

	}

	private void initSocket() {//初始化Socket

		try {

			Socket so = new Socket(HOST, PORT);

			mSocket = new WeakReference<Socket>(so);

			mReadThread = new ReadThread(so);

			mReadThread.start();

			mHandler.postDelayed(heartBeatRunnable, HEART_BEAT_RATE);//初始化成功后，就准备发送心跳包

			mHandler.postDelayed(reSendMsgRunnable,RE_SEND_MSG);

		} catch (UnknownHostException e) {

			e.printStackTrace();

		} catch (IOException e) {

			e.printStackTrace();

		}

	}

	private void releaseLastSocket(WeakReference<Socket> mSocket) {

		try {

			if (null != mSocket) {

				Socket sk = mSocket.get();

				if (sk != null && !sk.isClosed()) {

					sk.close();

				}

				sk = null;

				mSocket = null;

			}

		} catch (IOException e) {

			e.printStackTrace();

		}

	}

	class InitSocketThread extends Thread {

		@Override

		public void run() {

			super.run();

			initSocket();

		}

	}

	// Thread to read content from Socket

	class ReadThread extends Thread {

		private WeakReference<Socket> mWeakSocket;

		private boolean isStart = true;

		public ReadThread(Socket socket) {

			mWeakSocket = new WeakReference<Socket>(socket);

		}

		public void release() {

			isStart = false;

			if(mWeakSocket != null){

				releaseLastSocket(mWeakSocket);

			}

		}

        @Override

		public void run() {

			super.run();

			Socket socket = mWeakSocket.get();

			if (null != socket) {

				try {

					InputStream is = socket.getInputStream();

					byte[] buffer = new byte[1024 * 4];

					int length = 0;

					while (!socket.isClosed() && !socket.isInputShutdown()

							&& isStart && ((length = is.read(buffer)) != -1)) {

						if (length > 0) {

							String message = new String(Arrays.copyOf(buffer,

									length));

							//trim能够去掉末尾的\r\n

							//每条消息以\r\n分开

							//直接通过LogHelper打印message无显示，能够加trim或者replace \r\n

							LogHelper.e(TAG, "原始message------> " + message.replace("\r\n","||||||"));

							socketMsgStr = socketMsgStr + message;

							if (!socketMsgStr.contains("\r\n")) {

								LogHelper.e(TAG, "socketMsgStr without分隔符");

								continue;

							}

							//返回的消息流有可能是拼接在一起的2条消息，所以须要处理

							String[] a = socketMsgStr.split("\r\n");

							for(int i = 0;i<a.length;i++){

								Intent intent = new Intent(MESSAGE_ACTION);

								if(a[i].charAt(a[i].length()-1)!='}'){

									continue;

								}

								intent.putExtra("message", a[i]);

								mLocalBroadcastManager.sendBroadcast(intent);

								int rIndex = socketMsgStr.indexOf("\r\n");

								if (rIndex + 2 == socketMsgStr.length()) {

									socketMsgStr = "";

								} else {

									socketMsgStr = socketMsgStr.substring(socketMsgStr.indexOf("\r\n") + 2);

								}

							}

							LogHelper.e(TAG, "截取之后的message------> " + socketMsgStr.trim());

						}

					}

				} catch (Exception e) {//还会遇到SocketException

				}

			}

		}

	}

}

附：假设哪位朋友有更好的Android Socket编程框架推荐或者优化方面的资源。欢迎共享。大家一起进步！

巴特西

【Android实战】Socket消息通信

最新文章

热门文章