import socket 
def scan(port): 
  s = socket.socket() 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
if __name__ == '__main__': 
  map(scan,range(1,65536))

這樣一個最簡單的端口掃描器出來了。
等等喂，半天都沒反應，那是因為socket是阻塞的，每次連接要等很久才超時。
我們自己給它加上的超時。

s.settimeout(0.1)

再跑一遍，感覺快多了。

多線程版本

import socket 
import threading 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
if __name__ == '__main__': 
  threads = [threading.Thread(target=scan, args=(i,)) for i in xrange(1,65536)] 
  map(lambda x:x.start(),threads)

運行一下，哇，好快，快到拋出錯誤了。thread.error: can't start new thread。
想一下，這個進程開啟了65535個線程，有兩種可能，一種是超過最大線程數(shù)了，一種是超過最大socket句柄數(shù)了。在linux可以通過ulimit來修改。
如果不修改最大限制，怎么用多線程不報錯呢？
加個queue,變成生產(chǎn)者-消費者模式,開固定線程。

多線程+隊列版本

import socket 
import threading 
from Queue import Queue 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
def worker(): 
  while not q.empty(): 
    port = q.get() 
    try: 
      scan(port) 
    finally: 
      q.task_done() 
 
if __name__ == '__main__': 
  q = Queue() 
  map(q.put,xrange(1,65535)) 
  threads = [threading.Thread(target=worker) for i in xrange(500)] 
  map(lambda x:x.start(),threads) 
  q.join()

這里開500個線程，不停的從隊列取任務來做。

multiprocessing+隊列版本
總不能開65535個進程吧？還是用生產(chǎn)者消費者模式

import multiprocessing 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
def worker(q): 
  while not q.empty(): 
    port = q.get() 
    try: 
      scan(port) 
    finally: 
      q.task_done() 
 
if __name__ == '__main__': 
  q = multiprocessing.JoinableQueue() 
  map(q.put,xrange(1,65535)) 
  jobs = [multiprocessing.Process(target=worker, args=(q,)) for i in xrange(100)] 
  map(lambda x:x.start(),jobs)

注意這里把隊列作為一個參數(shù)傳入到worker中去，因為是process safe的queue，不然會報錯。
還有用的是JoinableQueue()，顧名思義就是可以join()的。

gevent的spawn版本

from gevent import monkey; monkey.patch_all(); 
import gevent 
import socket 
... 
if __name__ == '__main__': 
  threads = [gevent.spawn(scan, i) for i in xrange(1,65536)] 
  gevent.joinall(threads)

注意monkey patch必須在被patch的東西之前import,不然會Exception KeyError.比如不能先import threading,再monkey patch.

gevent的Pool版本

from gevent import monkey; monkey.patch_all(); 
import socket 
from gevent.pool import Pool 
... 
if __name__ == '__main__': 
  pool = Pool(500) 
  pool.map(scan,xrange(1,65536)) 
  pool.join()

concurrent.futures版本

import socket 
from Queue import Queue 
from concurrent.futures import ThreadPoolExecutor 
... 
if __name__ == '__main__': 
  q = Queue() 
  map(q.put,xrange(1,65536)) 
  with ThreadPoolExecutor(max_workers=500) as executor: 
    for i in range(500): 
      executor.submit(worker,q)

您可能感興趣的文章: