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

にを渡す代わりにthreadproc、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 に答える