和大家分享一下tornado 如何实现异步处理

ysitbook

下面的程序只能支持多线程的异步处理方式，扩展性一般，如果，可以改进为进程的方式扩展性会好很多 ：）
之所以使用tornado 是因为，tornado 性能比较乐观

Client 端代码

1. >>> import socket
   
2. >>> d = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   
3. >>> d.connect((HOST,PORT))

复制代码

Server 端代码，需要Linux 服务器

1. #encoding = utf-8
   
2. import errno
   
3. import functools
   
4. import socket
   
5. import time
   
6. import os
   
7. import traceback
   
8. import memcache
   
9. from tornado import ioloop, iostream
   
10. from threading import Thread
    
11. from Queue import Queue
    
12. from config import *
    
13. 
    
14. streamDict = {}
    
15. streamRequests = {}
    
16. 
    
17. ################################# Utils ###############################
    
18. def stream_add(stream,count):
    
19.     """
    
20.     添加 Stream
    
21.     """
    
22.     streamkey = str(stream)
    
23.     streamDict[streamkey] = (stream,0)
    
24.     streamRequests[streamkey] = Queue()
    
25. 
    
26. def stream_remove(streamkey,stream):
    
27.     """
    
28.     删除 Stream
    
29.     """
    
30.     try:
    
31.         stream.close()
    
32.     except:
    
33.         pass
    
34.     del streamDict[streamkey]
    
35.     del streamRequests[streamkey]
    
36. 
    
37. ################################# Response ###############################
    
38. class ClockResponse(Thread):
    
39.     """
    
40.     定时处理请求接口
    
41.     """
    
42. 
    
43.     def __init__ (self):
    
44.         """
    
45.         初始化
    
46.         """
    
47.         Thread.__init__(self)
    
48.         self.flag = True
    
49.         self.count = 0
    
50. 
    
51.     def run(self):
    
52.         """
    
53.         运行线程
    
54.         """
    
55.         while self.flag:
    
56.             #打印 LOG,N次/打印
    
57.             ct = self.count % 10
    
58.             if ct == 0:
    
59.                 print 'now connections:%s'%(len(streamDict))
    
60.             self.count = ct + 1
    
61. 
    
62.             #循环处理请求
    
63.             for streamkey,(stream,num) in streamDict.items():
    
64.                 try:
    
65.                     #这里返回Response
    
66.                     queue = streamRequests[streamkey]
    
67.                     print "queue.get()",queue.get()
    
68.                     print "q.qsize()",queue.qsize()
    
69.                     stream.write('haha')
    
70.                 except:
    
71.                     streamDict[streamkey] = (stream,num+1)
    
72.                     #去死吧.......
    
73.                     if num>DEAD_VALUE:
    
74.                         stream_remove(streamkey,stream)
    
75.             time.sleep(SLEEP_TIME)
    
76. 
    
77.     def stop(self):
    
78.         self.flag = False
    
79. 
    
80. ################################# Request ###############################
    
81. class SocketRequest(object):
    
82.     """
    
83.     Socket 请求
    
84.     """
    
85. 
    
86.     def __init__(self, stream, address):
    
87.         self.stream = stream
    
88.         self.streamkey = str(stream)
    
89.         self.address = address
    
90.         #开始读取数据
    
91.         self.stream.read_until("\r\n", self.on_request)
    
92. 
    
93.     def on_request(self,data):
    
94.         """
    
95.         接收处理请求
    
96.         """
    
97.         #将数据加载到 Queue
    
98.         print "on_request" , data
    
99.         queue = streamRequests[self.streamkey]
    
100.         queue.put(data)
     
101.         #继续读取请求
     
102.         self.stream.read_until("\r\n", self.on_request)
     
103. 
     
104. 
     
105. def connection_ready(sock, fd, events):
     
106.     """
     
107.     启动 Connection 监听,处理程序
     
108.     """
     
109.     while True:
     
110.         try:
     
111.             connection, address = sock.accept()
     
112.         except socket.error, e:
     
113.             if e[0] not in (errno.EWOULDBLOCK, errno.EAGAIN):
     
114.                 raise
     
115.             return
     
116.         #non-block
     
117.         connection.setblocking(0)
     
118.         stream = iostream.IOStream(connection)
     
119.         #保存stream
     
120.         stream_add(stream,0)
     
121.         #接收Socket 请求
     
122.         SocketRequest(stream,address)
     
123. 
     
124. 
     
125. if __name__ == '__main__':
     
126.    
     
127.     #pidfile = open(os.path.join(settings.PROJECT_PATH,'tornadoServer.pid'),'w')
     
128.     #pidfile.write(str(os.getpid()))
     
129.     #pidfile.close()
     
130.    
     
131.     #接收请求
     
132.     clock = ClockResponse()
     
133.     clock.start()
     
134.    
     
135.     #定义Socket
     
136.     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) #这里的0 dummy protocol for TCP
     
137.     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     
138.     sock.setblocking(0)
     
139.     sock.bind(("", SERVER_PORT))
     
140.     sock.listen(9999)
     
141.    
     
142.     #获得ioloop
     
143.     io_loop = ioloop.IOLoop.instance()
     
144.     callback = functools.partial(connection_ready, sock)
     
145.     io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
     
146. 
     
147.     #启动服务
     
148.     try:
     
149.         io_loop.start()
     
150.     except KeyboardInterrupt:
     
151.         io_loop.stop()
     
152.         clock.stop()
     
153.         print "exited cleanly"

复制代码

TCP/IP module for linux 2.6 dummy版

1、概述
    英文单词dummy是虚假的意思，dummy版的TCP/IP module就是实现一个虚假的传输层协议，该协议不能完成任何实际的网络通讯。它通过module自身提供的一个经过改造的环回网络设备驱动程序mylo，展示网络数据在TCP/IP协议栈中发送和接收的整个流程。其意义在于为后续的实际的STREAM, DGRAM, RAW的开发提供一个基本的程序框架，以及初步验证这个框架的可行性，同时也帮助更好的理解2.6内核中整个TCP/IP协议栈的实现原理。
                                          

2、编译与安装
    dummy版的module与第一版相比较，在模块划分和Makefile上作了很多改进。当前初步按TCP/IP协议栈的层次结构分为五个模块，下面是整个源代码目录树的基本情况：
    ipv4                      顶层目录。
     |
     |--inet                    inet域，主要完成inet域最顶层的一些基本操作。
     |
     |--application         应用层，现在主要为一个dummy协议的测试应用程序。
     |
     |--transport           传输层，目前只有一个dummy协议。
     |
     |--network             网络层。
     |
     |--datalink              数据链路层，目前只实现了环回网络设备驱动程序mylo。
     |
     |--core                  核心模块，连接数据链路层和网络层，传递它们之间的数据。
     |
     |--include              头文件
     |
     |--load.sh unload.sh     装载和卸载脚本
     |
     |--Makfile env.mk          makefile文件。

olivetree123

这么好的东西为什么没人顶呢，虽然我看不懂，但是我还是要顶一下!