0

我有一个针对任何特定 url 执行负载测试的代码。但是我必须对具有不同 URL 的 Web 服务进行负载测试。为此，我需要创建一个 URL 数组，每个线程都应该访问数组中给定的所有 URL。我怎样才能做到这一点？这是我的代码：

import httplib2

import socket

import time

from threading import Event
from threading import Thread
from threading import current_thread
from urllib import urlencode

# Modify these values to control how the testing is done

# How many threads should be running at peak load.
NUM_THREADS = 50

# How many minutes the test should run with all threads active.
TIME_AT_PEAK_QPS = 20 # minutes

# How many seconds to wait between starting threads.
# Shouldn't be set below 30 seconds.
DELAY_BETWEEN_THREAD_START = 30 # seconds

quitevent = Event()

def threadproc():
    """This function is executed by each thread."""
    print "Thread started: %s" % current_thread().getName()
    h = httplib2.Http(timeout=30)
    while not quitevent.is_set():
        try:
            # HTTP requests to exercise the server go here
            # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
            resp, content = h.request(
                "http://www.google.com")
            if resp.status != 200:
             print "Response not OK"
            # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
        except socket.timeout:
            pass

print "Thread finished: %s" % current_thread().getName()


if __name__ == "__main__":
    runtime = (TIME_AT_PEAK_QPS * 60 + DELAY_BETWEEN_THREAD_START * NUM_THREADS)
    print "Total runtime will be: %d seconds" % runtime
    threads = []
    try:
        for i in range(NUM_THREADS):
            t = Thread(target=threadproc)
            t.start()
            threads.append(t)
            time.sleep(DELAY_BETWEEN_THREAD_START)
        print "All threads running"
        time.sleep(TIME_AT_PEAK_QPS*60)
        print "Completed full time at peak qps, shutting down threads"
    except:
        print "Exception raised, shutting down threads"
quitevent.set()
time.sleep(3)
for t in threads:
    t.join(1.0)
print "Finished"
4

1 回答 1

0

不要传递threadprocto Thread，而是扩展类：

class Worker(Thread):
    def __init__(self, urls):
        super(Worker, self).__init__()
        self.urls = urls

    def run(self):
        for url in self.urls:
            self.fetch(url)

也就是说，除非您这样做是为了更好地了解线程以及负载测试在内部是如何工作的，否则我建议您改用Jmeter等成熟的测试框架。首先，您必须积累多年的经验。

于 2012-08-06T07:56:26.283 回答