python+ffmpeg视频并发直播压力测试
通过python与ffmpeg结合使用,可生成进行视频点播、直播的压力测试脚本。可支持不同类型的视频流,比如rtmp或者hls形式。
通过如下方式执行脚本:pythonmultiRealPlay.py[rtmp|http][threadcounts][intervalTime]
[rtmp|http]:视频播放的不同形式
[threadcounts]:并发线程数
[intervalTime]:启动每个线程的间隔时间
代码:
#!/usr/bin/python #-*-coding:utf-8-*- ''' Createdon2015年7月22日 @author:LiBiao ''' importdatetime,time importthreading importsubprocess importos,base64 importsys importQueue queue=Queue.Queue() #需要手动配置的参数 #启动序号 SLEEP_TIME=0 #直播地址 FULL_ADDR={} #需要手动配置的参数 RTMP_ADDR='rtmp://192.168.1.208:1935/live/' HTTP_ADDR='http://192.168.1.208:80/live' liveID='100002750'#来自于万视无忧中创建的直播 urlKey='a1e5c680f7bfc85851de8ab2e63b0a33'#来自于万视无忧安全设置模块 liveResCode='71ac6c06d3'#直播源码 #生成MD5值 defgetMD5_Value(inputdata): try: importhashlib hash=hashlib.md5(inputdata.encode('utf-8')) exceptImportError: #forpython<<2.5 importmd5 hash=md5.new() returnhash.hexdigest() #直播地址组装 defbuild_live_addr(): t=time.strftime('%Y%m%d%H%M%S',time.localtime())[2:] data='%s#%s#%s'%(liveID,t,urlKey) secret=getMD5_Value(data) rtmp_addr='%s%s?liveID=%s&time=%s&secret=%s'%(RTMP_ADDR,liveResCode,liveID,t,secret) http_addr='%s/%s/playlist.m3u8?liveID=%s&time=%s&secret=%s'%(HTTP_ADDR,liveResCode,liveID,t,secret) FULL_ADDR['rtmp']=rtmp_addr FULL_ADDR['http']=http_addr returnFULL_ADDR #获取本机ip地址,用来产生区别于其他机器的数据 defget_local_ip(): try: ip=os.popen("ifconfig|grep'inetaddr'|awk'{print$2}'").read() ip=ip[ip.find(':')+1:ip.find('\n')] exceptException,e: printe returnip classVideo_To_Live(threading.Thread): def__init__(self,queue): threading.Thread.__init__(self) self.queue=queue defrun(self): liveAddr=self.queue.get() #printliveAddr try: printliveAddr subprocess.call('./ffmpeg-i\"%s\"-c:vcopy-c:acopy-bsf:aaac_adtstoasc-y-fflv-timeout4000/dev/null2>/dev/null'%liveAddr,stdout=subprocess.PIPE,shell=True) exceptExceptionase: wiriteLog('ERROR',str(e)) self.queue.task_done() if__name__=="__main__": time.sleep(SLEEP_TIME) parser=argparse.ArgumentParser(description="LivePlay") parser.add_argument('--liveType',action="store",dest="liveType",required=False) parser.add_argument('--pnum',action="store",dest="pnum",type=int,required=False) parser.add_argument('--itime',action="store",dest="itime",required=False) given_args=parser.parse_args() liveType=given_args.liveType threadNum=given_args.pnum intervalTime=given_args.itime print"%d个%s进程开始运行........"%(threadNum,Video_To_Live) foriinxrange(threadNum): videotolive=Video_To_Live(queue) videotolive.setDaemon(True) videotolive.start() foriinxrange(threadNum): ifliveTypein["http","rtmp"]: addr=build_live_addr() liveaddr=addr[liveType] queue.put(liveaddr) time.sleep(intervalTime) queue.join() print"进程退出"
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。