java分布式开发TCP/IP NIO无阻塞 Socket（(基于消息方式实现系统间的通信) ）(转)

java分布式开发TCP/IP NIO无阻塞 Socket（(基于消息方式实现系统间的通信) ）

博客分类：
java 网络编程

java
分布式 NIO socket编程

在java中可以基于java.nio.channels中的Channel和Selector的相关类来实现TCP/IP+NIO方式的系统间通信。

用于系统间通信依靠SocketChannel和ServerSocketChannel，SocketChannel用于建立连接，监听事件及操作读写，ServerSocketChannel用于监听端口及监听连接事件，可通过Selector来获取是否有要处理的事件。

服务端java代码：

Java代码



package com.java.distributed.message.tcpip;

import java.io.IOException;

import java.net.InetSocketAddress;

import java.net.ServerSocket;

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;

public class NIOServer {

/**

* @param args

* @throws IOException

*/

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

int port =7889;

//打开选择器

Selector selector=Selector.open();

//打开服务器套接字通道

ServerSocketChannel ssc=ServerSocketChannel.open();

//检索与此通道关联的服务器套接字

ServerSocket serverSocket=ssc.socket();

//将 ServerSocket 绑定到特定地址（IP 地址和端口号）

serverSocket.bind(new InetSocketAddress(port));

System.out.println("server listen on port:"+port);

//调整通道的阻塞模式

ssc.configureBlocking(false);

//向给定的选择器注册此通道，返回一个选择键。SelectionKey.OP_ACCEPT--用于套接字接受操作的操作集位

ssc.register(selector, SelectionKey.OP_ACCEPT);

while(true){

//timeout:为正，则在等待某个通道准备就绪时最多阻塞 timeout 毫秒；如果为零，则无限期地阻塞；必须为非负数

int nKeys=selector.select(1000);

if(nKeys>0){

for(SelectionKey key:selector.selectedKeys()){

/*测试此键的通道是否已准备好接受新的套接字连接--

* 如果此键的通道不支持套接字接受操作，则此方法始终返回 false

* */

if(key.isAcceptable()){

ServerSocketChannel server=(ServerSocketChannel) key.channel();

SocketChannel sc=server.accept();

if(sc==null){

continue;

}

sc.configureBlocking(false);

sc.register(selector, SelectionKey.OP_READ);

}else if(key.isReadable()){

//分配一个新的字节缓冲区

ByteBuffer buffer=ByteBuffer.allocate(1024);

SocketChannel sc=(SocketChannel) key.channel();

int readBytes=0;

String message=null;

try{

int ret;

try{

while((ret=sc.read(buffer))>0){

readBytes +=ret;

}

}catch(Exception e ){

readBytes=0;

//ignore

}finally{

//反转此缓冲区。首先对当前位置设置限制，然后将该位置设置为零

buffer.flip();

}

if(readBytes>0){

message=Charset.forName("UTF-8").decode(buffer).toString();

buffer=null;

}

}finally{

if(buffer!=null)

buffer.clear();

}

if(readBytes>0){

System.out.println("message from client:"+message);

if("quit".equalsIgnoreCase(message.trim())){

sc.close();

selector.close();

System.out.println("Server has been shutdown!");

System.exit(0);

}

String outMessage="server response:"+message;

sc.write(Charset.forName("UTF-8").encode(outMessage));

}

}

}

selector.selectedKeys().clear();

}

}

}

}

客户端java代码：

Java代码



package com.java.distributed.message.tcpip;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.net.InetSocketAddress;

import java.net.SocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.SelectionKey;

import java.nio.channels.Selector;

import java.nio.channels.SocketChannel;

import java.nio.charset.Charset;

public class NIOClient {

/**

* @param args

* @throws IOException

*/

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

int port =7889;

SocketChannel channel=SocketChannel.open();

channel.configureBlocking(false);

SocketAddress target=new InetSocketAddress("127.0.0.1",port);

channel.connect(target);

Selector selector=Selector.open();

//用于套接字连接操作的操作集位

channel.register(selector, SelectionKey.OP_CONNECT);

BufferedReader systemIn=new BufferedReader(new InputStreamReader(System.in));

while(true){

if(channel.isConnected()){

String command=systemIn.readLine();

channel.write(Charset.forName("UTF-8").encode(command));

if(command==null||"quit".equalsIgnoreCase(command.trim())){

systemIn.close();

channel.close();

selector.close();

System.out.println("Client quit !");

System.exit(0);

}

}

int nKeys=selector.select(1000);

if(nKeys>0){

for(SelectionKey key:selector.selectedKeys()){

if(key.isConnectable()){

SocketChannel sc=(SocketChannel) key.channel();

sc.configureBlocking(false);

sc.register(selector, SelectionKey.OP_READ);

sc.finishConnect();

}else if(key.isReadable()){

ByteBuffer buffer=ByteBuffer.allocate(1024);

SocketChannel sc=(SocketChannel) key.channel();

int readBytes=0;

try{

int ret=0;

try{

while((ret=sc.read(buffer))>0){

readBytes+=ret;

}

}finally{

buffer.flip();

}

if (readBytes > 0) {

System.out.println(Charset.forName("UTF-8")

.decode(buffer).toString());

buffer = null;

}

}finally {

if (buffer != null) {

buffer.clear();

}

}

}

}

selector.selectedKeys().clear();

}

}

}

}