NIO应用实现多客户端与服务端通信

服务端程序：

package com.bh.server;


import java.io.IOException;
import java.net.InetSocketAddress;  
import java.net.ServerSocket;  
import java.net.Socket;  
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.nio.charset.CharsetDecoder;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;  
import java.util.Map;  
import java.util.Set;  

/**
 * @author cyq
 * NIO通讯服务端
 */
public class NIOSServer {

    private int port = 8888;
    //解码buffer
    private CharsetDecoder decode = Charset.forName("UTF-8").newDecoder();
    /*发送数据缓冲区*/
    private ByteBuffer sBuffer = ByteBuffer.allocate(1024);
    /*接受数据缓冲区*/
    private ByteBuffer rBuffer = ByteBuffer.allocate(1024);
    /*映射客户端channel */
    private Map<String, SocketChannel> clientsMap = new HashMap<String, SocketChannel>(); 
    private Selector selector;
    private SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd HH:mm:ss", java.util.Locale.US);

    public NIOSServer(){
        try {
            init();
            listen();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private void init() throws IOException{
        /* 
         *启动服务器端，配置为非阻塞，绑定端口，注册accept事件 
         *ACCEPT事件：当服务端收到客户端连接请求时，触发该事件 
         */  
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  
        serverSocketChannel.configureBlocking(false);  
        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);
            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 = decode.decode(rBuffer.asReadOnlyBuffer()).toString();
                System.out.println(client.toString()+":"+receiveText);
                sBuffer.clear();
                sBuffer.put((sdf.format(new Date())+"服务器收到你的消息").getBytes());
                sBuffer.flip();
                client.write(sBuffer);
                dispatch(client, receiveText);
                client = (SocketChannel) selectionKey.channel();
                client.register(selector, SelectionKey.OP_READ);
            }  
        }   
    }  

    /** 
     * 把当前客户端信息 推送到其他客户端 
     */
    private void dispatch(SocketChannel client,String info) throws IOException{  

        Socket s = client.socket();  
        String name = "["+s.getInetAddress().toString().substring(1)+":"+Integer.toHexString(client.hashCode())+"]";  
        if(!clientsMap.isEmpty()){
            for(Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()){
                SocketChannel temp = entry.getValue();
                if(!client.equals(temp)){
                    sBuffer.clear();
                    sBuffer.put((name+":"+info).getBytes());
                    sBuffer.flip();
                    //输出到通道
                    temp.write(sBuffer);
                }
            }
        }
        clientsMap.put(name, client);
    }

    public static void main(String[] args) throws IOException {

        new NIOSServer();
    }  
}  
客户端程序:

package com.bh.client;
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.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;

public class NIOClient {

/*发送数据缓冲区*/
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 NIOClient(){

    SERVER = new InetSocketAddress("localhost", 8888);
    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 NIOClient();  
}  

/**
 * @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()); 
            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();
                            /*
                             * 未注册WRITE事件，因为大部分时间channel都是可以写的 
                             */
                            sBuffer.put(charset.encode(sendText.trim()));
                            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();
        rBuffer.clear();
        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);
        }
    }
} 

}