Mina、Netty、Twisted一起学（九）：异步IO和回调函数

用过JavaScript或者jQuery的同学都知道，JavaScript特别是jQuery中存在大量的回调函数，比如Ajax、jQuery的动画等。

$.get(url, function() {
	doSomething1(); // (3)
}); // (1)
doSomething2();  // (2)

上面的代码是jQuery的Ajax，因为Ajax是异步的，所以在请求URL的过程中并不会堵塞程序，也就是程序运行到（1）并不用等待Ajax请求的结果，就继续往下运行（2）。而$.get的第二个參数是一个回调函数，当Ajax请求完毕后。才会调用这个回调函数运行（3）。

这个样例仅仅是用来理解一下异步的概念，没玩过JS看不懂的同学也没关系。

在传统的IO（BIO/堵塞IO）中，全部IO操作都会堵塞当前线程，直到操作完毕，全部步骤都是一步一步进行。比如：

OutputStream output = socket.getOutputstream();
out.write(data); // IO操作完毕后返回
out.flush();
System.out.println("消息发送完毕");

可是在异步网络编程中，因为IO操作是异步的，也就是一个IO操作不会堵塞去等待操作结果。程序就会继续向下运行。

在MINA、Netty、Twisted中，非常多网络IO操作都是异步的，比方向网络的还有一端write写数据、client连接server的connect操作等。

比如Netty的write方法（以及writeAndFlush方法），运行完write语句后并不表示数据已经发送出去。而仅仅是開始发送这个数据，程序并不堵塞等待发送完毕，而是继续往下运行。所以假设有某些操作想在write完毕后再运行，比如write完毕后关闭连接。以下这些写法就有问题了，它可能会在数据write出去之前先关闭连接：

Channel ch = ...;
ch.writeAndFlush(message);
ch.close();

那么问题就来了。挖掘机...既然上面的写法不对。那么怎样在IO操作完毕后再做一些其它操作？

当异步IO操作完毕后，不管成功或者失败。都会再通知程序。至于怎样处理发送完毕的通知，在MINA、Netty、Twisted中。都会有相似回调函数的实现方式。

Netty：

在Netty中，write及writeAndFlush方法有个返回值，类型是ChannelFuture。

ChannelFuture的addListener方法能够加入ChannelFutureListener监听器。ChannelFutureListener接口的抽象方法operationComplete会在write完毕（不管成功或失败）时被调用，我们仅仅须要实现这种方法就可以处理一些write完毕后的操作。比如write完毕后关闭连接。

@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg)
		throws UnsupportedEncodingException {
	
	// 读操作省略...
	
	// 发送数据到client
	ChannelFuture future = ctx.writeAndFlush("message"); // 返回值类型为ChannelFuture
	future.addListener(new ChannelFutureListener() {
		
		// write操作完毕后调用的回调函数
		@Override
		public void operationComplete(ChannelFuture future) throws Exception {
			if(future.isSuccess()) { // 是否成功
				System.out.println("write操作成功");
			} else {
				System.out.println("write操作失败");
			}
			ctx.close(); // 假设须要在write后关闭连接，close应该写在operationComplete中。
注意close方法的返回值也是ChannelFuture
			}
	});
		
}

上面代码中，close操作是在operationComplete中进行，这样能够保证不会在write完毕之前close连接。

注意close关闭连接相同是异步的，其返回值也是ChannelFuture。

MINA：

MINA和Netty相似。session.write返回值是WriteFuture类型。

WriteFuture也能够通过addListener方法加入IoFutureListener监听器。IoFutureListener接口的抽象方法operationComplete会在write完毕（不管成功或失败）时被调用。

假设须要在write完毕后进行一些操作。仅仅需实现operationComplete方法。

@Override
public void messageReceived(IoSession session, Object message)
		throws Exception {

	// 读操作省略...

	// 发送数据到client
	WriteFuture future = session.write("message");
	future.addListener(new IoFutureListener<WriteFuture>() {

		// write操作完毕后调用的回调函数
		@Override
		public void operationComplete(WriteFuture future) {
			if(future.isWritten()) {
				System.out.println("write操作成功");
			} else {
				System.out.println("write操作失败");
			}
		}
	});
}

MINA和Netty不同在于关闭连接的close方法。当中无參数的close()方法已经弃用。而是使用带一个boolean类型參数的close(boolean immediately)方法。

immediately为true则直接关闭连接，为false则是等待全部的异步write完毕后再关闭连接。

所以以下这样的写法是正确的：

session.write("message");
session.close(false); // 尽管write是异步的，可是immediately參数为false会等待write完毕后再关闭连接

Twisted：

在Twisted中，twisted.internet.defer.Deferred相似于MINA、Netty中的Future，能够用于加入在异步IO完毕后的回调函数。

可是Twisted并没有在write方法中返回Deffered。尽管write方法也是异步的。以下就用Twisted实现一个简单的TCPclient来学习Deffered的使用方法。

# -*- coding:utf-8 –*- 

from twisted.internet import reactor
from twisted.internet.protocol import Protocol
from twisted.internet.endpoints import TCP4ClientEndpoint, connectProtocol

class ClientProtocol(Protocol):
    def sendMessage(self):
        self.transport.write("Message") # write也是异步的
        self.transport.loseConnection() # loseConnection会等待write完毕后再关闭连接

# 连接server成功的回调函数
def connectSuccess(p):
    print "connectSuccess"
    p.sendMessage()

# 连接server失败的回调函数
def connectFail(failure):
    print "connectFail"

point = TCP4ClientEndpoint(reactor, "localhost", 8080)
d = connectProtocol(point, ClientProtocol()) # 异步连接到server，返回Deffered
d.addCallback(connectSuccess) # 设置连接成功后的回调函数
d.addErrback(connectFail) # 设置连接失败后的回调函数
reactor.run()

在Twisted中，关闭连接的loseConnection方法会等待异步的write完毕后再关闭，相似于MINA中的session.close(false)。Twisted中想要直接关闭连接能够使用abortConnection，相似MINA中的session.close(true)。

作者：叉叉哥   转载请注明出处：http://blog.csdn.net/xiao__gui/article/details/40016907