Ppython爬取微博文字与图片(不使用Cookie) 网页版微博是纯正的HTML，而且调用的微博自家的API来获取图片。网址:https://m.weibo.cn/api/container/即为微博api里面包含了个人的信息与微博文字与图片存储地址。进入api页面我们可以很清晰的看到各种信息都用json存储起来了。我们再利用python中的json库提取出来即可。这比其它利用cookie模拟登陆要方便很多，我们只要输入被爬虫用户的微博ID然后运行便能自动爬取。ID从这个复制链接里面可以看出来。代码采用Python3# -*- coding: utf-8 -*- import urllib.request import json import requests import os path = 'C:\\Users\\dell\\Desktop\\weibo\\' id = '2093492691' proxy_addr = "122.241.72.191:808" pic_num = 0 weibo_name = "programmer" def use_proxy(url,proxy_addr): req=urllib.request.Request(url) req.add_header("User-Agent", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.221 Safari/537.36 SE 2.X MetaSr 1.0") proxy=urllib.request.ProxyHandler({'http':proxy_addr}) opener=urllib.request.build_opener(proxy,urllib.request.HTTPHandler) urllib.request.install_opener(opener) data=urllib.request.urlopen(req).read().decode('utf-8','ignore') return data def get_containerid(url): data=use_proxy(url,proxy_addr) content=json.loads(data).get('data') for data in content.get('tabsInfo').get('tabs'): if(data.get('tab_type')=='weibo'): containerid=data.get('containerid') return containerid def get_userInfo(id): url='https://m.weibo.cn/api/container/getIndex?type=uid&value='+id data=use_proxy(url,proxy_addr) content=json.loads(data).get('data') profile_image_url=content.get('userInfo').get('profile_image_url') description=content.get('userInfo').get('description') profile_url=content.get('userInfo').get('profile_url') verified=content.get('userInfo').get('verified') guanzhu=content.get('userInfo').get('follow_count') name=content.get('userInfo').get('screen_name') fensi=content.get('userInfo').get('followers_count') gender=content.get('userInfo').get('gender') urank=content.get('userInfo').get('urank') print("微博昵称："+name+"\n"+"微博主页地址："+profile_url+"\n"+"微博头像地址："+profile_image_url+"\n"+"是否认证："+str(verified)+"\n"+"微博说明："+description+"\n"+"关注人数："+str(guanzhu)+"\n"+"粉丝数："+str(fensi)+"\n"+"性别："+gender+"\n"+"微博等级："+str(urank)+"\n") def get_weibo(id,file): global pic_num i=1 while True: url='https://m.weibo.cn/api/container/getIndex?type=uid&value='+id weibo_url='https://m.weibo.cn/api/container/getIndex?type=uid&value='+id+'&containerid='+get_containerid(url)+'&page='+str(i) try: data=use_proxy(weibo_url,proxy_addr) content=json.loads(data).get('data') cards=content.get('cards') if(len(cards)>0): for j in range(len(cards)): print("-----正在爬取第"+str(i)+"页，第"+str(j)+"条微博------") card_type=cards[j].get('card_type') if(card_type==9): mblog=cards[j].get('mblog') attitudes_count=mblog.get('attitudes_count') comments_count=mblog.get('comments_count') created_at=mblog.get('created_at') reposts_count=mblog.get('reposts_count') scheme=cards[j].get('scheme') text=mblog.get('text') if mblog.get('pics') != None: # print(mblog.get('original_pic')) # print(mblog.get('pics')) pic_archive = mblog.get('pics') for _ in range(len(pic_archive)): pic_num += 1 print(pic_archive[_]['large']['url']) imgurl = pic_archive[_]['large']['url'] img = requests.get(imgurl) f = open(path+weibo_name+'\\'+str(pic_num)+ str(imgurl[-4:]), 'ab') # 存储图片，多媒体文件需要参数b（二进制文件） f.write(img.content) # 多媒体存储content f.close() with open(file,'a',encoding='utf-8') as fh: fh.write("----第"+str(i)+"页，第"+str(j)+"条微博----"+"\n") fh.write("微博地址："+str(scheme)+"\n"+"发布时间："+str(created_at)+"\n"+"微博内容："+text+"\n"+"点赞数："+str(attitudes_count)+"\n"+"评论数："+str(comments_count)+"\n"+"转发数："+str(reposts_count)+"\n") i+=1 else: break except Exception as e: print(e) pass if __name__ == "__main__": if os.path.isdir(path+weibo_name): pass else: os.mkdir(path+weibo_name) file = path+weibo_name+'\\'+weibo_name+".txt" get_userInfo(id) get_weibo(id, file)最终的效果图(爬取的微博txt文件):

Pygame详解：music 模块 pygame.mixer.musicPygame 中控制音频流的模块。函数pygame.mixer.music.load()  ——  载入一个音乐文件用于播放 pygame.mixer.music.play()  ——  开始播放音乐流 pygame.mixer.music.rewind()  ——  重新开始播放音乐 pygame.mixer.music.stop()  ——  结束音乐播放 pygame.mixer.music.pause()  ——  暂停音乐播放 pygame.mixer.music.unpause()  ——  恢复音乐播放 pygame.mixer.music.fadeout()  ——  淡出的效果结束音乐播放 pygame.mixer.music.set_volume()  ——  设置音量 pygame.mixer.music.get_volume()  ——  获取音量 pygame.mixer.music.get_busy()  ——  检查是否正在播放音乐 pygame.mixer.music.set_pos()  ——  设置播放的位置 pygame.mixer.music.get_pos()  ——  获取播放的位置 pygame.mixer.music.queue()  ——  将一个音乐文件放入队列中，并排在当前播放的音乐之后 pygame.mixer.music.set_endevent()  ——  当播放结束时发出一个事件 pygame.mixer.music.get_endevent()  ——  获取播放结束时发送的事件Pygame 中播放音乐的模块和 pygame.mixer 模块是密切联系的。使用音乐模块去控制在调音器上的音乐播放。音乐（music）播放和声音（sound）播放的不同之处在于音乐是流式的，并且绝对不会在一开始就把一个音乐文件全部载入。调音系统在工作刚开始时仅支持单音乐流。注意：对于 MP3 格式的支持是受限制的。在一些系统上，一种不受支持的格式将会是系统崩溃，例如 Debian Linux。为了游戏的稳定性，建议使用 OGG 进行替代。函数详解pygame.mixer.music.load()载入一个音乐文件用于播放。load(filename) -> Noneload(object) -> None该函数将会载入一个音乐文件名或者文件对象，并且准备播放。如果已经有音乐流正在播放，该音乐流将被停止。另外，函数不会开始播放音乐。pygame.mixer.music.play()开始播放音乐流。play(loops=0, start=0.0) -> None该函数用于播放已载入的音乐流。如果音乐已经开始播放，则将会重新开始播放。loops 参数控制重复播放的次数，例如 play(5) 意味着被载入的音乐将会立即开始播放 1 次并且再重复 5 次，共 6 次。如果 loops = -1，则表示无限重复播放。start 参数控制音乐从哪里开始播放。开始的位置取决于音乐的格式。MP3 和 OGG 使用时间表示播放位置（以秒为单位）。MOD使用模式顺序编号表示播放位置。如果音乐文件无法设置开始位置，则传递了start参数后会产生一个NotImplementedError 错误。pygame.mixer.music.rewind()重新开始播放音乐。rewind() -> None从文件开头开始重新播放音乐。pygame.mixer.music.stop()结束音乐播放。stop() -> None如果音乐正在播放则立即结束播放。pygame.mixer.music.pause()暂停音乐流的播放。pause() -> None通过调用 pygame.mixer.music.unpause() 函数继续播放音乐。pygame.mixer.music.unpause()恢复音乐播放。unpause() -> None在播放暂停后使用该函数可以继续音乐流的播放。pygame.mixer.music.fadeout()淡出的效果结束音乐播放。fadeout(time) -> None该函数将会在音乐淡出（也就是不在有声音放出）一段指定长度的时间（以毫秒为单位）后结束播放。注意：该函数在调用后会一直处于阻塞状态，直到音乐已经淡出。pygame.mixer.music.set_volume()设置音量。set_volume(value) -> None设置音乐的播放音量。value 参数值范围为 0.0~1.0。当新的音乐文件被载入，音量会被重置。pygame.mixer.music.get_volume()获取音量。get_volume() -> value返回正在播放的音乐的音量（此音量应该是调音器音量，注意与其他音量参数区分）。返回值范围为 0.0~1.0。pygame.mixer.music.get_busy()检查是否正在播放音乐。get_busy() -> bool如果有音乐流正在播放，此方法返回 True。否则返回 False。pygame.mixer.music.set_pos()设置播放的位置。set_pos(pos) -> None设置播放的起始位置。pos 参数是一个浮点数（或者一个可以转换为浮点数的数值），其值取决于音乐文件的格式：对于 MOD 文件，它是模块中的整型模式号；对于 OGG 文件，它是一个以音频开头为零点的绝对时间值（以秒为单位）；对于 MP3 文件，它是以当前播放位置为零点的绝对时间值（以秒为单位）。为了对一个 MP3 文件的进行绝对定位，建议首先调用 rewind() 函数（其他文件格式不受支持）。SDL_mixer 更新的版本提供了更好的定位支持。如果一种特殊的格式不支持定位，将会产生一个 SDLError 错误。该函数会调用 SDL_mixer 内的 Mix_SetMusicPosition() 函数。pygame.mixer.music.get_pos()获取播放的位置。get_pos() -> time此函数会获得音乐的播放时长（以毫秒为单数的数值）。返回值仅代表已经音乐已经播放了多久，并不考虑任何起始位置偏移量。pygame.mixer.music.queue()将一个音乐文件放入队列中，并排在当前播放的音乐之后。queue(filename) -> None此函数将会载入一个音乐文件并将其放入队列中。当前的音乐一旦播放完毕，正在排队的音乐文件就会开始播放。如果当前音乐被人为停止或者切换到其他音乐，则正在排队的音乐会被丢弃。下面的示例意思是先播放 6 次 Bach 然后再播放 1 次 Mozart：pygame.mixer.music.load('bach.ogg') pygame.mixer.music.play(5) # Plays six times, not five! pygame.mixer.music.queue('mozart.ogg')pygame.mixer.music.set_endevent()当播放结束时发出一个事件。set_endevent() -> Noneset_endevent(type) -> None调用此函数会使 Pygame 在音乐结束播放后发出信号（通过事件队列）。type 参数决定了什么样的事件将被放入事件队列中。任何时候音乐结束，都会放入指定事件到队列中（不仅仅是第一次）。调用该函数并不带任何参数，表示停止投放事件到队列中。pygame.mixer.music.get_endevent()获取播放结束时发送的事件。get_endevent() -> type返回音乐结束时被放入队列的事件类型。如果没有指定 endevent 事件，此方法会返回 pygame.NOEVENT 。

使用Python批量爬取美女图片 运行截图:相对于以前的版本此次增加自定义关键词实例代码:from bs4 import BeautifulSoup import requests,re,os headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36' } def Get_Url(url): response = requests.get(url, headers=headers).text soup = BeautifulSoup(response,'html.parser') for data in soup.find_all('a',class_='list-title text-md h-2x'): url_data = data.get('href') urls = "https://www.vmgirls.com/" + str(url_data) Down_Image(urls) print('-----------------------------------------------') print(urls) def Down_Image(url): response = requests.get(url, headers=headers).text soup = BeautifulSoup(response, 'html.parser') image_url = soup.find_all('img') for data in image_url: image_type = data.get('src').split('.')[-1] if image_type == 'jpg' or image_type == 'jpeg' or image_type == 'png': url_data = data.get('src') # print(url_data) dir_name = soup.find(class_='post-title h1').string if not os.path.exists(dir_name): os.mkdir(dir_name) # print(dir_name) image = requests.get("https:" + str(url_data), headers=headers).content file_name = url_data.split('/')[-1] # print(file_name) with open(dir_name + '/' + file_name, 'wb') as f: f.write(image) print('正在写入----->' + dir_name + '/' + file_name) if __name__ == '__main__': print(' ---------------------------------------------------------------------') print('| |') print('| Author:culprit --- iamdd |') print('| |') print(' ---------------------------------------------------------------------') print('支持搜索关键词(例如少女、小姐姐、青春。。。)') key = input('请输入关键词(默认回车不输入)：') if key == '': url = "https://www.vmgirls.com/" else: url = "https://www.vmgirls.com/?s=" + key Get_Url(url)

导入pygame时出现的问题:显示版本号和欢迎界面 导入pygame时出现的问题，见下方代码>>> import pygame pygame 1.9.4 Hello from the pygame community. https://www.pygame.org/contribute.html不是报错, 仅仅只是欢迎而已, 如果你不想看到欢迎语句, 以下会教你如何删去这两句话:找到你Python的安装路径, 按照下图的路径, 找到这个__init__.py文件:以任意文本编辑器(此处是notepad++)打开找到最后两行, 注释掉, 按ctrl+s即可:再次导入，就没有欢迎语句了:彩蛋！！放上pygame中__init__.py中的结语:Thanks for supporting pygame. Without support now, there won't be pygame later.

Python多线程并发实例及其优化 这篇文章主要介绍了python多线程并发实例及其优化,threading是扩展模块，在thread的基础上进行了封装及改进。所以只需要使用threading这个模块就能完成并发的测试,需要的朋友可以参考下单线程执行python的内置模块提供了两个内置模块：thread和threading，thread是源生模块，threading是扩展模块，在thread的基础上进行了封装及改进。所以只需要使用threading这个模块就能完成并发的测试实例创建并启动一个单线程import threading def myTestFunc(): print("我是一个函数") t = threading.Thread(target=myTestFunc) # 创建一个线程 t.start() # 启动线程执行结果C:\Python36\python.exe D:/MyThreading/myThread.py 我是一个线程函数 Process finished with exit code 0其实单线程的执行结果和单独执行某一个或者某一组函数结果是一样的，区别只在于用线程的方式执行函数，而线程是可以同时执行多个的，函数是不可以同时执行的。多线程执行上面介绍了单线程如何使用，多线程只需要通过循环创建多个线程，并循环启动线程执行就可以了实例import threading from datetime import datetime def thread_func(): # 线程函数 print('我是一个线程函数', datetime.now()) def many_thread(): threads = [] for _ in range(10): # 循环创建10个线程 t = threading.Thread(target=thread_func) threads.append(t) for t in threads: # 循环启动10个线程 t.start() if __name__ == '__main__': many_thread()执行结果C:\Python36\python.exe D:/MyThreading/manythread.py 我是一个线程函数 2019-06-23 16:54:58.205146 我是一个线程函数 2019-06-23 16:54:58.205146 我是一个线程函数 2019-06-23 16:54:58.206159 我是一个线程函数 2019-06-23 16:54:58.206159 我是一个线程函数 2019-06-23 16:54:58.206159 我是一个线程函数 2019-06-23 16:54:58.207139 我是一个线程函数 2019-06-23 16:54:58.207139 我是一个线程函数 2019-06-23 16:54:58.207139 我是一个线程函数 2019-06-23 16:54:58.208150 我是一个线程函数 2019-06-23 16:54:58.208150 Process finished with exit code 0通过循环创建10个线程，并且执行了10次线程函数，但需要注意的是python的并发并非绝对意义上的同时处理，因为启动线程是通过循环启动的，还是有先后顺序的，通过执行结果的时间可以看出还是有细微的差异，但可以忽略不记。当然如果线程过多就会扩大这种差异。我们启动500个线程看下程序执行时间实例import threading from datetime import datetime def thread_func(): # 线程函数 print('我是一个线程函数', datetime.now()) def many_thread(): threads = [] for _ in range(500): # 循环创建500个线程 t = threading.Thread(target=thread_func) threads.append(t) for t in threads: # 循环启动500个线程 t.start() if __name__ == '__main__': start = datetime.today().now() many_thread() duration = datetime.today().now() - start print(duration)执行结果0:00:00.111657 Process finished with exit code 0500个线程共执行了大约0.11秒那么针对这种问题我们该如何优化呢？我们可以创建25个线程，每个线程执行20次线程函数，这样在启动下一个线程的时候，上一个线程已经在循环执行了，这样就大大减少了并发的时间差异优化import threading from datetime import datetime def thread_func(): # 线程函数 print('我是一个线程函数', datetime.now()) def execute_func(): for _ in range(20): thread_func() def many_thread(): start = datetime.now() threads = [] for _ in range(25): # 循环创建500个线程 t = threading.Thread(target=execute_func) threads.append(t) for t in threads: # 循环启动500个线程 t.start() duration = datetime.now() - start print(duration) if __name__ == '__main__': many_thread()输出结果(仅看程序执行间隔)0:00:00.014959 Process finished with exit code 0后面的优化执行500次并发一共花了0.014秒。比未优化前的500个并发快了几倍，如果线程函数的执行时间比较长的话，那么这个差异会更加显著，所以大量的并发测试建议使用后者，后者比较接近同时“并发”守护线程多线程还有一个重要概念就是守护线程。那么在这之前我们需要知道主线程和子线程的区别，之前创建的线程其实都是main()线程的子线程，即先启动主线程main()，然后执行线程函数子线程。那么什么是守护线程？即当主线程执行完毕之后，所有的子线程也被关闭(无论子线程是否执行完成)。默认不设置的情况下是没有守护线程的，主线程执行完毕后，会等待子线程全部执行完毕，才会关闭结束程序。但是这样会有一个弊端，当子线程死循环了或者一直处于等待之中，则程序将不会被关闭，被被无限挂起，我们把上述的线程函数改成循环10次, 并睡眠2秒，这样效果会更明显import threading from datetime import datetime import time def thread_func(): # 线程函数 　　 time.sleep(2) i = 0 while(i < 11): print(datetime.now()) i += 1 def many_thread(): threads = [] for _ in range(10): # 循环创建500个线程 t = threading.Thread(target=thread_func) threads.append(t) for t in threads: # 循环启动500个线程 t.start() if __name__ == '__main__': many_thread() print("thread end")执行结果C:\Python36\python.exe D:/MyThreading/manythread.py thread end 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.468612 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.469559 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.470556 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.471554 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.472557 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.473548 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.474545 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.475552 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.476548 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 2019-06-23 19:08:00.477546 Process finished with exit code 0根据上述结果可以看到主线程打印了“thread end”之后（主线程结束），子线程还在继续执行，并未随着主线程的结束而结束下面我们通过 setDaemon方法给子线程添加守护线程，我们把循环改为死循环，再来看看输出结果(注意守护线程要加在start之前)import threading from datetime import datetime def thread_func(): # 线程函数 i = 0 while(1): print(datetime.now()) i += 1 def many_thread(): threads = [] for _ in range(10): # 循环创建500个线程 t = threading.Thread(target=thread_func) threads.append(t) t.setDaemon(True) # 给每个子线程添加守护线程 for t in threads: # 循环启动500个线程 t.start() if __name__ == '__main__': many_thread() print("thread end")输出结果2019-06-23 19:12:35.564539 2019-06-23 19:12:35.564539 2019-06-23 19:12:35.564539 2019-06-23 19:12:35.564539 2019-06-23 19:12:35.564539 2019-06-23 19:12:35.564539 2019-06-23 19:12:35.565529 2019-06-23 19:12:35.565529 2019-06-23 19:12:35.565529 thread end Process finished with exit code 0通过结果我们可以发现，主线程关闭之后子线程也会随着关闭，并没有无限的循环下去，这就像程序执行到一半强制关闭执行一样，看似暴力却很有用，如果子线程发送一个请求未收到请求结果，那不可能永远等下去，这时候就需要强制关闭。所以守护线程解决了主线程和子线程关闭的问题。阻塞线程上面说了守护线程的作用，那么有没有别的方法来解决上述问题呢？ 其实是有的，那就是阻塞线程，这种方式更加合理，使用join()方法阻塞线程，让主线程等待子线程执行完成之后再往下执行，再关闭所有子线程，而不是只要主线程结束，不管子线程是否执行完成都终止子线程执行。下面我们给子线程添加上join()（主要join要加到start之后）import threading from datetime import datetime import time def thread_func(): # 线程函数 time.sleep(1) i = 0 while(i < 11): print(datetime.now()) i += 1 def many_thread(): threads = [] for _ in range(10): # 循环创建500个线程 t = threading.Thread(target=thread_func) threads.append(t) t.setDaemon(True) # 给每个子线程添加守护线程 for t in threads: # 循环启动500个线程 t.start() for t in threads: t.join() # 阻塞线程 if __name__ == '__main__': many_thread() print("thread end")执行结果程序会一直执行，但是不会打印“thread end”语句，因为子线程并未结束，那么主线程就会一直等待。疑问：有人会觉得这和什么都不设置是一样的，其实会有一点区别的，从守护线程和线程阻塞的定义就可以看出来，如果什么都没设置，那么主线程会先执行完毕打印后面的“thread end”，而等待子线程执行完毕。两个都设置了，那么主线程会等待子线程执行结束再继续执行。而对于死循环或者一直等待的情况，我们可以给join设置超时等待，我们设置join的参数为2，那么子线程会告诉主线程让其等待2秒，如果2秒内子线程执行结束主线程就继续往下执行，如果2秒内子线程未结束，主线程也会继续往下执行，执行完成后关闭子线程输出结果import threading from datetime import datetime import time def thread_func(): # 线程函数 time.sleep(1) i = 0 while(1): print(datetime.now()) i += 1 def many_thread(): threads = [] for _ in range(10): # 循环创建500个线程 t = threading.Thread(target=thread_func) threads.append(t) t.setDaemon(True) # 给每个子线程添加守护线程 for t in threads: # 循环启动500个线程 t.start() for t in threads: t.join(2) # 设置子线程超时2秒 if __name__ == '__main__': many_thread() print("thread end")你运行程序后会发现，运行了大概2秒的时候，程序会数据“thread end” 然后结束程序执行， 这就是阻塞线程的意义，控制子线程和主线程的执行顺序总结最好呢，再次说一下守护线程和阻塞线程的定义1.守护线程：子线程会随着主线程的结束而结束，无论子线程是否执行完毕2.阻塞线程：主线程会等待子线程的执行结束，才继续执行以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

Python压力测试脚本CPU #!/usr/bin/env python3 import time import urllib.request import threading from time import sleep # 性能测试页面 PERF_TEST_URL = "目标域名" # 配置:模拟运行状态 THREAD_NUM = 1200 # 并发线程总数 ONE_WORKER_NUM = 100 # 每个线程的循环次数 LOOP_SLEEP = 0.1 # 每次请求时间间隔(秒) # 出错数 ERROR_NUM = 0 #具体的处理函数，负责处理单个任务 def doWork(index): t = threading.currentThread() try: html = urllib.request.urlopen(PERF_TEST_URL).read() except Exception as e: print(e) global ERROR_NUM ERROR_NUM += 1 def working(): t = threading.currentThread() # print( "["+t.name+"] Sub Thread Begin" ) i = 0 while i < ONE_WORKER_NUM: i += 1 doWork(i) sleep(LOOP_SLEEP) # print( "["+t.name+"] Sub Thread End" ) def main(): t1 = time.time() Threads = [] # 创建线程 for i in range(THREAD_NUM): t = threading.Thread(target=working, name="T"+str(i)) t.setDaemon(True) Threads.append(t) for t in Threads: t.start() for t in Threads: t.join() print( "main thread end" ) t2 = time.time() print( "========================================" ) print( "URL:", PERF_TEST_URL ) print( "任务数量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM*ONE_WORKER_NUM ) print( "总耗时(秒):", t2-t1 ) print( "每次请求耗时(秒):", (t2-t1) / (THREAD_NUM*ONE_WORKER_NUM) ) print( "每秒承载请求数:", 1 / ((t2-t1) / (THREAD_NUM*ONE_WORKER_NUM)) ) print( "错误数量:", ERROR_NUM ) if __name__ == "__main__": main()

Python接口压力测试 1、单进程多线程模式# #!/usr/bin/env python # # -*- coding:utf-8 -*- import time import logging import requests import threading from concurrent import futures # download_url = 'http://192.168.188.110:8081//workspace/record_download/polls/82003533467_18b305da-e313-11e8-aa39-00163e0a6bde.mp3' # download_url = 'http://192.168.188.110:8081//workspace/record_download/polls/test.log' download_url = 'http://192.168.188.110:8081//workspace/record_download/polls/9921_057128214999_18210532807_20181113110420_00163e104dbfbb8b11e8e6f0d0990876(3).wav' workers = 1000 mutex = threading.Lock() session = requests.Session() contain = {'average_cost':0,'min_cost':0,'max_cost':0,'hit_count':0} def handle(cost): with mutex: min_cost = contain['min_cost'] max_cost = contain['max_cost'] hit_count = contain['hit_count'] average_cost = contain['average_cost'] if min_cost == 0: contain['min_cost'] = cost if min_cost > cost: contain['min_cost'] = cost if max_cost < cost: contain['max_cost'] = cost average_cost = (average_cost*hit_count + cost) / (hit_count + 1) hit_count +=1 contain['average_cost'] = average_cost contain['hit_count'] = hit_count logging.info(contain) def download_one(): while True: try: stime = time.time() request = requests.Request(method='GET', url=download_url,) prep = session.prepare_request(request) response = session.send(prep, timeout=100) etime = time.time() # print(response.content) logging.info('thread[%s] status[%s] cost[%s]',threading.current_thread().ident, response.status_code,etime-stime) handle(float(etime-stime)) except Exception as e: logging.error(e) print(e) def main(): with futures.ThreadPoolExecutor(workers) as executor: for i in range(workers): executor.submit(download_one) if __name__ == '__main__': logging.basicConfig(filename="client.log", level=logging.INFO, format="%(asctime)s [%(filename)s:%(lineno)d] %(message)s", datefmt="%m/%d/%Y %H:%M:%S [%A]") main()2、多进程多线程模式# #!/usr/bin/env python # # -*- coding:utf-8 -*- import os import time import logging import requests import threading from multiprocessing import Lock,Manager from concurrent import futures download_url = 'http://192.168.188.105:8888' workers = 250 cpu_count = 4 session = requests.Session() def handle(cost,mutex,contain): with mutex: min_cost = contain['min_cost'] max_cost = contain['max_cost'] hit_count = contain['hit_count'] average_cost = contain['average_cost'] if min_cost == 0: contain['min_cost'] = cost if min_cost > cost: contain['min_cost'] = cost if max_cost < cost: contain['max_cost'] = cost average_cost = (average_cost*hit_count + cost) / (hit_count + 1) hit_count +=1 contain['average_cost'] = average_cost contain['hit_count'] = hit_count logging.info(contain) def download_one(mutex,contain): while True: try: stime = time.time() request = requests.Request(method='GET', url=download_url,) prep = session.prepare_request(request) response = session.send(prep, timeout=50) etime = time.time() print(response.status_code) logging.info('process[%s] thread[%s] status[%s] cost[%s]',os.getpid(),threading.current_thread().ident, response.status_code,etime-stime) handle(float(etime-stime),mutex,contain) # time.sleep(1) except Exception as e: logging.error(e) print(e) def new_thread_pool(mutex,contain): with futures.ThreadPoolExecutor(workers) as executor: for i in range(workers): executor.submit(download_one,mutex,contain) def subprocess(): manager = Manager() mutex = manager.Lock() contain = manager.dict({'average_cost': 0, 'min_cost': 0, 'max_cost': 0, 'hit_count': 0}) with futures.ProcessPoolExecutor(cpu_count) as executor: for i in range(cpu_count): executor.submit(new_thread_pool,mutex,contain) if __name__ == '__main__': logging.basicConfig(filename="client.log", level=logging.INFO, format="%(asctime)s [%(filename)s:%(lineno)d] %(message)s", datefmt="%m/%d/%Y %H:%M:%S [%A]") subprocess()

python打包成exe可执行文件教程 本文，介绍一下把python代码打包成windows的可执行文件的方法。步骤1安装pywin32，可以参考《怎么给python安装pywin32模块？》一定要注意对应的python版本，否则不能安装。步骤2用命令行调用pip安装pyinstaller。步骤3准备一个py文件。我这里准备的是一个爬取电子书的python文件——00.py。步骤4准备一个图片，作为exe的图标：b.ico注意图片格式是ico的，大小为32*32，不要太大。没有的话，可以在网上免费转格式。把00.py和b.ico放到一个目录里面——C:\a步骤5把cmd的当前目录切换到C:\a。步骤6然后执行命令行：pyinstaller -F -i b.ico 00.py注意:文件名后面不要有后缀，直接00和b步骤7运行之后，cmd变成了这样。步骤8在C:\a目录里面，多出了几个文件夹。可执行文件00.exe就在dist文件夹里面。步骤9双击可执行文件，会弹出一个命令提示符端口，提示你输入书号。按照格式输入书号，点击回车键，就可以在dist文件夹里面，看到一章章的电子书被下载下来了。exe的图标，一定要是ico格式。

Python学习案例之视频人脸检测识别 前言与大家分享了简单的图片人脸识别技术，其实在实际应用中，很多是通过视频流的方式进行识别，比如人脸识别通道门禁考勤系统、人脸动态跟踪识别系统等等。案例这里我们还是使用 opencv 中自带了 haar人脸特征分类器，通过读取一段视频来识别其中的人脸。代码实现：# -*- coding: utf-8 -*- __author__ = "苏画" __blog__ = "https://qemao.com" import cv2 import os # 保存好的视频检测人脸并截图 def CatchPICFromVideo(window_name, camera_idx, catch_pic_num, path_name): cv2.namedWindow(window_name) # 视频来源 cap = cv2.VideoCapture(camera_idx) # 告诉OpenCV使用人脸识别分类器 classfier = cv2.CascadeClassifier(os.getcwd()+"\\haarcascade\\haarcascade_frontalface_alt.xml") # 识别出人脸后要画的边框的颜色，RGB格式, color是一个不可增删的数组 color = (0, 255, 0) num = 0 while cap.isOpened(): ok, frame = cap.read() # 读取一帧数据 if not ok: break grey = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) # 将当前桢图像转换成灰度图像 # 人脸检测，1.2和2分别为图片缩放比例和需要检测的有效点数 faceRects = classfier.detectMultiScale(grey, scaleFactor=1.2, minNeighbors=3, minSize=(32, 32)) if len(faceRects) > 0: # 大于0则检测到人脸 for faceRect in faceRects: # 单独框出每一张人脸 x, y, w, h = faceRect # 将当前帧保存为图片 img_name = "%s/%d.jpg" % (path_name, num) # print(img_name) image = frame[y - 10: y + h + 10, x - 10: x + w + 10] cv2.imwrite(img_name, image, [int(cv2.IMWRITE_PNG_COMPRESSION), 9]) num += 1 if num > (catch_pic_num): # 如果超过指定最大保存数量退出循环 break # 画出矩形框 cv2.rectangle(frame, (x - 10, y - 10), (x + w + 10, y + h + 10), color, 2) # 显示当前捕捉到了多少人脸图片了，这样站在那里被拍摄时心里有个数，不用两眼一抹黑傻等着 font = cv2.FONT_HERSHEY_SIMPLEX cv2.putText(frame, 'num:%d/100' % (num), (x + 30, y + 30), font, 1, (255, 0, 255), 4) # 超过指定最大保存数量结束程序 if num > (catch_pic_num): break # 显示图像 cv2.imshow(window_name, frame) c = cv2.waitKey(10) if c & 0xFF == ord('q'): break # 释放摄像头并销毁所有窗口 cap.release() cv2.destroyAllWindows() if __name__ == '__main__': # 连续截100张图像 CatchPICFromVideo("get face", os.getcwd()+"\\video\\kelake.mp4", 100, "E:\\VideoCapture") 动图有点花，将就着看吧：如果是捕捉摄像头，只需要改变以下代码即可：如果获取摄像头，参数修改为 0 即可 cap = cv2.VideoCapture(0)源码{abtn icon="" color="#3666fa" href="http://sh.qemao.com/tools/tol/tu/_filew/store/Python%E4%BA%BA%E8%84%B8%E6%A3%80%E6%B5%8B%E8%AF%86%E5%88%AB.zip" radius="5px" content="点击下载"/}

利用python实现人脸识别 分享几个较详细的人脸识别教程，因为这几个教程都写的挺详细了，这里我就不在重复了。直接上网站基于Python3.7和opencv的人脸识别（含数据收集，模型训练）：https://www.cnblogs.com/xp12345/p/9818435.html视频人脸检测——OpenCV版：https://www.cnblogs.com/vipstone/p/8933916.html用树莓派实现实时的人脸检测：https://shumeipai.nxez.com/2018/03/09/real-time-face-recognition-an-end-to-end-project-with-raspberry-pi.html