即时通讯里android客户端心跳机制的分析和实现

81 阅读 0 评论 54 点赞

我是靠谱客的博主专注奇异果，最近开发中收集的这篇文章主要介绍即时通讯里android客户端心跳机制的分析和实现，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

最近项目中用到了即时通讯（IM）的模块，因此也就对即时通讯的长连接方式进行了研究，有了一点点自己的心得，所以写下了这篇博客。本篇博客是作者参考了他人的优秀博客和自己的一点点小感悟而写成的，如有理解不到位的地方，还望指正。

现在大多数的移动端应用都有实时得到消息的能力，简单来说，有发送消息的主动权和接收消息的被动权。
什么是主动权呢？就是客户端主动向服务器请求数据的过程（例如使用http拉取数据），这个过程叫做poling(轮询).
什么是被动权呢？就是服务器在于客户端保持长连接的情况下主动向客户端去推送数据（对应于现在流行的极光推送,或者是使用XMPP来长连接），这个过程叫做push.
心跳包就是在客户端和服务器间定时通知对方自己状态的一个自己定义的命令字，按照一定的时间间隔发送，类似于心跳，所以叫做心跳包。用来判断对方（设备，进程或其它网元）是否正常运行，采用定时发送简单的通讯包，如果在指定时间段内未收到对方响应，则判断对方已经离线。
之所以会有心跳机制，原因有两点：

什么是主动权呢？就是客户端主动向服务器请求数据的过程（例如使用http拉取数据），这个过程叫做poling(轮询).
什么是被动权呢？就是服务器在于客户端保持长连接的情况下主动向客户端去推送数据（对应于现在流行的极光推送,或者是使用XMPP来长连接），这个过程叫做push.
心跳包就是在客户端和服务器间定时通知对方自己状态的一个自己定义的命令字，按照一定的时间间隔发送，类似于心跳，所以叫做心跳包。用来判断对方（设备，进程或其它网元）是否正常运行，采用定时发送简单的通讯包，如果在指定时间段内未收到对方响应，则判断对方已经离线。

之所以会有心跳机制，原因有两点：

第一是我们现在仍然是IPV4的方式，而IPV4的数量有限，我们手机里的Ip地址实际上是移动无线网络运营商分给我们的内网IP地址，而运营商本身就需要维护一个外网IP和端口到内网IP和端口的映射关系，所以，为了确保众多内网的手机可以跟网络服务器通讯，大部分移动无线网络运营商都在链路一段时间没有数据通讯时，会淘汰网路地址转换表（NAT表）中的对应项，造成链路中断。

第二是在android里，在内存不足的情况下，此时系统进程，系统服务也是会被杀死的，因此，就无法保证客户端和服务器进行一个长时间有效的长连接。
所以，为了让服务器获知客户端是在线还是掉线的情况，往往就需要客户端定时发送简单的信息告诉服务器，我仍然和你保持着连接。使用TCP的，用send发，使用UDP的，用sendto发，服务器收到后，就知道当前客户端还处于“活着”的状态，否则，如果隔一定时间未收到这样的包，则服务器认为客户端已经断开，进行相应的断开逻辑处理。
先来说下server端的实现思路：如果应用是基于tcp的，可以简单地通过SO_KEEPALIVE实现心跳。TCP在设置的KeepAlive定时器到达时向对端发一个检测TCP segment，如果没收到ACK或RST，尝试几次后，就认为客户端已经掉线。不过，这里有个缺点，这里是server主动发出检测包，对服务器的性能有影响。
现在我们来说说Client端自己实现心跳的情况：
整个设计思路如下，在客户端中开启一个service,service中使用socket和服务器保持连接，同时加入心跳机制用来维持这个长连接。当service收到服务端发来的回复之后（这个回复可能来自心跳包，也可能来自正常的数据交互），以广播的形式发送给activity,告诉activity要进行相应的逻辑处理（一般是ui的变化）

以下是客户端实现心跳包的具体伪代码：

public class BackService extends Service {
    public static final String MESSAGE_ACTION="org.feng.message_ACTION";
    public static final String HEART_BEAT_ACTION="org.feng.heart_beat_ACTION";
    private LocalBroadcastManager mLocalBroadcastManager;
    private static final long HEART_BEAT_RATE = 3 * 1000;//定义的心跳时间为3秒
    private WeakReference<Socket> mSocket;//最好使用弱引用
    //发送心跳
    private Handler mHandler = new Handler();
    private Runnable heartBeatRunnnable = new Runnable() {
        @Override
        public void run() {
            if (System.currentTimeMillis() - sendTime >= HEART_BEAT_RATE) {
                boolean isSuccess = sendMsg("");//就发送一个rn过去 如果发送失败，就重新初始化一个socket
                if (!isSuccess) {
                    mHandler.removeCallbacks(heartBeatRunnnable);
                    mReadThread.release();//释放读线程
                    releaseLastSocket(mSocket);//释放失效的socket
                    new InitSocketThread().start();//重连
                }
                mHandler.postDelayed(this, HEART_BEAT_RATE);
            }
        }
    };

    class InitSocketThread extends Thread {
        @Override
        public void run() {
            super.run();
            initSocket();//初始化连接
        }
    }

    class ReadThread extends Thread {
        private WeakReference<Socket> mWeakSocket;//采用弱引用是为了方便gc的回收
        private boolean isStart = true;

        public ReadThread(Socket socket) {
            mWeakSocket = new WeakReference<Socket>(socket);
        }

        public void release() {
            isStart = false;
            releaseLastSocket(mWeakSocket);
        }

        @Override
        public void run() {
            Socket socket = mWeakSocket.get();
            if (null != socket) {
                //为了方便阅读，简略接收数据的代码
                // 实际上调用socket.getInputStream()就可以获取到服务器发回的数据了
                String message = 服务器发回来的数据
                //收到服务器过来的消息，就通过Broadcast发送出去
                if (message.equals("ok")) {
                    //处理心跳回复
                    Intent intent = new Intent(HEART_BEAT_ACTION);
                    mLocalBroadcastManager.sendBroadcast(intent);
                } else {
                    //其他消息回复
                    Intent intent = new Intent(MESSAGE_ACTION);
                    intent.putExtra("message", message);
                    mLocalBroadcastManager.sendBroadcast(intent);
                }
            }
        }

    };
    private void initSocket() {//初始化Socket
        Socket so = new Socket(HOST, PORT);
        mSocket = new WeakReference<Socket>(so);
        mReadThread = new ReadThread(so);
        mReadThread.start();
        mHandler.postDelayed(heartBeatRunnnable, HEART_BEAT_RATE);//初始化成功后，就准备发送心跳包

    }

    private IMyAidlInterface.Stub iMyAidlInterface = new IMyAidlInterface.Stub() {
        @Override
        public void basicTypes(int anInt, long aLong, boolean aBoolean, float aFloat, double aDouble, String aString) throws RemoteException {

        }

        @Override
        public String sendMessage(String message) throws RemoteException {
            return sendMsg(message);
        }
    };
    @Override
    public void onCreate() {
        super.onCreate();
        new InitSocketThread().start();
        mLocalBroadcastManager=LocalBroadcastManager.getInstance(this);

    }
    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return iMyAidlInterface;
    }
    //考虑到了ipc的情况
    public String sendMsg(String msg) {
        if (null == mSocket || null == mSocket.get()) {
            return "error";
        }
        Socket soc = mSocket.get();
        try {
            if (!soc.isClosed() && !soc.isOutputShutdown()) {
                OutputStream os = soc.getOutputStream();
                String message = msg + "rn";
                os.write(message.getBytes());
                os.flush();
                sendTime = System.currentTimeMillis();//每次发送成数据，就改一下最后成功发送的时间，节省心跳间隔时间
            } else {
                return "error";
            }
        } catch (IOException e) {
            e.printStackTrace();
            return "error";
        }
        return "success";
    }

}

activity中的代码如下：

public class MainActivity extends AppCompatActivity {
    private IMyAidlInterface iMyAidlInterface;
    private Intent  mServiceIntent;
    private MessageBackReceiver mReciver=new MessageBackReceiver();
    private LocalBroadcastManager mLocalBroadcastManager;
    private IntentFilter mIntentFilter;
    private ServiceConnection connection=new ServiceConnection() {
        @Override
        public void onServiceConnected(ComponentName name, IBinder service) {
            //这里稍微解释一下，在与服务器连接时会拿到服务端的IBinder对象，即服务端的BinderProxy
            //调用客户端的asInterface方法来将该IBinder对象封装成本地Proxy(代理，用来执行服务器的远程函数)
            iMyAidlInterface=IMyAidlInterface.Stub.asInterface(service);
        }

        @Override
        public void onServiceDisconnected(ComponentName name) {
            iMyAidlInterface=null;
        }
    };
    class MessageBackReceiver extends BroadcastReceiver{


        @Override
        public void onReceive(Context context, Intent intent) {
            String action=intent.getAction();
            if (action.equals(BackService.HEART_BEAT_ACTION)) {
                //收到了服务器的心跳回复
            } else {
                //收到了服务器的其他回复
            }
        }
    }
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        mServiceIntent = new Intent(this, BackService.class);
        bindService(mServiceIntent, connection, BIND_AUTO_CREATE);
        mIntentFilter = new IntentFilter();
        mIntentFilter.addAction(BackService.HEART_BEAT_ACTION);
        mIntentFilter.addAction(BackService.MESSAGE_ACTION);
        mLocalBroadcastManager = LocalBroadcastManager.getInstance(this);
        mLocalBroadcastManager.registerReceiver(mReciver, mIntentFilter);//接收service发送来的广播
    }

}

上述代码主要参考了博客：https://my.oschina.net/fengcunhan/blog/178155