java nio socket长连接_nio实现Socket长连接和心跳
前段時(shí)間用bio方式，也就是傳統(tǒng)io實(shí)現(xiàn)了socket的長(zhǎng)連接和心跳，總覺著服務(wù)端開啟多線程管理socket連接的方式過(guò)于消耗資源，數(shù)據(jù)并發(fā)的情況下可能會(huì)影響到性能，因此就嘗試使用nio改進(jìn)原來(lái)的代碼。
然而改進(jìn)的過(guò)程卻不像我起初設(shè)想的那般容易，可以說(shuō)一波三折，原因主要是nio讀寫都是字節(jié)流，LZ一開始依然通過(guò)ObjectOutputStream.writeObject直接向Socket服務(wù)端發(fā)送數(shù)據(jù)，然而問(wèn)題出現(xiàn)了，每次從ByteBuffer解析出來(lái)字節(jié)流都不一樣，LZ使出渾身解數(shù)，一個(gè)字節(jié)一個(gè)字節(jié)的讀取啊，問(wèn)題沒有了，可是由于是長(zhǎng)連接，數(shù)據(jù)怎么解析啊，查資料，找大神，最后一個(gè)網(wǎng)友說(shuō)有可能是粘包和分包的問(wèn)題，一時(shí)暈菜，LZ網(wǎng)絡(luò)可是渣渣啊，行吧，惡補(bǔ)一番，想了解的童鞋可以看看這個(gè)。http://blog.csdn.net/sunmenggmail/article/details/38952131
實(shí)現(xiàn)原理就像很多協(xié)議那樣，自定義一套傳輸協(xié)議，比如消息長(zhǎng)度(int型，4個(gè)字節(jié))+消息體的方式，根據(jù)解析的消息長(zhǎng)度定長(zhǎng)解析消息內(nèi)容，雖然最后證明LZ的問(wèn)題不是由于粘包和分包造成的，但是LZ就這樣歪打正著，給實(shí)現(xiàn)了！！！數(shù)據(jù)不正常的問(wèn)題后來(lái)通過(guò)DataOutputStream和DataInputStream的方式也得到了解決。
廢話多了，帖代碼。
服務(wù)端：
package com.feng.test.longconnection1;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import org.apache.commons.lang.ArrayUtils;
/**
*
* @author songfeng
* @version 1.0
* @since 2015-10-24
* @category com.feng.test.longconnection
*
*/
public class Server
{
private Map heatTimeMap = new HashMap();
public Server(int port)
{
Selector selector = null;
ServerSocketChannel serverChannel = null;
try
{
//獲取一個(gè)ServerSocket通道
serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port));
//獲取通道管理器
selector = Selector.open();
//將通道管理器與通道綁定，并為該通道注冊(cè)SelectionKey.OP_ACCEPT事件，
//只有當(dāng)該事件到達(dá)時(shí)，Selector.select()會(huì)返回，否則一直阻塞。
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0)
{
//選擇注冊(cè)過(guò)的io操作的事件
Iterator it = selector.selectedKeys().iterator();
while (it.hasNext())
{
SelectionKey readyKey = it.next();
//刪除已選key，防止重復(fù)處理
it.remove();
if (readyKey.isAcceptable())
{
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) readyKey.channel();
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
// 連接成功后，注冊(cè)接收服務(wù)器消息的事件
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
}
else if(readyKey.isReadable())
{
SocketChannel socketChannel = (SocketChannel)readyKey.channel();
Object obj = receiveData(socketChannel);
String msg = "Server back:";
if(obj instanceof String)
{
String id = obj.toString().split(",")[0];
if(heatTimeMap.get(id) != null
&& System.currentTimeMillis() - heatTimeMap.get(id) > 5000)
{
socketChannel.socket().close();
}
else
{
heatTimeMap.put(id, System.currentTimeMillis());
}
long time = System.currentTimeMillis();
msg += time + "\n";
sendData(socketChannel, msg);
}
else if(obj instanceof Pojo)
{
msg += ((Pojo)obj).getName() + "\n";
sendData(socketChannel, msg);
}
}
}
}
}
catch (Exception e)
{
e.printStackTrace();
}
finally
{
try
{
selector.close();
if(serverChannel != null)
{
serverChannel.close();
}
}
catch (Exception e)
{
e.printStackTrace();
}
}
}
private static Object receiveData(SocketChannel socketChannel)
{
Object obj = null;
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ByteBuffer intBuffer = ByteBuffer.allocate(4);
ByteBuffer objBuffer = ByteBuffer.allocate(1024);
int size = 0;
int sum = 0;
int objlen = 0;
byte[] bytes = null;
try
{
while((size = socketChannel.read(intBuffer)) > 0)
{
intBuffer.flip();
bytes = new byte[size];
intBuffer.get(bytes);
baos.write(bytes);
intBuffer.clear();
if(bytes.length == 4)
{
objlen = bytesToInt(bytes,0);
}
if(objlen > 0)
{
byte[] objByte = new byte[0];
while(sum != objlen)
{
size = socketChannel.read(objBuffer);
if(size > 0)
{
objBuffer.flip();
bytes = new byte[size];
objBuffer.get(bytes,0,size);
baos.write(bytes);
objBuffer.clear();
objByte = ArrayUtils.addAll(objByte, bytes);
sum += bytes.length;
}
}
obj = ByteToObject(objByte);
break;
}
}
}
catch (Exception e)
{
e.printStackTrace();
}
finally
{
try
{
baos.close();
}
catch (Exception e)
{
e.printStackTrace();
}
}
return obj;
}
private static void sendData(SocketChannel socketChannel,Object obj)
{
byte[] bytes = ObjectToByte(obj);
ByteBuffer buffer = ByteBuffer.wrap(bytes);
try
{
socketChannel.write(buffer);
}
catch (IOException e)
{
e.printStackTrace();
}
}
/**
* byte數(shù)組中取int數(shù)值，本方法適用于(低位在前，高位在后)的順序。
*
* @param ary
* byte數(shù)組
* @param offset
* 從數(shù)組的第offset位開始
* @return int數(shù)值
*/
public static int bytesToInt(byte[] ary, int offset) {
int value;
value = (int) ((ary[offset]&0xFF)
| ((ary[offset+1]<<8) & 0xFF00)
| ((ary[offset+2]<<16)& 0xFF0000)
| ((ary[offset+3]<<24) & 0xFF000000));
return value;
}
public static Object ByteToObject(byte[] bytes)
{
Object obj = null;
try
{
// bytearray to object
ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
ObjectInputStream oi = new ObjectInputStream(bi);
obj = oi.readObject();
bi.close();
oi.close();
}
catch (Exception e)
{
//e.printStackTrace();
}
return obj;
}
public static byte[] ObjectToByte(Object obj)
{
byte[] bytes = null;
try
{
// object to bytearray
ByteArrayOutputStream bo = new ByteArrayOutputStream();
ObjectOutputStream oo = new ObjectOutputStream(bo);
oo.writeObject(obj);
bytes = bo.toByteArray();
bo.close();
oo.close();
}
catch (Exception e)
{
e.printStackTrace();
}
return bytes;
}
public static void main(String[] args)
{
Server server = new Server(55555);
}
}
客戶端：
package com.feng.test.longconnection1;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
/**
*
* @author songfeng
* @version 1.0
* @since 2015-10-24
* @category com.feng.test.longconnection
*
*/
public class Client
{
private Socket socket;
private String ip;
private int port;
private String id;
DataOutputStream dos;
DataInputStream dis;
public Client(String ip, int port,String id)
{
try
{
this.ip = ip;
this.port = port;
this.id = id;
this.socket = new Socket(ip, port);
//this.socket.setKeepAlive(true);
dos = new DataOutputStream(socket.getOutputStream());
dis = new DataInputStream(socket.getInputStream());
new Thread(new heartThread()).start();
new Thread(new MsgThread()).start();
}
catch (Exception e)
{
e.printStackTrace();
}
}
public void sendMsg(Object content)
{
try
{
int len = ObjectToByte(content).length;
ByteBuffer dataLenBuf = ByteBuffer.allocate(4);
dataLenBuf.order(ByteOrder.LITTLE_ENDIAN);
dataLenBuf.putInt(0, len);
dos.write(dataLenBuf.array(), 0 , 4);
dos.flush();
dos.write(ObjectToByte(content));
dos.flush();
}
catch (Exception e)
{
e.printStackTrace();
closeSocket();
}
}
public void closeSocket()
{
try
{
socket.close();
dos.close();
dis.close();
}
catch (IOException e)
{
e.printStackTrace();
}
}
public static byte[] ObjectToByte(Object obj)
{
byte[] bytes = null;
try
{
// object to bytearray
ByteArrayOutputStream bo = new ByteArrayOutputStream();
ObjectOutputStream oo = new ObjectOutputStream(bo);
oo.writeObject(obj);
bytes = bo.toByteArray();
bo.close();
oo.close();
}
catch (Exception e)
{
e.printStackTrace();
}
return bytes;
}
public static Object ByteToObject(byte[] bytes)
{
Object obj = null;
try
{
// bytearray to object
ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
ObjectInputStream oi = new ObjectInputStream(bi);
obj = oi.readObject();
bi.close();
oi.close();
}
catch (Exception e)
{
e.printStackTrace();
}
return obj;
}
class heartThread implements Runnable
{
@Override
public void run()
{
while(true)
{
try
{
Thread.sleep(1000);
long time = System.currentTimeMillis();
//System.out.println("client send:" + time);
sendMsg("Client" + id + "," + time);
}
catch (Exception e)
{
e.printStackTrace();
}
}
}
}
class MsgThread implements Runnable
{
@Override
public void run()
{
int temp;
while(true)
{
try
{
if(socket.getInputStream().available() > 0)
{
byte[] bytes = new byte[1024];
int len = 0;
while((char)(temp = dis.read()) != '\n')
{
bytes[len]=(byte)temp;
len++;
}
System.out.println(ByteToObject(bytes));
}
}
catch (Exception e)
{
closeSocket();
}
}
}
}
public static void main(String[] args)
{
Client client1 = new Client("127.0.0.1", 55555, "1");
client1.sendMsg(new Pojo("songfeng", 26, new ArrayList()));
try
{
Thread.sleep(500);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
Client client2 = new Client("127.0.0.1", 55555, "2");
}
}
數(shù)據(jù)類:
package com.feng.test.longconnection1;
import java.io.Serializable;
import java.util.List;
/**
*
* @author songfeng
* @version 1.0
* @since 2015-10-16
* @category com.feng.test.longconnection
*
*/
public class Pojo implements Serializable
{
/**
* 序列化
*/
private static final long serialVersionUID = -8868529619983791261L;
private String name;
private int age;
private List likeThing;
public Pojo(String name, int age, List likeThing)
{
super();
this.name = name;
this.age = age;
this.likeThing = likeThing;
}
public String getName()
{
return name;
}
public void setName(String name)
{
this.name = name;
}
public int getAge()
{
return age;
}
public void setAge(int age)
{
this.age = age;
}
public List getLikeThing()
{
return likeThing;
}
public void setLikeThing(List likeThing)
{
this.likeThing = likeThing;
}
}
總結(jié)
以上是生活随笔為你收集整理的java nio socket长连接_nio实现Socket长连接和心跳的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： 台电u盘量产工具_简单几步，让U盘起死回
- 下一篇： SVM多分类器算法-一对多