Linux允许python使用多少个网络端口？

Question

所以我一直在尝试在 python 中对一些互联网连接进行多线程处理。我一直在使用多处理模块，所以我可以绕过“全局解释器锁”。但似乎系统只给python一个开放的连接端口，或者至少它只允许一次连接发生。这是我所说的一个例子。

*请注意，这是在 linux 服务器上运行的

from multiprocessing import Process, Queue
import urllib
import random

# Generate 10,000 random urls to test and put them in the queue
queue = Queue()
for each in range(10000):
    rand_num = random.randint(1000,10000)
    url = ('http://www.' + str(rand_num) + '.com')
    queue.put(url)

# Main funtion for checking to see if generated url is active
def check(q):
    while True:
        try:
            url = q.get(False)
            try:
                request = urllib.urlopen(url)
                del request
                print url + ' is an active url!'
            except:
                print url + ' is not an active url!'
        except:
            if q.empty():
                break

# Then start all the threads (50)
for thread in range(50):
    task = Process(target=check, args=(queue,))
    task.start()

因此，如果您运行它，您会注意到它在函数上启动了 50 个实例，但一次只运行一个。您可能认为“全局解释器锁”正在执行此操作，但事实并非如此。尝试将函数更改为数学函数而不是网络请求，您将看到所有 50 个线程同时运行。

那么我必须使用套接字吗？或者我可以做些什么来让 python 访问更多端口？或者有什么我没有看到的？让我知道你的想法！谢谢！

*编辑

所以我编写了这个脚本来更好地测试 requests 库。好像我之前没有很好地测试过它。 （我主要用过 urllib 和 urllib2）

from multiprocessing import Process, Queue
from threading import Thread
from Queue import Queue as Q
import requests
import time

# A main timestamp
main_time = time.time()

# Generate 100 urls to test and put them in the queue
queue = Queue()
for each in range(100):
    url = ('http://www.' + str(each) + '.com')
    queue.put(url)

# Timer queue
time_queue = Queue()

# Main funtion for checking to see if generated url is active
def check(q, t_q): # args are queue and time_queue
    while True:
        try:
            url = q.get(False)
            # Make a timestamp
            t = time.time()
            try:
                request = requests.head(url, timeout=5)
                t = time.time() - t
                t_q.put(t)
                del request
            except:
                t = time.time() - t
                t_q.put(t)
        except:
            break

# Then start all the threads (20)
thread_list = []
for thread in range(20):
    task = Process(target=check, args=(queue, time_queue))
    task.start()
    thread_list.append(task)

# Join all the threads so the main process don't quit
for each in thread_list:
    each.join()
main_time_end = time.time()

# Put the timerQueue into a list to get the average
time_queue_list = []
while True:
    try:
        time_queue_list.append(time_queue.get(False))
    except:
        break

# Results of the time
average_response = sum(time_queue_list) / float(len(time_queue_list))
total_time = main_time_end - main_time
line =  "Multiprocessing: Average response time: %s sec. -- Total time: %s sec." % (average_response, total_time)
print line

# A main timestamp
main_time = time.time()

# Generate 100 urls to test and put them in the queue
queue = Q()
for each in range(100):
    url = ('http://www.' + str(each) + '.com')
    queue.put(url)

# Timer queue
time_queue = Queue()

# Main funtion for checking to see if generated url is active
def check(q, t_q): # args are queue and time_queue
    while True:
        try:
            url = q.get(False)
            # Make a timestamp
            t = time.time()
            try:
                request = requests.head(url, timeout=5)
                t = time.time() - t
                t_q.put(t)
                del request
            except:
                t = time.time() - t
                t_q.put(t)
        except:
            break

# Then start all the threads (20)
thread_list = []
for thread in range(20):
    task = Thread(target=check, args=(queue, time_queue))
    task.start()
    thread_list.append(task)

# Join all the threads so the main process don't quit
for each in thread_list:
    each.join()
main_time_end = time.time()

# Put the timerQueue into a list to get the average
time_queue_list = []
while True:
    try:
        time_queue_list.append(time_queue.get(False))
    except:
        break

# Results of the time
average_response = sum(time_queue_list) / float(len(time_queue_list))
total_time = main_time_end - main_time
line =  "Standard Threading: Average response time: %s sec. -- Total time: %s sec." % (average_response, total_time)
print line

# Do the same thing all over again but this time do each url at a time
# A main timestamp
main_time = time.time()

# Generate 100 urls and test them
timer_list = []
for each in range(100):
    url = ('http://www.' + str(each) + '.com')
    t = time.time()
    try:
        request = requests.head(url, timeout=5)
        timer_list.append(time.time() - t)
    except:
        timer_list.append(time.time() - t)
main_time_end = time.time()

# Results of the time
average_response = sum(timer_list) / float(len(timer_list))
total_time = main_time_end - main_time
line = "Not using threads: Average response time: %s sec. -- Total time: %s sec." % (average_response, total_time)
print line

如您所见，它非常适合多线程。实际上，我的大多数测试表明 threading 模块实际上比 multiprocessing 模块快。 （我不明白为什么！）这是我的一些结果。

Multiprocessing: Average response time: 2.40511314869 sec. -- Total time: 25.6876308918 sec.
Standard Threading: Average response time: 2.2179402256 sec. -- Total time: 24.2941861153 sec.
Not using threads: Average response time: 2.1740363431 sec. -- Total time: 217.404567957 sec.

这是在我的家庭网络上完成的，我的服务器上的响应时间要快得多。我认为我的问题已经间接回答了，因为我在一个更复杂的脚本上遇到了问题。所有的建议都帮助我很好地优化了它。谢谢大家！

Answer 1

它在函数上启动 50 个实例，但一次只运行一个

您误解了 htop 的结果。只有少数（如果有的话）python 副本可以在任何特定实例上运行。它们中的大多数将被阻塞等待网络 I/O。

这些进程实际上是并行运行的。

尝试将函数更改为数学函数而不是网络请求，您将看到所有 50 个线程同时运行。

将任务更改为数学函数仅说明了 CPU 绑定（例如数学）和 IO 绑定（例如 urlopen）进程之间的区别。前者总是可运行的，后者很少可运行。

它一次只打印一个。如果它实际上正在运行多个进程，它将一次打印出许多进程。

它一次打印一个，因为您正在向终端写入行。因为这些行是无法区分的，所以您无法判断它们是全部由一个线程编写的，还是每个由一个单独的线程依次编写的。

Answer 2

首先，使用multiprocessing 并行化网络I/O 是一种矫枉过正。使用内置的threading 或像gevent 这样的轻量级greenlet 库是一个更好的选择，而且开销更少。 GIL 与阻塞 IO 调用无关，因此您完全不必担心。

其次，如果您正在监视 stdout，则查看您的子进程/线程/greenlets 是否并行运行的一种简单方法是在子进程/线程/greenlets 生成之后，在函数的最开头打印出一些内容。例如，像这样修改您的check() 函数

def check(q):
    print 'Start checking urls!'
    while True:
        ...

如果您的代码正确，您应该会看到在打印出任何url + ' is [not] an active url!' 之前打印出许多Start checking urls! 行。它可以在我的机器上运行，所以看起来你的代码是正确的。

Answer 3

看来您的问题实际上与gethostbyname(3) 的串行行为有关。这在this SO thread 中进行了讨论。

试试这个使用 Twisted 异步 I/O 库的代码：

import random
import sys
from twisted.internet import reactor
from twisted.internet import defer
from twisted.internet.task import cooperate
from twisted.web import client

SIMULTANEOUS_CONNECTIONS = 25
# Generate 10,000 random urls to test and put them in the queue
pages = []
for each in range(10000):
    rand_num = random.randint(1000,10000)
    url = ('http://www.' + str(rand_num) + '.com')
    pages.append(url)

# Main function for checking to see if generated url is active
def check(page):
    def successback(data, page):
        print "{} is an active URL!".format(page)

    def errback(err, page):
        print "{} is not an active URL!; errmsg:{}".format(page, err.value)

    d = client.getPage(page, timeout=3) # timeout in seconds
    d.addCallback(successback, page)
    d.addErrback(errback, page)
    return d

def generate_checks(pages):
    for i in xrange(0, len(pages)):
        page = pages[i]
        #print "Page no. {}".format(i)
        yield check(page)

def work(pages):
    print "started work(): {}".format(len(pages))
    batch_size = len(pages) / SIMULTANEOUS_CONNECTIONS
    for i in xrange(0, len(pages), batch_size):
        task = cooperate(generate_checks(pages[i:i+batch_size]))

print "starting..."
reactor.callWhenRunning(work, pages)
reactor.run()