实现通过UDP自动获取IP并建立TCP连接
2017-03-02 本文已影响483人
wwzlp
由于项目要求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ㄒ)/~~。
