在线程中运行类方法(python)

我目前正在学习 Python 和类，我有一个基本的问题，但我没有找到任何答案。假设我有一门虚拟课程

class DomainOperations:
def __init__(self, domain):
self.domain = domain
self.domain_ip = ''
self.website_thumbnail = ''


def resolve_domain(self):
#resolve domain to ipv4 and save to self.domain_ip


def generate_website_thumbnail(self):
#generate website thumbnail and save the url to self.website_thumbnail

我想运行同时解决 _ 域和生成 _ 网站 _ 缩略图和线程完成后，我想打印的 IP 和缩略图。

编辑: 我知道我应该使用线程，也许像这样的东西

r = DomainOperations('google.com')


t1 = threading.Thread(target=r.resolve_domain)
t1.start()


t2 = threading.Thread(target=r.generate_website_thumbnail)
t2.start()

但是我应该在类之外使用它们吗? 我应该编写另一个类来处理线程吗？

正确的方式是怎么做到的？

129474 次浏览
小开
最佳答案

如果您从类中调用它们，其简单程度如下:

import threading


class DomainOperations:


def __init__(self):
self.domain_ip = ''
self.website_thumbnail = ''


def resolve_domain(self):
self.domain_ip = 'foo'


def generate_website_thumbnail(self):
self.website_thumbnail= 'bar'


def run(self):
t1 = threading.Thread(target=self.resolve_domain)
t2 = threading.Thread(target=self.generate_website_thumbnail)
t1.start()
t2.start()
t1.join()
t2.join()
print(self.domain_ip, self.website_thumbnail)


if __name__ == '__main__':
d = DomainOperations()
d.run()
小开

您可以在 DomainOperation 中继承 Thread 类，这样代码将更加干净和容易理解。必须重写 Run ()方法。

from threading import Thread


class DomainOperations(Thread):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.domain_ip = ''
self.website_thumbnail = ''


def resolve_domain(self):
self.domain_ip = 'foo'


def generate_website_thumbnail(self):
self.website_thumbnail= 'bar'


def run(self):
#domain will be resolved on first thread
self.resolve_domain()
#thumbnail will be resolved on second OR newly created below thread
thread2 = Thread(target=self.generate_website_thumbnail)
thread.start()
# thread1 will wait for thread2
self.join()
# thread2 will wait for thread1, if it's late.
thread2.join()
# here it will print ip and thumbnail before exiting first thread
print(self.domain_ip, self.website_thumbnail)

这样开始你的线程。

if __name__ == '__main__':
thread1 = DomainOperations()
thread1.start()
小开 
def post_test(tbid, line_num, response_time):
"""
:param tbid: 参数id
:return:
"""


# 请求参数
data = {'tbId': tbid, 'conditions': [{"key": "", "type": 1}], 'pageNum': 1, 'pageSize': 12}
# 请求启动时间


start = time.time()
# post请求
r = requests.post(url=url, data=json.dumps(data), headers=headers)
# 请求结束时间
end = time.time()
# 保留两位小数
finall_time = float('%.2f' % float(end - start))
text = json.loads(r.text)
# IO写入 只写入200的
with open('text6.csv', 'a', newline='') as csvfile:
if text['statusCode'] == '200':
throughput = line_num * response_time / finall_time
throughput = float('%.2f' % float(throughput))
print('the perf_counter time of %s is %s and the content is %s ,throughput is %s' % (
tbid, finall_time, json.loads(r.text), throughput))
spamwriter = csv.writer(csvfile, dialect='excel')
spamwriter.writerow([tbid] + [finall_time] + [throughput])
def start_thread(csv_name):
tbid, response_time_sort, throughput_sort = read_csv(csv_name)
print(tbid)
line_num = len(tbid)
response_times = 5


for j in range(response_times):
for i in tbid:
t = threading.Thread(target=post_test, args=(i, line_num, response_times))
t.start()
t.join()

我不知道如何在类中调用方法，特别是如果它有初始化参数的话，但是您可以尝试使用这个方法。我试图用多个过程来解决这个问题。


提交答案