http://blog.csdn.net/lcllcl987/archive/2007/04/16/1566114.aspx
nio學(xué)習(xí)：
最近花了點(diǎn)時(shí)間研究了一下nio，及其開(kāi)源框架MINA，現(xiàn)把心得總結(jié)如下：
1：傳統(tǒng)socket:阻塞式通信
每建立一個(gè)Socket連接時(shí)，同時(shí)創(chuàng)建一個(gè)新線程對(duì)該Socket進(jìn)行單獨(dú)通信（采用阻塞的方式通信）。
這種方式具有很高的響應(yīng)速度，并且控制起來(lái)也很簡(jiǎn)單，在連接數(shù)較少的時(shí)候非常有效，但是如果
對(duì)每一個(gè)連接都產(chǎn)生一個(gè)線程的無(wú)疑是對(duì)系統(tǒng)資源的一種浪費(fèi)，如果連接數(shù)較多將會(huì)出現(xiàn)資源不足的情況
example:
server code:
public class MultiUserServer extends Thread {
private Socket client;
public MultiUserServer(Socket c) {
this.client = c;
}
public void run() {
try {
BufferedReader in = new BufferedReader(new InputStreamReader(client
.getInputStream()));
PrintWriter out = new PrintWriter(client.getOutputStream());
// Mutil User but can't parallel
while (true) {
String str = in.readLine();
System.out.println(str);
SocketLog.debug("receive message: " + str);
out.println("has receive....");
out.flush();
if (str.equals("end"))
break;
}
client.close();
} catch (IOException ex) {
}
}
public static void main(String[] args) throws IOException {
int port = 5678;
if (args.length > 0)
port = Integer.parseInt(args[0]);
ServerSocket server = new ServerSocket(port);
SocketLog.debug("the server socket application is created!");
while (true) {
// transfer location change Single User or Multi User
MultiUserServer mu = new MultiUserServer(server.accept());
mu.start();
}
}
}
client code:
public class Client {
static Socket server;
public static void main(String[] args) throws Exception {
//set socket proxy.
String proxyHost = "192.168.254.212";
String proxyPort = "1080";
System.getProperties().put("socksProxySet","true");
System.getProperties().put("socksProxyHost",proxyHost);
System.getProperties().put("socksProxyPort",proxyPort);
String host = "132.201.69.80";
int port = 13086;
if (args.length > 1)
{
host = args[0];
port = Integer.parseInt(args[1]);
}
System.out.println("connetioning:" + host + ":" + port);
server = new Socket(host, port);
BufferedReader in = new BufferedReader(new InputStreamReader(server
.getInputStream()));
PrintWriter out = new PrintWriter(server.getOutputStream());
BufferedReader wt = new BufferedReader(new InputStreamReader(System.in));
while (true) {
String str = wt.readLine();
out.println(str);
out.flush();
if (str.equals("end")) {
break;
}
System.out.println(in.readLine());
}
server.close();
}
}
2.nio:非阻塞通訊模式
２．１NIO 設(shè)計(jì)背后的基石：反應(yīng)器模式，用于事件多路分離和分派的體系結(jié)構(gòu)模式。
反應(yīng)器模式的核心功能如下：
將事件多路分用
將事件分派到各自相應(yīng)的事件處理程序
NIO 的非阻塞 I/O 機(jī)制是圍繞 選擇器和 通道構(gòu)建的。 Channel 類表示服務(wù)器和客戶機(jī)之間的
一種通信機(jī)制。Selector 類是 Channel 的多路復(fù)用器。 Selector 類將傳入客戶機(jī)請(qǐng)求多路分
用并將它們分派到各自的請(qǐng)求處理程序。
通道(Channel 類)：表示服務(wù)器和客戶機(jī)之間的一種通信機(jī)制。
選擇器(Selector類)：是 Channel 的多路復(fù)用器。Selector 類將傳入的客戶機(jī)請(qǐng)求多路分用并將它們
分派到各自的請(qǐng)求處理程序。
簡(jiǎn)單的來(lái)說(shuō)：
NIO是一個(gè)基于事件的IO架構(gòu)，最基本的思想就是：有事件我通知你，你再去做你的事情．
而且NIO的主線程只有一個(gè)，不像傳統(tǒng)的模型，需要多個(gè)線程以應(yīng)對(duì)客戶端請(qǐng)求，也減輕
了JVM的工作量。
當(dāng)Channel注冊(cè)至Selector以后，經(jīng)典的調(diào)用方法如下：
while (somecondition) {
int n = selector.select(TIMEOUT);
if (n == 0)
continue;
for (Iterator iter = selector.selectedKeys().iterator(); iter
.hasNext();) {
if (key.isAcceptable())
doAcceptable(key);
if (key.isConnectable())
doConnectable(key);
if (key.isValid() && key.isReadable())
doReadable(key);
if (key.isValid() && key.isWritable())
doWritable(key);
iter.remove();
}
}
nio中取得事件通知，就是在selector的select事件中完成的。在selector事件時(shí)有一個(gè)線程
向操作系統(tǒng)詢問(wèn)，selector中注冊(cè)的Channel&&SelectionKey的鍵值對(duì)的各種事件是否有發(fā)生，
如果有則添加到selector的selectedKeys屬性Set中去，并返回本次有多少個(gè)感興趣的事情發(fā)生。
如果發(fā)現(xiàn)這個(gè)值>0，表示有事件發(fā)生，馬上迭代selectedKeys中的SelectionKey，
根據(jù)Key中的表示的事件，來(lái)做相應(yīng)的處理。
實(shí)際上，這段說(shuō)明表明了異步socket的核心，即異步socket不過(guò)是將多個(gè)socket的調(diào)度（或者還有他們的線程調(diào)度）
全部交給操作系統(tǒng)自己去完成，異步的核心Selector，不過(guò)是將這些調(diào)度收集、分發(fā)而已。
2.2 nio example:
server code：
public class NonBlockingServer
{
public Selector sel = null;
public ServerSocketChannel server = null;
public SocketChannel socket = null;
public int port = 4900;
String result = null;
public NonBlockingServer()
{
System.out.println("Inside default ctor");
}
public NonBlockingServer(int port)
{
System.out.println("Inside the other ctor");
this.port = port;
}
public void initializeOperations() throws IOException,UnknownHostException
{
System.out.println("Inside initialization");
sel = Selector.open();
server = ServerSocketChannel.open();
server.configureBlocking(false);
InetAddress ia = InetAddress.getLocalHost();
InetSocketAddress isa = new InetSocketAddress(ia,port);
server.socket().bind(isa);
}
public void startServer() throws IOException
{
System.out.println("Inside startserver");
initializeOperations();
System.out.println("Abt to block on select()");
SelectionKey acceptKey = server.register(sel, SelectionKey.OP_ACCEPT );
while (acceptKey.selector().select() > 0 )
{
Set readyKeys = sel.selectedKeys();
Iterator it = readyKeys.iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey)it.next();
it.remove();
if (key.isAcceptable()) {
System.out.println("Key is Acceptable");
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
socket = (SocketChannel) ssc.accept();
socket.configureBlocking(false);
SelectionKey another = socket.register(sel,SelectionKey.OP_READ|SelectionKey.OP_WRITE);
}
if (key.isReadable()) {
System.out.println("Key is readable");
String ret = readMessage(key);
if (ret.length() > 0) {
writeMessage(socket,ret);
}
}
if (key.isWritable()) {
System.out.println("THe key is writable");
String ret = readMessage(key);
socket = (SocketChannel)key.channel();
if (result.length() > 0 ) {
writeMessage(socket,ret);
}
}
}
}
}
public void writeMessage(SocketChannel socket,String ret)
{
System.out.println("Inside the loop");
if (ret.equals("quit") || ret.equals("shutdown")) {
return;
}
try
{
String s = "This is context from server!-----------------------------------------";
Charset set = Charset.forName("us-ascii");
CharsetDecoder dec = set.newDecoder();
CharBuffer charBuf = dec.decode(ByteBuffer.wrap(s.getBytes()));
System.out.println(charBuf.toString());
int nBytes = socket.write(ByteBuffer.wrap((charBuf.toString()).getBytes()));
System.out.println("nBytes = "+nBytes);
result = null;
}
catch(Exception e)
{
e.printStackTrace();
}
}
public String readMessage(SelectionKey key)
{
int nBytes = 0;
socket = (SocketChannel)key.channel();
ByteBuffer buf = ByteBuffer.allocate(1024);
try
{
nBytes = socket.read(buf);
buf.flip();
Charset charset = Charset.forName("us-ascii");
CharsetDecoder decoder = charset.newDecoder();
CharBuffer charBuffer = decoder.decode(buf);
result = charBuffer.toString();
}
catch(IOException e)
{
e.printStackTrace();
}
return result;
}
public static void main(String args[])
{
NonBlockingServer nb;
if (args.length < 1)
{
nb = new NonBlockingServer();
}
else
{
int port = Integer.parseInt(args[0]);
nb = new NonBlockingServer(port);
}
try
{
nb.startServer();
System.out.println("the nonBlocking server is started!");
}
catch (IOException e)
{
e.printStackTrace();
System.exit(-1);
}
}
}
client code:
public class Client {
public SocketChannel client = null;
public InetSocketAddress isa = null;
public RecvThread rt = null;
private String host;
private int port;
public Client(String host, int port) {
this.host = host;
this.port = port;
}
public void makeConnection() {
String proxyHost = "192.168.254.212";
String proxyPort = "1080";
System.getProperties().put("socksProxySet", "true");
System.getProperties().put("socksProxyHost", proxyHost);
System.getProperties().put("socksProxyPort", proxyPort);
int result = 0;
try {
client = SocketChannel.open();
isa = new InetSocketAddress(host, port);
client.connect(isa);
client.configureBlocking(false);
receiveMessage();
} catch (UnknownHostException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
long begin = System.currentTimeMillis();
sendMessage();
long end = System.currentTimeMillis();
long userTime = end - begin;
System.out.println("use tiem: " + userTime);
try {
interruptThread();
client.close();
System.exit(0);
} catch (IOException e) {
e.printStackTrace();
}
}
public int sendMessage() {
System.out.println("Inside SendMessage");
String msg = null;
ByteBuffer bytebuf;
int nBytes = 0;
try {
msg = "It's message from client!";
System.out.println("msg is "+msg);
bytebuf = ByteBuffer.wrap(msg.getBytes());
for (int i = 0; i < 1000; i++) {
nBytes = client.write(bytebuf);
System.out.println(i + " finished");
}
interruptThread();
try {
Thread.sleep(5000);
} catch (Exception e) {
e.printStackTrace();
}
client.close();
return -1;
} catch (IOException e) {
e.printStackTrace();
}
return nBytes;
}
public void receiveMessage() {
rt = new RecvThread("Receive THread", client);
rt.start();
}
public void interruptThread() {
rt.val = false;
}
public static void main(String args[]) {
if (args.length < 2) {
System.err.println("You should put 2 args: host,port");
} else {
String host = args[0];
int port = Integer.parseInt(args[1]);
Client cl = new Client(host, port);
cl.makeConnection();
}
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
String msg;
}
public class RecvThread extends Thread {
public SocketChannel sc = null;
public boolean val = true;
public RecvThread(String str, SocketChannel client) {
super(str);
sc = client;
}
public void run() {
int nBytes = 0;
ByteBuffer buf = ByteBuffer.allocate(2048);
try {
while (val) {
while ((nBytes = nBytes = client.read(buf)) > 0) {
buf.flip();
Charset charset = Charset.forName("us-ascii");
CharsetDecoder decoder = charset.newDecoder();
CharBuffer charBuffer = decoder.decode(buf);
String result = charBuffer.toString();
System.out.println("the server return: " + result);
buf.flip();
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
3：Socket網(wǎng)絡(luò)框架 MINA
MINA是一個(gè)網(wǎng)絡(luò)應(yīng)用框架，在不犧牲性能和可擴(kuò)展性的前提下用于解決如下問(wèn)題：
1：快速開(kāi)發(fā)自己的英勇。
2：高可維護(hù)性，高可復(fù)用性：網(wǎng)絡(luò)I/O編碼，消息的編/解碼，業(yè)務(wù)邏輯互相分離。
3：相對(duì)容易的進(jìn)行單元測(cè)試。
3.1 IoFilters：
IoFilter為MINA的功能擴(kuò)展提供了接口。它攔截所有的IO事件進(jìn)行事件的預(yù)處理和后處理(AOP)。我們可以把它想象成
Servlet的filters。
IoFilter能夠?qū)崿F(xiàn)以下幾種目的：
事件日志
性能檢測(cè)
數(shù)據(jù)轉(zhuǎn)換(e.g. SSL support)，codec
防火墻…等等
3.2 codec: ProtocolCodecFactory
MINA提供了方便的Protocol支持。如上說(shuō)講，codec在IoFilters中設(shè)置。
通過(guò)它的Encoder和Decoder，可以方便的擴(kuò)展并支持各種基于Socket的網(wǎng)絡(luò)協(xié)議，比如HTTP服務(wù)器、FTP服務(wù)器、Telnet服務(wù)器等等。
要實(shí)現(xiàn)自己的編碼/解碼器(codec)只需要實(shí)現(xiàn)interface: ProtocolCodecFactory即可.
在MINA 1.0版本，MINA已經(jīng)實(shí)現(xiàn)了幾個(gè)常用的(codec factory):
DemuxingProtocolCodecFactory,
NettyCodecFactory,
ObjectSerializationCodecFactory,
TextLineCodecFactory
其中：
TextLineCodecFactory:
A ProtocolCodecFactory that performs encoding and decoding between a text line data and a Java
string object. This codec is useful especially when you work with a text-based protocols such as SMTP and IMAP.
ObjectSerializationCodecFactory:
A ProtocolCodecFactory that serializes and deserializes Java objects. This codec is very useful when
you have to prototype your application rapidly without any specific codec.
DemuxingProtocolCodecFactory：
A composite ProtocolCodecFactory that consists of multiple MessageEncoders and MessageDecoders. ProtocolEncoder
and ProtocolDecoder this factory returns demultiplex incoming messages and buffers to appropriate MessageEncoders
and MessageDecoders.
NettyCodecFactory:
A MINA ProtocolCodecFactory that provides encoder and decoder for Netty2 Messages and MessageRecognizers.
3.3 business logic: IoHandler
MINA中，所有的業(yè)務(wù)邏輯都有實(shí)現(xiàn)了IoHandler的class完成
interfaceHandles:
all protocol events fired by MINA. There are 6 event handler methods, and they are all invoked by MINA automatically.
當(dāng)事件發(fā)生時(shí)，將觸發(fā)IoHandler中的方法:
sessionCreated, sessionOpened, sessionClosed, sessionIdle, exceptionCaught, messageReceived, messageSent
MINA 1.O中，IoHandler的實(shí)現(xiàn)類：
ChainedIoHandler, DemuxingIoHandler, IoHandlerAdapter, SingleSessionIoHandlerDelegate, StreamIoHandler
具體細(xì)節(jié)可參考javadoc。
3.4 MINA的高級(jí)主題：線程模式
MINA通過(guò)它靈活的filter機(jī)制來(lái)提供多種線程模型。
沒(méi)有線程池過(guò)濾器被使用時(shí)MINA運(yùn)行在一個(gè)單線程模式。
如果添加了一個(gè)IoThreadPoolFilter到IoAcceptor，將得到一個(gè)leader-follower模式的線程池。
如果再添加一個(gè)ProtocolThreadPoolFilter，server將有兩個(gè)線程池；
一個(gè)(IoThreadPoolFilter)被用于對(duì)message對(duì)象進(jìn)行轉(zhuǎn)換，另外一個(gè)(ProtocolThreadPoolFilter)被用于處理業(yè)務(wù)邏輯。
SimpleServiceRegistry加上IoThreadPoolFilter和ProtocolThreadPoolFilter的缺省實(shí)現(xiàn)即可適用于需
要高伸縮性的應(yīng)用。如果想使用自己的線程模型，請(qǐng)參考SimpleServiceRegistry的源代碼，并且自己
初始化Acceptor。
IoThreadPoolFilter threadPool = new IoThreadPoolFilter();threadPool.start();
IoAcceptor acceptor = new SocketAcceptor();
acceptor.getFilterChain().addLast( "threadPool", threadPool);
ProtocolThreadPoolFilter threadPool2 = new ProtocolThreadPoolFilter();
threadPool2.start();
ProtocolAcceptor acceptor2 = new IoProtocolAcceptor( acceptor );
acceptor2.getFilterChain().addLast( "threadPool", threadPool2 );
...
threadPool2.stop();
threadPool.stop();
采用MINA進(jìn)行socket開(kāi)發(fā)，一般步驟如下：
1：
server:
IoAcceptor acceptor = new SocketAcceptor(); //建立client接收器
or client:
SocketConnector connector = new SocketConnector(); //建立一個(gè)連接器
2：server的屬性配置：
SocketAcceptorConfig cfg = new SocketAcceptorConfig();
cfg.setReuseAddress(true);
cfg.getFilterChain().addLast(
"codec",
new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) ); //對(duì)象序列化 codec factory
cfg.getFilterChain().addLast( "logger", new LoggingFilter() );
3：綁定address和business logic
server:
acceptor.bind(
new InetSocketAddress( SERVER_PORT ),
new ServerSessionHandler( ), cfg ); // 綁定address和handler
client:
connector.connect(new InetSocketAddress( HOSTNAME, PORT ),
new ClientSessionHandler(msg), cfg );
下面的這個(gè)簡(jiǎn)單的example演示client和server傳遞object的過(guò)程：
Message.java
public class Message implements Serializable {
private int type;
private int status;
private String msgBody;
public Message(int type, int status, String msgBody)
{
this.type = type;
this.status = status;
this.msgBody = msgBody;
}
public String getMsgBody() {
return msgBody;
}
public void setMsgBody(String msgBody) {
this.msgBody = msgBody;
}
public int getStatus() {
return status;
}
public void setStatus(int status) {
this.status = status;
}
public int getType() {
return type;
}
public void setType(int type) {
this.type = type;
}
}
Client.java
public class Client
{
private static final String HOSTNAME = "localhost";
private static final int PORT = 8080;
private static final int CONNECT_TIMEOUT = 30; // seconds
public static void main( String[] args ) throws Throwable
{
SocketConnector connector = new SocketConnector();
// Configure the service.
SocketConnectorConfig cfg = new SocketConnectorConfig();
cfg.setConnectTimeout( CONNECT_TIMEOUT );
cfg.getFilterChain().addLast(
"codec",
new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) );
cfg.getFilterChain().addLast( "logger", new LoggingFilter() );
IoSession session;
Message msg = new Message(0,1,"hello");
connector.connect(new InetSocketAddress( HOSTNAME, PORT ),
new ClientSessionHandler(msg), cfg );
}
}
ClientSessionHandler.java
public class ClientSessionHandler extends IoHandlerAdapter
{
private Object msg;
public ClientSessionHandler(Object msg)
{
this.msg = msg;
}
public void sessionOpened( IoSession session )
{
session.write(this.msg);
}
public void messageReceived( IoSession session, Object message )
{
System.out.println("in messageReceived!");
Message rm = (Message ) message;
SessionLog.debug(session, rm.getMsgBody());
System.out.println("message is: " + rm.getMsgBody());
session.write(rm);
}
public void exceptionCaught( IoSession session, Throwable cause )
{
session.close();
}
}
Server.java
public class Server
{
private static final int SERVER_PORT = 8080;
public static void main( String[] args ) throws Throwable
{
IoAcceptor acceptor = new SocketAcceptor();
// Prepare the service configuration.
SocketAcceptorConfig cfg = new SocketAcceptorConfig();
cfg.setReuseAddress( true );
cfg.getFilterChain().addLast(
"codec",
new ProtocolCodecFilter( new ObjectSerializationCodecFactory() ) );
cfg.getFilterChain().addLast( "logger", new LoggingFilter() );
acceptor.bind(
new InetSocketAddress( SERVER_PORT ),
new ServerSessionHandler( ), cfg );
System.out.println( "The server Listening on port " + SERVER_PORT );
}
}
ServerSessionHandler.java
public class ServerSessionHandler extends IoHandlerAdapter
{
public void sessionOpened( IoSession session )
{
// set idle time to 60 seconds
session.setIdleTime( IdleStatus.BOTH_IDLE, 60 );
session.setAttribute("times",new Integer(0));
}
public void messageReceived( IoSession session, Object message )
{
System.out.println("in messageReceived");
int times = ((Integer)(session.getAttribute("times"))).intValue();
System.out.println("tiems = " + times);
// communicate 30 times,then close the session.
if (times < 30)
{
times++;
session.setAttribute("times", new Integer(times));
Message msg;
msg = (Message) message;
msg.setMsgBody("in server side: " + msg.getMsgBody());
System.out.println("begin send msg: " + msg.getMsgBody());
session.write(msg);
}
else
{
session.close();
}
}
public void sessionIdle( IoSession session, IdleStatus status )
{
SessionLog.info( session, "Disconnecting the idle." );
// disconnect an idle client
session.close();
}
public void exceptionCaught( IoSession session, Throwable cause )
{
// close the connection on exceptional situation
session.close();
}
}
MINA自己附帶的Demo已經(jīng)很好的說(shuō)明了它的運(yùn)用。
值得一提的是它的SumUp：客戶端發(fā)送幾個(gè)數(shù)字，服務(wù)端求和后并返回結(jié)果。這個(gè)簡(jiǎn)單的程序演示了如何自己實(shí)現(xiàn)CODEC。
補(bǔ)充提示：
下載并運(yùn)行MINA的demo程序還頗非周折:
運(yùn)行MINA demo appli擦tion：
1:在JDK5
產(chǎn)生錯(cuò)誤:
Exception in thread "main" java.lang.NoClassDefFoundError: edu/emory/mathcs/backport/java/util/concurrent/Executor
at org.apache.mina.example.reverser.Main.main(Main.java:44)
察看mina的QA email:
http://www.mail-archive.com/mina-dev@directory.apache.org/msg02252.html
原來(lái)需要下載:backport-util-concurrent.jar并加入classpath
http://dcl.mathcs.emory.edu/util/backport-util-concurrent/
繼續(xù)運(yùn)行還是報(bào)錯(cuò):
Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory
原來(lái)MINA采用了slf4j項(xiàng)目作為log，繼續(xù)下載
slf4j-simple.jar等，并加入classpath:
http://www.slf4j.org/download.html