创建线程
import threading
import _thread
import time
def job():
print('这是一个需要执行的任务')
#激活的线程个数
print('当前线程的个数:',threading.active_count())
#打印当前线程的详细信息
print('当前线程信息:',threading.current_thread())
time.sleep(10)
if __name__=='__main__':
job()
_thread模块创建多线程
import _thread
import threading
import time
def job(name):
print('这是一个需要执行的任务')
#激活的线程个数
print('激活的线程个数:',threading.active_count())
#当前线程的信息
print('当前线程信息:',threading.current_thread())
print(name)
time.sleep(10)
if __name__=='__main__':
#创建多个线程,但是没有开始执行任务
_thread.start_new_thread(job,('thread1',))
_thread.start_new_thread(job,('thread2',))
# while True:
job(job)
threading创建线程的方法
import _thread
import threading
import time
def job(name):
print('这是一个需要执行的任务:%s' %(name))
#激活的线程个数
print('激活的线程个数:',threading.active_count())
#线程信息
print('当前线程信息:',threading.current_thread())
time.sleep(5)
print(name,time.ctime())
if __name__=='__main__':
job('job1')
#创建多个线程
t1=threading.Thread(target=job,name='work1',args=('heiha',))
t1.start()
t2=threading.Thread(target=job,name='work2',args=('haha',))
t2.start()
print('hello')
不使用多线程及多线程的join方法
不使用
import time
def music(name):
for i in range(2):
print('正在听音乐%s' %(name))
time.sleep(3)
def code(name):
for i in range(2):
print('正在编写代码%s' %(name))
time.sleep(5)
if __name__=='__main__':
start_time=time.time()
music('that girl')
code('爬虫..')
print('花费时间:%s' %(time.time()-start_time))
使用join方法
import threading
import time
def music(name):
for i in range(2):
print('正在听音乐%s' %(name))
time.sleep(3)
def code(name):
for i in range(2):
print('正在编写代码%s' %(name))
time.sleep(5)
if __name__=='__main__':
start_time=time.time()
t1=threading.Thread(target=music,args=('that girl',))
t1.start()
t2=threading.Thread(target=code,args=('爬虫',))
# t1.start()
t2.start()
#等待所有的子线程执行结束之后,继续执行主线程的内容
t1.join()
t2.join()
print('花费时间:%s' %(time.time()-start_time))
_threading的set_daemon方法实现
#当主线程执行结束,让没有执行的线程强制结束;set_daemon
import threading
import time
#任务1
def music(name):
for i in range(2):
print('正在听音乐%s' %(name))
time.sleep(3)
#任务2
def code(name):
for i in range(2):
print('正在编写代码%s' %(name))
time.sleep(5)
if __name__=='__main__':
start_time=time.time()
t1=threading.Thread(target=music,args=('that girl',))
t2=threading.Thread(target=code,args=('爬虫',))
#将t1线程设定为守护线程,如果设置为True,子线程启动,当主线程执行结束,子线程也执行结束
#设置setDaemon必须在启动线程之前设置
t1.setDaemon(True)
t2.setDaemon(True)
t1.start()
t2.start()
t1.join()
t2.join()
print('花费时间:%s' %(time.time()-start_time))
join的详细理解
1. 管理线程的模块: _thread, threading
2. _thread创建多线程: _thread.start_new_thread(线程需要执行的任务,(任务需要的参数, 元组数据类型 ))
3. threading创建多线程第一种方式:实例化一个对象(Thread)
#t1 = threading.Thread(target=任务函数名, args=(x1,x2), name=‘threadName’)
4. 启动线程: t.start()
5. join方法: 会等待, 直到t1线程执行结束;阻塞正在调用的线程
6. setDaemon:
import threading
import time
def job():
time.sleep(2)
print('job1')
# t1=threading.Thread(target=job)
# t1.start()
# t1.join() #会等待,直到t1线程执行结束;阻塞正在调用的线程(t2)
# t2=threading.Thread(target=job)
# t2.start()
# t2.join()#会等待,直到t2线程执行结束,阻塞正在调用的线程
t1=threading.Thread(target=job)
t1.start()
t2=threading.Thread(target=job)
t2.start()
t1.join()
t2.join()
print('main thread end')
threading创建线程方法2_任务无需参数和需要传参数
无需参数
import threading
#类的继承
class IpThread(threading.Thread):
#重写构造方法
def __init__(self,jobname):
super(IpThread, self).__init__()
self.jobname=jobname
#将多线程需要执行的任务重写到run方法中
def run(self):
print('this is a job')
t1=IpThread(jobname='new job')
t1.start()
需要参数
import json
import threading
#类的继承
from urllib.error import HTTPError
from urllib.request import urlopen
import time
class IpThread(threading.Thread):
#重写构造方法;如果执行的任务需要传递参数;那将参数通过构造函数与self绑定
def __init__(self,jobname,ip):
super(IpThread, self).__init__()
self.jobname=jobname
self.ip=ip
#将多线程需要执行的任务重写到run方法中
def run(self):
try:
#需要一个参数,传ip
url="http://ip.taobao.com/service/getIpInfo.php?ip=%s" % (self.ip)
#根据url获取网页中的内容;并且解码为utf-8格式,识别中文
text=urlopen(url).read().decode('utf-8')
except HTTPError as e:
print('Error:%s获取地理位置网络错误' %(self.ip))
else:
#将获取的字符串类型转换为字典,方便处理
d=json.loads(text)['data']
country=d['country']
city=d['city']
print('%s:' %(self.ip),country,city)
def use_thread():
start_time=time.time()
ips = ['172.25.254.250', '8.8.8.8',
'172.25.254.250']
threads = []
for ip in ips:
t = IpThread(jobname="爬虫", ip=ip)
threads.append(t)
t.start()
# 等待所有的子线程执行结束
[thread.join() for thread in threads]
print("Success, 运行时间为%s" % (time.time() - start_time))
if __name__ == "__main__":
use_thread()
线程同步之线程锁
import threading
def add(lock):
#2.操作变量之前进行加锁
# lock.acquire()
global money
for i in range(1000000):
money+=1
#3.操作变量完成后进行解锁
# lock.release()
def reduce(lock):
# 2.操作变量之前进行加锁
# lock.acquire()
global money
for i in range(1000000):
money -= 1
# 3.操作变量完成后进行解锁
# lock.release()
if __name__=='__main__':
money=0
#1,实例化锁对象
lock=threading.Lock()
t1=threading.Thread(target=add,args=(lock,))
t2=threading.Thread(target=reduce,args=(lock,))
t1.start()
t2.start()
#等待所有子线程执行结束
t1.join()
t2.join()
print('最终金额为:%s' %(money))
队列与多线程
# 1). 理论上多线程执行任务, 会产生一些数据, 为其他程序执行作铺垫;
# 2). 多线程是不能返回任务执行结果的, 因此需要一个容器来存储多线程产生的数据
# 3). 这个容器如何选择? list(栈, 队列), tuple(x), set(x), dict(x), 此处选择队列来实现
import threading
from collections import Iterable
from mytimeit import timeit
from queue import Queue
def job(l,queue):
#将任务的结果存储到队列中
queue.put(sum(l))
@timeit
def use_thread():
#实例化一个队列,用来存储每个线程执行的结果
q=Queue()
#入队
q.put(1)
li = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6], [2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]]
threads=[]
for i in li:
t=threading.Thread(target=job,args=(i,q))
threads.append(t)
t.start()
#join方法等待所有子线程执行结束
[thread.join() for thread in threads]
#从队列里面拿出所有运行结果
result=[q.get() for _ in li]
print(result)
print(isinstance(q,Iterable))
if __name__=='__main__':
use_thread()
Threadpool线程池
# 注意: python3.2版本以后才可以使用;
from concurrent.futures import ThreadPoolExecutor
import time
#需要执行的任务
def job():
print('this is a job')
return 'hello'
if __name__=='__main__':
#实例化对象,线程池包含10个线程来处理任务
pool=ThreadPoolExecutor(max_workers=10)
#往线程池扔需要执行的任务,返回一个对象
f1=pool.submit(job)
f2=pool.submit(job)
#判断任务是否执行结束
print(f1.done())
time.sleep(3)
print(f2.done())
#获取任务执行的结果
print(f1.result())
print(f2.result())
线程池执行多个循环执行任务多次
from concurrent.futures import ThreadPoolExecutor
from urllib.request import urlopen
"""
Name:06_多线程paramiko执行命令.py
Author: lvah
Date: 2018-09-15
Email: [email protected]
Desc:
这是一个python脚本.
"""
# 基于ssh用于连接远程服务器做操作:远程执行命令, 上传文件, 下载文件
import threading
import paramiko
from paramiko.ssh_exception import NoValidConnectionsError, AuthenticationException
def connect(cmd, hostname, port=22, user='root'):
# ssh [email protected]
# 创建一个ssh对象;
client = paramiko.SSHClient()
# 返回一个私钥对象
private_key = paramiko.RSAKey.from_private_key_file('id_rsa')
# 2. 解决问题:如果之前没有;连接过的ip, 会出现
# Are you sure you want to continue connecting (yes/no)? yes
# 自动选择yes
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 3. 连接服务器
client.connect(hostname=hostname,
port=port,
username=user,
pkey=private_key
)
# 4. 执行操作
stdin, stdout, stderr = client.exec_command(cmd)
except NoValidConnectionsError as e:
print("%s连接失败" %(hostname))
except AuthenticationException as e:
print("%s密码错误" %(hostname))
else:
# 5. 获取命令的执行结果;
result = stdout.read().decode('utf-8')
print("%s运行结果:" %(hostname), result)
finally:
# 6. 关闭连接
client.close()
# ******************方法1: 每个ip使用一个线程来处理********************
# # 用来存储创建的所有线程对象;
# threads = []
# for count in range(254):
# host = '172.25.254.%s' %(count+1)
# # print(host.center(50, '*'))
# t = threading.Thread(target=connect, args=('uname', host))
# threads.append(t)
# t.start()
#
#
# # join方法, 等待所有的子线程执行结束;
# _ = [thread.join() for thread in threads]
#
# print("任务执行结束........")
# ***********************方法2: 线程池只有50个线程处理所有的任务****************
# 创建线程池对象
pool = ThreadPoolExecutor(max_workers=50)
# 依次向线程池提交任务
for count in range(254):
host = '172.25.254.%s' % (count + 1)
pool.submit(connect, 'uname', host)
线程池与map函数
from urllib.error import HTTPError
from urllib.request import urlopen
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
import time
URLS = ['http://httpbin.org', 'http://example.com/',
'https://api.github.com/'] * 3
def get_page(url, timeout=3):
try:
content = urlopen(url).read()
return {'url':url, 'len':len(content)}
except HTTPError as e:
return {'url':url, 'len':0}
# 方法1: submit提交任务
start_time = time.time()
pool = ThreadPoolExecutor(max_workers=20)
futuresObj = [pool.submit(get_page, url) for url in URLS]
# 注意: 传递的时包含futures对象的序列, as_complete, 返回已经执行完任务的future对象,
# 直到所有的future对应的任务执行完成, 循环结束;
# for finish_fs in as_completed(futuresObj):
# print(finish_fs.result() )
for future in futuresObj:
print(future.result())
print("执行时间:%s" %(time.time()-start_time))
# 方法2:通过map方式执行
# pool = ThreadPoolExecutor(max_workers=20)
# for res in pool.map(get_page, URLS):
# print(res)