的Python socket.gethostbyname_ex（）多线程失败

Question

我编程应该可以解决多个主机名到使用多线程的IP地址的脚本。

但是，它未能在一些随机点冻结。这怎么解决？

num_threads = 100 
conn = pymysql.connect(host='xx.xx.xx.xx', unix_socket='/tmp/mysql.sock', user='user', passwd='pw', db='database') 
cur = conn.cursor() 
def mexec(befehl): 
    cur = conn.cursor() 
    cur.execute(befehl) 

websites = ['facebook.com','facebook.org' ... ... ... ...] \#10.000 websites in array 
queue = Queue() 
def getips(i, q): 
    while True: 
     #--resolve IP-- 
     try: 
      result = socket.gethostbyname_ex(site) 
      print(result) 
      mexec("UPDATE sites2block SET ip='"+result+"', updated='yes' ") #puts site in mysqldb 
     except (socket.gaierror): 
      print("no ip") 
      mexec("UPDATE sites2block SET ip='no ip', updated='yes',") 
     q.task_done() 
#Spawn thread pool 
for i in range(num_threads): 
    worker = Thread(target=getips, args=(i, queue)) 
    worker.setDaemon(True) 
    worker.start() 
#Place work in queue 
for site in websites: 
    queue.put(site) 
#Wait until worker threads are done to exit 
queue.join() 

Answer 1

你可以使用一个警戒值信号线，没有工作并加入线程，而不是queue.task_done()和queue.join()：

#!/usr/bin/env python 
import socket 
from Queue import Queue 
from threading import Thread 

def getips(queue): 
    for site in iter(queue.get, None): 
     try: # resolve hostname 
      result = socket.gethostbyname_ex(site) 
     except IOError, e: 
      print("error %s reason: %s" % (site, e)) 
     else: 
      print("done %s %s" % (site, result)) 

def main(): 
    websites = "youtube google non-existent.example facebook yahoo live".split() 
    websites = [name+'.com' for name in websites] 

    # Spawn thread pool 
    queue = Queue() 
    threads = [Thread(target=getips, args=(queue,)) for _ in range(20)] 
    for t in threads: 
     t.daemon = True 
     t.start() 

    # Place work in queue 
    for site in websites: queue.put(site) 
    # Put sentinel to signal the end 
    for _ in threads: queue.put(None) 
    # Wait for completion 
    for t in threads: t.join() 

main() 

gethostbyname_ex()功能已经过时了。要支持IPv4/v6地址，您可以使用socket.getaddrinfo()。

Answer 2

我的第一个想法是，你因超载的DNS错误 - 也许你的解析器只是不允许你这样做超过一定每次查询的多。

---

此外，我发现了一些问题：

1. 你忘了在while循环正确分配site - 这可能会更好地由for循环遍历队列，或东西来代替。在您的版本，您使用site变量从模块级的命名空间，这可能会导致双的和其他人跳过查询。

   在这个地方，你可以控制，如果队列中仍有条目或等待一些。如果两者都不是，你可以退出你的线程。

2. 为了安全起见，你最好做

   def mexec(befehl, args=None): 
       cur = conn.cursor() 
       cur.execute(befehl, args) 
   

   为了做到事后

   mexec("UPDATE sites2block SET ip=%s, updated='yes'", result) #puts site in mysqldb 
   

---

为了留在未来的协议兼容，你应该使用socket.getaddrinfo()而不是socket.gethostbyname_ex(site)。你可以得到你想要的所有IP（起初，你可以限制到IPv4，但切换到IPv6更容易），也可以把它们全部放入数据库。

---

为了您的队列，代码样本可能是

def queue_iterator(q): 
    """Iterate over the contents of a queue. Waits for new elements as long as the queue is still filling.""" 
    while True: 
     try: 
      item = q.get(block=q.is_filling, timeout=.1) 
      yield item 
      q.task_done() # indicate that task is done. 
     except Empty: 
      # If q is still filling, continue. 
      # If q is empty and not filling any longer, return. 
      if not q.is_filling: return 

def getips(i, q): 
    for site in queue_iterator(q): 
     #--resolve IP-- 
     try: 
      result = socket.gethostbyname_ex(site) 
      print(result) 
      mexec("UPDATE sites2block SET ip=%s, updated='yes'", result) #puts site in mysqldb 
     except (socket.gaierror): 
      print("no ip") 
      mexec("UPDATE sites2block SET ip='no ip', updated='yes',") 
# Indicate it is filling. 
q.is_filling = True 
#Spawn thread pool 
for i in range(num_threads): 
    worker = Thread(target=getips, args=(i, queue)) 
    worker.setDaemon(True) 
    worker.start() 
#Place work in queue 
for site in websites: 
    queue.put(site) 
queue.is_filling = False # we are done filling, if q becomes empty, we are done. 
#Wait until worker threads are done to exit 
queue.join() 

应该做的伎俩。

---

另一个问题是你的并行插入MySQL。您一次只能执行一次MySQL查询。所以，你既可以保护通过threading.Lock()或RLock()访问，或者你可以把答案成被另一个线程，它甚至可以捆绑他们处理的另一队列。

Answer 3

您可能会发现更容易使用concurrent.futures比threading，multiprocessing，Queue直接：

#!/usr/bin/env python3 
import socket 
# pip install futures on Python 2.x 
from concurrent.futures import ThreadPoolExecutor as Executor 

hosts = "youtube.com google.com facebook.com yahoo.com live.com".split()*100 
with Executor(max_workers=20) as pool: 
    for results in pool.map(socket.gethostbyname_ex, hosts, timeout=60): 
     print(results) 

注意：您可以轻松地使用线程进程切换：

from concurrent.futures import ProcessPoolExecutor as Executor 

你需要它，如果gethostbyname_ex()是不是线程安全的在你的OS例如，it might be the case on OSX。

如果您想以处理gethostbyname_ex()可能出现的异常：

import concurrent.futures 

with Executor(max_workers=20) as pool: 
    future2host = dict((pool.submit(socket.gethostbyname_ex, h), h) 
         for h in hosts) 
    for f in concurrent.futures.as_completed(future2host, timeout=60): 
     e = f.exception() 
     print(f.result() if e is None else "{0}: {1}".format(future2host[f], e)) 

它类似于the example from the docs。