前段時間用bio方式，也就是傳統io實現了socket的長連接和心跳，總覺著服務端開啟多線程管理socket連接的方式過于消耗資源，數據并發的情況下可能會影響到性能，因此就嘗試使用nio改進原來的代碼。

然而改進的過程卻不像我起初設想的那般容易，可以說一波三折，原因主要是nio讀寫都是字節流，LZ一開始依然通過ObjectOutputStream.writeObject直接向Socket服務端發送數據，然而問題出現了，每次從ByteBuffer解析出來字節流都不一樣，LZ使出渾身解數，一個字節一個字節的讀取啊，問題沒有了，可是由于是長連接，數據怎么解析啊，查資料，找大神，最后一個網友說有可能是粘包和分包的問題，一時暈菜，LZ網絡可是渣渣啊，行吧，惡補一番，想了解的童鞋可以看看這個。http://blog.csdn.net/sunmenggmail/article/details/38952131

實現原理就像很多協議那樣，自定義一套傳輸協議，比如消息長度(int型，4個字節)+消息體的方式，根據解析的消息長度定長解析消息內容，雖然最后證明LZ的問題不是由于粘包和分包造成的，但是LZ就這樣歪打正著，給實現了！！！數據不正常的問題后來通過DataOutputStream和DataInputStream的方式也得到了解決。

廢話多了，帖代碼。

服務端：

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

{

//獲取一個ServerSocket通道

serverChannel = ServerSocketChannel.open();

serverChannel.configureBlocking(false);

serverChannel.socket().bind(new InetSocketAddress(port));

//獲取通道管理器

selector = Selector.open();

//將通道管理器與通道綁定，并為該通道注冊SelectionKey.OP_ACCEPT事件，

//只有當該事件到達時，Selector.select()會返回，否則一直阻塞。

serverChannel.register(selector, SelectionKey.OP_ACCEPT);

while (selector.select() > 0)

{

//選擇注冊過的io操作的事件

Iterator it = selector.selectedKeys().iterator();

while (it.hasNext())

{

SelectionKey readyKey = it.next();

//刪除已選key，防止重復處理

it.remove();

if (readyKey.isAcceptable())

{

ServerSocketChannel serverSocketChannel = (ServerSocketChannel) readyKey.channel();

SocketChannel socketChannel = serverSocketChannel.accept();

socketChannel.configureBlocking(false);

// 連接成功后，注冊接收服務器消息的事件

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數組中取int數值，本方法適用于(低位在前，高位在后)的順序。

*

* @param ary

* byte數組

* @param offset

* 從數組的第offset位開始

* @return int數值

*/

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");

}

}

數據類:

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;

}

}

總結

以上是生活随笔為你收集整理的java nio socket长连接_nio实现Socket长连接和心跳的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。