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"进程退出"
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。