java NIO 服务器与多客户端双向通信 非阻塞
服务端代码：
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
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.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
public class ServerSocket {
private int port = 6001;
//解码buffer
// private CharsetDecoder decode = Charset.forName("UTF-8").newDecoder();
/*发送数据缓冲区*/
private ByteBuffer sBuffer = ByteBuffer.allocate(1024);
/*接受数据缓冲区*/
private ByteBuffer rBuffer = ByteBuffer.allocate(1024);
/*映射客户端channel */
private String sendText;
private Map<String, SocketChannel> clientsMap = new HashMap<String, SocketChannel>();
private Charset charset = Charset.forName("UTF-8");
private Selector selector;
private SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss", java.util.Locale.US);
int i=0;
public ServerSocket(){
try {
init();
listen();
} catch (Exception e) {
e.printStackTrace();
}
}
private void init() throws Exception{
/*
*启动服务器端，配置为非阻塞，绑定端口，注册accept事件
*ACCEPT事件：当服务端收到客户端连接请求时，触发该事件
*/
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
java.net.ServerSocket serverSocket = serverSocketChannel.socket();
serverSocket.bind(new InetSocketAddress(port));
selector = Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("server start on port:"+port);
}
/**
* 服务器端轮询监听，select方法会一直阻塞直到有相关事件发生或超时
*/
private void listen(){
while (true) {
try {
selector.select();//返回值为本次触发的事件数
Set<SelectionKey> selectionKeys = selector.selectedKeys();
for(SelectionKey key : selectionKeys){
handle(key);
}
selectionKeys.clear();//清除处理过的事件
} catch (Exception e) {
e.printStackTrace();
break;
}
}
}
/**
* 处理不同的事件
*/
private void handle(SelectionKey selectionKey) throws IOException {
ServerSocketChannel server = null;
SocketChannel client = null;
String receiveText=null;
int count=0;
if (selectionKey.isAcceptable()) {
/*
* 客户端请求连接事件
* serversocket为该客户端建立socket连接，将此socket注册READ事件，监听客户端输入
* READ事件：当客户端发来数据，并已被服务器控制线程正确读取时，触发该事件
*/
server = (ServerSocketChannel) selectionKey.channel();
client = server.accept();
client.configureBlocking(false);
clientsMap.put(client.getLocalAddress().toString().substring(1)+ i++,client);
client.register(selector, SelectionKey.OP_READ);
} else if (selectionKey.isReadable()) {
/*
* READ事件，收到客户端发送数据，读取数据后继续注册监听客户端
*/
client = (SocketChannel) selectionKey.channel();
rBuffer.clear();
count = client.read(rBuffer);
if (count > 0) {
rBuffer.flip();
receiveText = charset.decode(rBuffer.asReadOnlyBuffer()).toString();
System.out.println(client.getLocalAddress().toString().substring(1)+":"+receiveText);
sBuffer.clear();
sBuffer.put((sdf.format(new Date())+"服务器收到你的消息").getBytes("UTF-8"));
sBuffer.flip();
client.write(sBuffer);
new Thread(){
@Override
public void run() {
while(true){
try {
//String name = "["+client.getInetAddress().toString().substring(1)+"]";
InputStreamReader input = new InputStreamReader(System.in);
BufferedReader br = new BufferedReader(input);
sendText = br.readLine();
if(!clientsMap.isEmpty()){
for(Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()){
SocketChannel temp = entry.getValue();
String name=entry.getKey();
sBuffer.clear();
sBuffer.put((name+":"+sendText).getBytes("UTF-8"));
sBuffer.flip();
//输出到通道
temp.write(sBuffer);
}
}
} catch (IOException e) {
e.printStackTrace();
break;
}
}
}
}.start();
client = (SocketChannel) selectionKey.channel();
client.register(selector, SelectionKey.OP_READ);
}
}
}
public static void main(String[] args) throws IOException {
new ServerSocket();
}
}
客户端代码：
import com.sun.org.apache.xerces.internal.impl.xpath.regex.ParseException;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;
public class Client {
/*发送数据缓冲区*/
private static ByteBuffer sBuffer = ByteBuffer.allocate(1024);
/*接受数据缓冲区*/
private static ByteBuffer rBuffer = ByteBuffer.allocate(1024);
/*服务器端地址*/
private InetSocketAddress SERVER;
private Selector selector;
private SocketChannel client;
private String receiveText;
private String sendText;
private int count=0;
private Charset charset = Charset.forName("UTF-8");
private SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss", java.util.Locale.US);
public Client(){
SERVER = new InetSocketAddress("localhost", 6001);
init();
}
/**
*
*/
public void init(){
try {
/*
* 客户端向服务器端发起建立连接请求
*/
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_CONNECT);
socketChannel.connect(SERVER);
/*
* 轮询监听客户端上注册事件的发生
*/
while (true) {
selector.select();
Set<SelectionKey> keySet = selector.selectedKeys();
for(final SelectionKey key : keySet){
handle(key);
}
keySet.clear();
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
new Client();
}
/**
* @param selectionKey
* @throws IOException
* @throws ParseException
*/
private void handle(SelectionKey selectionKey) throws IOException, ParseException{
if (selectionKey.isConnectable()) {
/*
* 连接建立事件，已成功连接至服务器
*/
client = (SocketChannel) selectionKey.channel();
if (client.isConnectionPending()) {
client.finishConnect();
System.out.println("connect success !");
sBuffer.clear();
sBuffer.put((sdf.format(new Date())+" connected!你好").getBytes("UTF-8"));
sBuffer.flip();
client.write(sBuffer);//发送信息至服务器
/*
* 启动线程一直监听客户端输入，有信心输入则发往服务器端
* 因为输入流是阻塞的，所以单独线程监听
*/
new Thread(){
@Override
public void run() {
while(true){
try {
sBuffer.clear();
InputStreamReader input = new InputStreamReader(System.in);
BufferedReader br = new BufferedReader(input);
sendText = br.readLine();
sBuffer.put(sendText.trim().getBytes("UTF-8"));
sBuffer.flip();
client.write(sBuffer);
} catch (IOException e) {
e.printStackTrace();
break;
}
}
}
}.start();
}
//注册读事件
client.register(selector, SelectionKey.OP_READ);
} else if (selectionKey.isReadable()) {
/*
* 读事件触发
* 有从服务器端发送过来的信息，读取输出到屏幕上后，继续注册读事件
* 监听服务器端发送信息
*/
client = (SocketChannel) selectionKey.channel();
count=client.read(rBuffer);
if(count>0){
receiveText = new String( rBuffer.array(),0,count);
System.out.println(receiveText);
client = (SocketChannel) selectionKey.channel();
client.register(selector, SelectionKey.OP_READ);
rBuffer.clear();
}
}
}
}