线程基类
import threading
class base_thread(threading.Thread):
def __init__(self, func):
threading.Thread.__init__(self)
self.func = func
#print(type(self.func))
def run(self):
self.func
http请求
#-*- coding: utf-8 -*-
import requests
import json
class ConfigHttp:
def __init__(self, host, port,headers):
self.host = host
self.port = port
self.headers = headers
# 封装HTTP GET请求方法
def get(self, url, params):
# params = urllib.parse.urlencode(params)
url = "http://"+self.host+":"+self.port+url
try:
r = requests.get(url, params=params, headers=self.headers)
r.encoding = 'UTF-8'
return r.text
except Exception:
print('no json data returned')
return {}
# 封装HTTP POST请求方法,支持上传图片
def post(self, url, data, files=None):
data = eval(data)
url = 'http://' + self.host + ':' + str(self.port)+url
r =requests.post(url, files=files, data=data)
print(data)
json_response = r.text
return json_response
import ConfigHttp
def sample_request():
headers= {'Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8','User-Agent':'Mozilla/5.0 (Windows NT 6.1; rv:29.0) Gecko/20100101 Firefox/29.0'}
cf = ConfigHttp(host="www.xxx.com",port=80,headers=headers)
#cf.get(url="/api/getuserinfo",{"id":2})
cf.get(url="/api/getuserinfo?id=2")
def multi_thread():
count = 100
threads = []
for i in range(0, 100):
threads.append(base_thread(sample_request()))
for j in range(0, 100):
threads[j].start()
for k in range(0, 100):
threads[k].join()
2024 - 快车库 - 我的知识库 重庆启连科技有限公司 渝ICP备16002641号-10
企客连连 表单助手 企服开发 榜单123