美文网首页Android Dev服务器学习技术分享
实现通过UDP自动获取IP并建立TCP连接

实现通过UDP自动获取IP并建立TCP连接

作者: wwzlp | 来源:发表于2017-03-02 14:35 被阅读483次

    由于项目要求TV端与移动(手机/平板)端进行离线通讯的需求,所以我选择了建立TCP连接来实现离线的功能。

    那么问题来了:

    1.再TV端输入IP地址(这个界面也是需要的),但是使用遥控机输入麻烦。

    2.如何使TV端自动获取移动端的IP。

    本来考虑的通过移动端上传IP,TV端再进行更新的方案。但是可能存在没网情况下IP未及时更新的情况,那么离线模式也将不可用,稳定性不高。

    后来想到可以用UDP广播来实现获取移动端IP,TV端通过发送UDP广播,如果移动端在同一网段,那么接收到广播后再把当前的IP通过UDP发送给TV端,拿到了IP,那问题自然也就解决了。

    第一步:建立TV端UDP接收器&发送器

    UDP接收器

    服务端要这边定好自己的端口,客户端通过这个端口发送(同网段),服务端就可以接收到客户端发送的广播了。

       public synchronized void initAndStart(){
           byte[] message =  new byte[100];
           try {
               datagramSocket = new DatagramSocket(TV_SERVER_PORT );
               datagramSocket.setBroadcast(true);
               DatagramPacket datagramPacket = new DatagramPacket(message , message.length);
               while (!isThreadDisable){
                   try {
                       lock.acquire();
                       datagramSocket.receive(datagramPacket);
                       String receiveMsg =new String(datagramPacket.getData()).trim();
                       Log.e("UdpServer", "收到消息 " + receiveMsg);
                       if (receiveMsg.startsWith("server_ip")){
                           String[] serverIp = receiveMsg.split("#");
                           if (serverIp.length == 2 && listener != null){
                               listener.onGetServerIp(serverIp[1]);
                           }
                       }
                       lock.release();
                   } catch (IOException e) {
                       e.printStackTrace();
                   }
               }
           } catch (SocketException e) {
               e.printStackTrace();
           }
       }
    

    UDP发送器

    值得注意的是,在安卓上并不能接收到跨网段的UDP广播(当然在你知道IP的情况下跨网段通信是可以的,这一点和TCP通信没什么区别),所以只能在同网段下实现UDP广播的收发。

    private  synchronized void send(String message) {
            if (TextUtils.isEmpty(message))
                return;
            if (datagramSocket == null || datagramSocket.isClosed()) {
                InetAddress inetAddress = null;
                try {
                    datagramSocket = new DatagramSocket();
                } catch (SocketException e) {
                    e.printStackTrace();
                }
                try {
                    inetAddress = InetAddress.getByName(broadcastIp);
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                }
                int msg_length = message.length();
                byte[] messageByte = message.getBytes();
                datagramPacket = new DatagramPacket(messageByte, msg_length, inetAddress, UDP_QUEUE_SERVER_PORT);
            }
            try {
                Log.e("UdpSender", "将要发送消息 " + message);
                lock.acquire();
                datagramSocket.send(datagramPacket);
                lock.release();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    

    注:我这边因为循环调用上面发送器的原因,没有把datagramSocket直接close掉。等到发送广播结束时要注意调用datagramSocket.close()来释放资源。

    第二步:建立移动端UDP接收器&发送器

    在移动端的收发和上面的大同小异,无非是发送和接收的内容不同罢了,这里我就不再详细贴代码了,参考第一步中的代码可以自行根据自己的业务来构造UDP的收发器。

    第三步:在移动端上建立TCP服务端

    这边我使用的xsocket的库,里面封装了一些接口比较方便。
    首先实现xsocket封装的handler接口:

    public class SocketServerHandler implements IDataHandler, IConnectHandler, IDisconnectHandler, IDestroyable, ISocketSender {
    private final String TAG = "SocketServerHandler";
    
        public SocketServerHandler(){
        }
    
        @Override
        public boolean onData(INonBlockingConnection iNonBlockingConnection) throws IOException, BufferUnderflowException, ClosedChannelException, MaxReadSizeExceededException {
            return true;
        }
    
        @Override
        public boolean onConnect(INonBlockingConnection iNonBlockingConnection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
            Log.e(TAG, "消息服务器,客户端连接上来了.onConnect" + iNonBlockingConnection);
        }
    
        @Override
        public boolean onDisconnect(INonBlockingConnection iNonBlockingConnection) throws IOException {
            Log.e(TAG, "消息服务器,客户端断开连接.onDisconnect");
            return true;
        }
    
        @Override
        public void destroy() {
        }
    
        @Override
        public synchronized void send(String message) {
          
        }
    }
    

    那么接下来我们要在客户端连接上时,去保存客户端的连接,以便之后发消息给客户端:

        private Lock lock = new ReentrantLock();
        private Set<INonBlockingConnection> connections = null;
    
    
      @Override
        public boolean onConnect(INonBlockingConnection iNonBlockingConnection) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
            Log.e(TAG, "消息服务器,客户端连接上来了.onConnect" + iNonBlockingConnection);
            lock.lock();
            try {
                connections.add(iNonBlockingConnection);
            } finally {
                lock.unlock();
            }
            return true;
        }
    
    @Override
        public boolean onDisconnect(INonBlockingConnection iNonBlockingConnection) throws IOException {
            Log.e(TAG, "消息服务器,客户端断开连接.onDisconnect");
            lock.lock();
            try {
                connections.remove(iNonBlockingConnection);
            } finally {
                lock.unlock();
            }
            return true;
        }
    

    好了,接下来就是实现如何去发送消息了。我选择使用BlockingQueue来实现消息的存取,一种实现了阻塞接口的队列,然后启动发送消息的线程去循环取这个队列就可以了,不说了,上代码:

    private BlockingQueue<String> messageQueue = null;//消息队列
    private Thread writeThread;//发消息线程
    private Timer timer = null;//用来发送心跳消息的轮询任务
    
    public SocketServerHandler() {
            messageQueue = new LinkedBlockingDeque<>(100);
            connections = new HashSet<>();
            writeThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        String msg;
                        try {
                            msg = messageQueue.take();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            return;
                        }
    
                        List<INonBlockingConnection> tConnections = new ArrayList<>();
                        lock.lock();
                        try {
                            if (connections.isEmpty()) {
                                continue;
                            }
                            tConnections.addAll(connections);
                        } finally {
                            lock.unlock();
                        }
    
                        StringBuilder sb = new StringBuilder();
                        sb.append(msg).append(IQueueMessage.SPLIT);
                        String tMsg = sb.toString();
    
                        for (INonBlockingConnection connection : tConnections) {
                            try {
                                if (connection.isOpen()) {
                                    Log.e(TAG, "客户端信息:" + connection);
                                    connection.write(tMsg);
                                }
                            } catch (Throwable t) {
                                t.printStackTrace();
                            }
                        }
    
                    }
                }
            });
            writeThread.start();
    
            timer = new Timer(TAG, true);
            long splitTime = 20 * 1000L;
            timer.schedule(new TimerTask() {
                @Override
                public void run() {
                    send(IQueueMessage.MESSAGE_HEART);
                }
            }, splitTime, splitTime);
        }
    

    在发送消息时,只需要向队列里面塞消息就可以了:

    @Override
        public synchronized void send(String message) {
            if (message != null) {
                Log.e(TAG, "发送消息:" + message);
                messageQueue.add(message);
            }
        }
    

    最后,启动socket服务,把刚才定义的handler放进去就可以了:

     private IServer iServer;
    
     SocketServerMonitor monitor = new SocketServerMonitor(new SocketServerHandler());
     handler.post(monitor);
    
    class SocketServerMonitor implements Runnable {
            SocketServerHandler serverHandler;
    
            public SocketServerMonitor(SocketServerHandler serverHandler) {
                this.serverHandler = serverHandler;
            }
    
            @Override
            public void run() {
                if (iServer != null && iServer.isOpen()) {
                    handler.postDelayed(this, 10 * 1000);
                    return;
                }
    
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            iServer = new Server(IQueueMessage.SOCKET_MESSAGE_PORT, serverHandler);
                            iServer.start();
                            iServer.addListener(new IServerListener() {
                                @Override
                                public void onInit() {
                                    Log.e("SocketServerMonitor", "消息服务器初始化...");
                                }
    
                                @Override
                                public void onDestroy() throws IOException {
                                    Log.e("SocketServerMonitor", "消息服务器onDestroy...");
                                }
                            });
                            Log.e("SocketServerMonitor", "启动/重启消息服务器成功");
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                handler.postDelayed(this, 10 * 1000);
            }
        }
    

    第四步:在TV端上建立TCP客户端,并实现socket连接监听器

    客户端使用INonBlockingConnection这个接口来实现socket的连接,同样也需要自定义handler来处理消息。基本和服务端一样:

    public class SocketClientHandler implements IDataHandler, IDisconnectHandler, IConnectHandler , IDestroyable , ISocketSender{
        final private static String TAG = "SocketClientHandler";
        /**
         * <code>是否连接上了</code>.
         */
        private boolean isConnected;
    
        private INonBlockingConnection serverConnection;
        private BlockingQueue<String> messageQueue = null;
        private Thread writeThread;
    
        public SocketClientHandler() {
            messageQueue = new LinkedBlockingQueue<>();
            writeThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        String msg;
                        try {
                            msg = messageQueue.take();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            return;
                        }
    
                        if (serverConnection == null)
                            continue;
    
                        StringBuilder sb = new StringBuilder();
                        sb.append(msg).append(IQueueMessage.SPLIT);
                        String sendMsg = sb.toString();
    
                        try {
                            if (serverConnection.isOpen()){
                                Log.e(TAG , "服务端信息:"+ serverConnection);
                                serverConnection.write(sendMsg);
                            }
                        }catch (Throwable t) {
                            t.printStackTrace();
                        }
                    }
                }
            });
            writeThread.start();
        }
    
        @Override
        public boolean onData(INonBlockingConnection arg) throws IOException, BufferUnderflowException, ClosedChannelException, MaxReadSizeExceededException {
            String msg = arg.readStringByDelimiter(IQueueMessage.SPLIT).trim();
            if (StringUtils.isBlank(msg))
                return true;
         
            if (IQueueMessage.MESSAGE_HEART.equals(msg)){
                Log.e(TAG, "心跳消息:" + msg);
                return true;
            }
            Log.e(TAG, "收到消息体:" + msg);
    
            return true;
        }
    
       
    
        @Override
        public boolean onConnect(INonBlockingConnection arg) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
            Log.e(TAG , "onSocketConnected");
            isConnected = true;
            serverConnection  = arg;
            return true;
        }
    
        @Override
        public boolean onDisconnect(INonBlockingConnection arg) throws IOException {
            Log.e(TAG , "onSocketDisConnected -- " + arg);
            isConnected = false;
            serverConnection = null;
            return true;
        }
    
        public boolean isConnected() {
            return isConnected;
        }
    
        public void reset() {
            isConnected = false;
        }
    
        @Override
        public synchronized  void send(String message) {
            if (message != null) {
                Log.e(TAG , "发送消息:" + message);
                messageQueue.add(message);
            }
        }
    
        @Override
        public void destroy() {
            Log.e(TAG, "客户端销毁");
            writeThread.interrupt();
        }
    }
    

    然后自己维护一个监听器,维护实现客户端的socket连接。

    public class SocketConnectionMonitor extends Handler {
        final private static String TAG = "SocketConnectionMonitor";
    
        private String serverIp;
        private int port;
        private SocketClientHandler socketClientHandler;
        private INonBlockingConnection nonBlockingConnection;
        private Application application;
        private boolean connecting;
    
        public SocketConnectionMonitor(String serverIp, int port,Application application) {
            this.serverIp = serverIp;
            this.port = port;
            this.application = application;
            socketClientHandler = new SocketClientHandler();
        }
    
        @Override
        public void handleMessage(Message msg) {
            super.handleMessage(msg);
            switch (msg.what) {
                case IMessageKey.KEY_MONITOR_SOCKET:
                    monitorConnection();
                    break;
            
            }
        }
    
        public synchronized void monitorConnection() {
            if (!isConnected() && !connecting) {
                if (!NetWorkUtils.isNetworkActive(application))
                    return;
    
                connecting = true;
                try {
                    connect();
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    connecting = false;
                }
            }
    
        }
    
        private void connect() {
            if (StringUtils.isBlank(serverIp) && !NetWorkUtils.isNetworkActive(application))
                return;
    
            if (nonBlockingConnection != null) {
                try {
                    Log.e(TAG, "initConnection  nonBlockingConnection.close();");
                    nonBlockingConnection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            try {
                Log.e(TAG, "initConnection  nonBlockingConnection.reset();");
                socketClientHandler.reset();
                Log.e(TAG, "initConnection" + serverIp + ":" + port);
                String[] ipStr = serverIp.split("\\.");
                byte[] ipBuf = new byte[4];
                for (int i = 0; i < 4; i++) {
                    ipBuf[i] = (byte) (Integer.parseInt(ipStr[i]) & 0xff);
                }
    
                InetAddress inetAddress = InetAddress.getByAddress(ipBuf);
                Log.e(TAG, "initConnection  connect");
                nonBlockingConnection = new NonBlockingConnection(inetAddress, port, socketClientHandler, 3000);
                Log.e(TAG, "initConnection  nonBlockingConnection.setIdleTimeoutMillis");
                nonBlockingConnection.setIdleTimeoutMillis(32000);
    
                Log.e(TAG, "initConnection - success" + serverIp + ":" + port);
            } catch (Throwable t) {
                t.printStackTrace();
                Log.e(TAG, "initConnection - failed " + serverIp + ":" + port);
            }
        }
    
        public boolean isConnected() {
            return socketClientHandler.isConnected();
        }
    
        public void setServerIp(String serverIp) {
            this.serverIp = serverIp;
            socketClientHandler.reset();
        }
    }
    

    最后启动建立线程去监测socket的连接状态,TV端与移动端的离线模式基本框架就已经完成了。

    if (!mMessageInited) {
                final String serverIp = mPlatform.getServerIp();
                mSocketMonitor = new SocketConnectionMonitor(serverIp, IQueueMessage.SOCKET_MESSAGE_PORT, this);
                mMonitorThread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        while (true) {
                            try {
                                mSocketMonitor.monitorConnection();
                                Thread.sleep(1000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                });
                mMonitorThread.start();
                mMessageInited = true;
            }
    

    当然,如果两个设备不在同一网段的话,TV端这边是无法自动获取移动端的IP的,所以还是留了一个输入IP的界面可以让用户自己输入IP。因为遥控器上只有上下左右确认键,输入的时候还是比较繁琐的,实现界面是要注意焦点的控制/(ㄒoㄒ)/~~。

    相关文章

      网友评论

        本文标题:实现通过UDP自动获取IP并建立TCP连接

        本文链接:https://www.haomeiwen.com/subject/mklcgttx.html