tornado实现高性能无阻塞udp通信(2)——实现异步udp客户端

上一篇实现了基于tornado的高性能无阻塞udp服务器，然而光有服务器还是不够的，那么这篇就来实现一个基于tornado的异步udp客户端。

先来分析下客户端的任务：因为整个系统是用来进行代替传统web api的，所以这里的udp客户端不是只发一个包就万事大吉了，而是发出一个包后等待server端返回结果。看似很简单的任务放到tornado上就没那么简单了。传统多进程模型每个访客对应一个进程，在这种模型下我们可以发出udp包之后阻塞该进程等待结果返回。然而对于tornado这种单进程模型，阻塞无异于自杀，所以需要异步写法从而使得发出去一个包之后，tornado进程迅速去处理其他请求等到结果返回时再通过回调函数继续处理该请求。

分析完之后就开始动手啦。先简单封装一个UDPRequest,便于后面处理。只有一点要注意的，数据以'\n\r\n\r'结尾。

class UDPRequest(object):
def __init__(self,addr,port,data):
self.addr = addr
self.port = port
self.data = data

def __getattribute__(self,name):
data = object.__getattribute__(self,name)
if name == 'data' and data.rfind('\r\n\r\n') != len(data)-4 or len(data) < 4:
data += '\r\n\r\n'
return data

再封装一个_UDPConnection,其实严格来说udp是没有链接的，但是因为我们在这里每个请求都要收到一个返回结果所以在这里就认为一个包发出去了就建立了一个'链接'，过一会结果就会从这个'链接'返回。
class _UDPConnection(object):
def __init__(self,io_loop,client,request,release_callback,
final_callback,max_buffer_size):
self.start_time = time.time()
self.io_loop = io_loop
self.client = client
self.request = request
self.release_callback = release_callback
self.final_callback = final_callback

addrinfo = socket.getaddrinfo(request.addr,request.port,
socket.AF_INET,socket.SOCK_DGRAM,0,0)
af,socktype,proto,canonname,sockaddr = addrinfo[0]
self.stream = IOStream(socket.socket(af,socktype,proto),
io_loop=self.io_loop,max_buffer_size=2500)
self.stream.connect(sockaddr,self._on_connect)

def _on_connect(self):
self.stream.write(self.request.data)
self.stream.read_until('\r\n\r\n',self._on_response)

def _on_response(self,data):
if self.release_callback is not None:
release_callback = self.release_callback
self.release_callback = None
release_callback()
self.stream.close()

简单讲一下_UDPConnection的思路，正如前面所说的，发送出数据包之后就认为这个socket已经和server建立的一对一的'链接'，于是利用这个socket构造一个IOStream,让tornado帮我们实现无阻塞io操作。一旦这个socket'链接好'，IOStream便会调用 _on_connect()函数，在_on_connect()函数中首先发送数据，然后设定一旦读到'\r\n\r\n'就认为数据包结束了便调用_on_response()函数。

ps:解释一下红字‘链接好’，并不是传统意义上的三次握手建立链接，实际上udp是不用链接的，这里只是为了兼容默认的IOStream，调用这个函数只是为了修改IOStream内部的一个参数让IOStream以为链接已经建立好了。

最后把client的代码放上来

class AsyncUDPClient(object):
def __init__(self, io_loop=None):
self.io_loop = io_loop or IOLoop.instance()
self.max_clients = 10
self.queue = collections.deque()
self.active = {}
self.max_buffer_size = 2500

def fetch(self,request,callback,**kwargs):
callback = stack_context.wrap(callback)
self.queue.append((request,callback))
self._process_queue()

def _process_queue(self):
with stack_context.NullContext():
while self.queue and len(self.active) < self.max_clients:
request, callback = self.queue.popleft()
key = object()
self.active[key] = (request,callback)
_UDPConnection(self.io_loop,self,request,
functools.partial(self._release_fetch,key),
callback,
self.max_buffer_size)

def _release_fetch(self,key):
del self.active[key]
self._process_queue()

不难理解，这个client就是生成_UDPConnection对象

到这里异步udp客户端就完成了，使用方法和官方的httpclient一模一样。当然，这里放出来的代码还有很多细节例如超时，丢包之类的没有处理，所以想用在生产环境的话还需要完善一些细节。