Python3 Interface Performance Test Example Code

First, look at the example code:

# -*- coding:utf-8 -*-


import requests
import datetime
import time
import threading

'''
allow_redirects = False Disable redirection, add in the request Parameter 
get Request with params Reference transmission 
post Request, data type form , use data Reference transmission 
post Request, data type form , use data Reference transmission 
post Request, data type json , json Reference transmission 
timeout: Request timeout, added in request Parameter 
nub = 10# Set the number of concurrent threads 
ResponseTime=float(result.elapsed.microseconds)/1000 # Get the response time, in units ms
ThinkTime = 0.5# Set thinking time 
AverageTime = "{:.3f}".format(float(sum(myrequest.times))/float(len(myrequest.times)))# Calculate the average value of the array and keep it 3 Decimal 
totaltime = float(hour)*60*60 + float(minute)*60 + float(second) # Calculate the total thinking time + Request time 
'''

class url_request:
    times = []
    error = []
    def weather_DC(self):
        myrequest=url_request()
        weatherinfo_search = 'https://restapi.amap.com/v3/weather/weatherInfo?parameters'
        params = {'key': 'cd1b11e80ffac05253196aa2a1233f25',
                  'city': 110101,
                  'extensions': 'base',
                  'output': 'JSON'}

        result = requests.get(url=weatherinfo_search, params=params)
        print(" Status code: ",result.status_code)
        print(" Return message: ",result.text)
        ResponseTime=float(result.elapsed.microseconds)/1000
        myrequest.times.append(ResponseTime)
        if result.status_code !=200 :
            myrequest.error.append("0")
if __name__=='__main__':
    myrequest=url_request()
    threads = []
    starttime = datetime.datetime.now()
    print(" Request start time: request start time %s" %starttime)
    nub = 10
    ThinkTime = 0.5
    for i in range(1, nub+1):
        t = threading.Thread(target=myrequest.weather_DC())
        threads.append(t)
    for t in threads:
        time.sleep(ThinkTime)
        print(" Number of threads: thread %s" %t)
        t.setDaemon(True)
        t.start()
        t.join()
    endtime = datetime.datetime.now()
    print(" Request end time: request end time %s" %endtime)
    time.sleep(3)
    AverageTime = "{:.3f}".format(float(sum(myrequest.times))/float(len(myrequest.times)))
    print(" Average response time: Average Response Time %s ms" %AverageTime)
    usetime = str(endtime - starttime)
    hour = usetime.split(':').pop(0)
    minute = usetime.split(':').pop(1)
    second = usetime.split(':').pop(2)
    totaltime = float(hour)*60*60 + float(minute)*60 + float(second)
    print(" Concurrent number: Concurrent processing %s" %nub)
    print("# Total time spent: use total time %s s" %(totaltime-float(nub*ThinkTime)))
    print(" Number of erroneous requests: fail request %s s" %myrequest.error.count("0"))

Instance extension:

Using ruquest to send requests and using multithreading to simulate concurrency

#!/user/bin/env python
#coding=utf-8
import requests
import datetime
import time
import threading

class url_request():
    times = []
    error = []
    def req(self,AppID,url):
        myreq=url_request()
        headers = {'User-Agent' : 'Mozilla/5.0 (Linux; Android 4.2.1; en-us; Nexus 4 Build/JOP40D) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.166 Mobile Safari/535.19'}
        payload = {'AppID':AppID,'CurrentURL':url}
        r = requests.post("http://xx.xxx.com/WeiXinJSAccessToken/json/WeChatJSTicket",headers=headers,data=payload)
        ResponseTime=float(r.elapsed.microseconds)/1000 # Get the response time, in units ms
        myreq.times.append(ResponseTime) # Write response time to array 
        if r.status_code !=200 :
            myreq.error.append("0")
if __name__=='__main__':
    myreq=url_request()
    threads = []
    starttime = datetime.datetime.now()
    print "request start time %s" %starttime 
    nub = 50# Set the number of concurrent threads 
    ThinkTime = 0.5# Set thinking time 
    for i in range(1, nub+1): 
        t = threading.Thread(target=myreq.req, args=('12','http://m.ctrip.com/webapp/cpage/#mypoints'))
        threads.append(t)
    for t in threads:
        time.sleep(ThinkTime) 
        #print "thread %s" %t # Print thread 
        t.setDaemon(True)
        t.start()
    t.join()
    endtime = datetime.datetime.now()
    print "request end time %s" %endtime  
    time.sleep(3)
    AverageTime = "{:.3f}".format(float(sum(myreq.times))/float(len(myreq.times))) # Calculate the average value of the array and keep it 3 Decimal 
    print "Average Response Time %s ms" %AverageTime # Average print response time 
    usetime = str(endtime - starttime)
    hour = usetime.split(':').pop(0)
    minute = usetime.split(':').pop(1)
    second = usetime.split(':').pop(2)
    totaltime = float(hour)*60*60 + float(minute)*60 + float(second) # Calculate the total thinking time + Request time 
    print "Concurrent processing %s" %nub # Print concurrent number 
    print "use total time %s s" %(totaltime-float(nub*ThinkTime)) # Total time spent printing 
    print "fail request %s" %myreq.error.count("0") # Number of print error requests 

request start time 2015-02-10 18:24:14.316000
request end time 2015-02-10 18:24:39.769000
Average Response Time 46.700 ms
Concurrent processing 50
use total time 25.453 s
fail request 1