python+ffmpeg视频并发直播压力测试

通过python与ffmpeg结合使用，可生成进行视频点播、直播的压力测试脚本。可支持不同类型的视频流，比如rtmp或者hls形式。
通过如下方式执行脚本：python multiRealPlay.py [rtmp|http] [thread counts] [interval Time]
[rtmp | http]：视频播放的不同形式
[thread counts]：并发线程数
[interval Time]：启动每个线程的间隔时间

代码：

#!/usr/bin/python
# -*- coding: utf-8 -*-
'''
Created on 2015年7月22日

@author: LiBiao
'''
import datetime,time
import threading
import subprocess
import os, base64
import sys
import Queue

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值
def getMD5_Value(inputdata):
  try:
    import hashlib
    hash = hashlib.md5(inputdata.encode('utf-8'))
  except ImportError:
    #for python << 2.5
    import md5
    hash = md5.new()

  return hash.hexdigest()

#直播地址组装
def build_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
  return FULL_ADDR

#获取本机ip地址，用来产生区别于其他机器的数据
def get_local_ip():
  try:
    ip = os.popen("ifconfig | grep 'inet addr' | awk '{print $2}'").read()
    ip = ip[ip.find(':') + 1:ip.find('\n')]
  except Exception,e:
    print e
  return ip

class Video_To_Live(threading.Thread):
  def __init__(self,queue):
    threading.Thread.__init__(self)
    self.queue = queue

  def run(self):
    liveAddr = self.queue.get()
    #print liveAddr
    try:
      print liveAddr
      subprocess.call('./ffmpeg -i \"%s\" -c:v copy -c:a copy -bsf:a aac_adtstoasc -y -f flv -timeout 4000 /dev/null 2>/dev/null' %liveAddr,stdout=subprocess.PIPE,shell=True)
    except Exception as e:
      wiriteLog('ERROR',str(e))
    self.queue.task_done()

if __name__ == "__main__":
  time.sleep(SLEEP_TIME)
  parser = argparse.ArgumentParser(description = "Live Play")
  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)
  for i in xrange(threadNum):
    videotolive = Video_To_Live(queue)
    videotolive.setDaemon(True)
    videotolive.start()

  for i in xrange(threadNum):
    if liveType in ["http","rtmp"]:
        addr = build_live_addr()
      liveaddr = addr[liveType]
    queue.put(liveaddr)
    time.sleep(intervalTime)
  queue.join()
  print "进程退出"

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我们。

您可能感兴趣的文章:

  • python调用系统ffmpeg实现视频截图、http发送
  • Python应用03 使用PyQT制作视频播放器实例
  • Python实现视频下载功能
  • Python实现Youku视频批量下载功能
  • Python OpenCV获取视频的方法
  • python爬取m3u8连接的视频
赞 (0)

相关推荐

  • Python OpenCV获取视频的方法

    之前有文章,使用Android平台的OpenCV接入了视频,控制的目标是手机的摄像头,这是OpenCV的好处,使用OpenCV可以使用跨平台的接口实现相同的功能,减少了平台间移植的困难.正如本文后面,将使用类似的接口,从笔记本的摄像头获取视频,所以,尝试本文代码需要有一台有摄像头的电脑. 不过,需要说明的的是,OpenCV的强项在于图像相关的处理,而不是视频的编解码,所以,不要使用OpenCV做多余的事情,我们使用OpenCV接入视频或者图片的目的,是为了对视频或图片进行处理. 关于Python

  • Python实现视频下载功能

    Python实现视频下载功能

    最近一两年短视频业务风生水起,各个视频网站都有各自特色的短视频内容.如果有这样一个程序,可以把各大视频网站的热门用户最新发布的视频都下载下来,不仅方便自己观看,还可以将没有版权的视频发布在个人社交网站上,增加自己的人气,岂不美哉? parker就是这样一个项目(项目地址:https://github.com/LiuRoy/parker),它采用celery框架定时爬取用户视频列表,将最新发布的视频通过you-get异步下载,可以很方便地实现分布式部署.因为各个网站的页面布局和接口更新比较频繁,为

  • python调用系统ffmpeg实现视频截图、http发送

    python 调用系统ffmpeg进行视频截图,并进行图片http发送ffmpeg ,视频.图片的各种处理. 最近在做视频.图片的版权等深度学习识别,用到了ffmpeg部分功能,功能如下:  调用ffmpeg 对不同目录视频进行截图,通过http发送到后台进行算法识别.  每5分钟扫描最近的视频,生成图片,发送完毕图片删除. 代码如下: #!/usr/bin/env python # -*- coding: utf-8 -*- """'定时任务每五分钟发送上一个5分钟视频 目

  • Python应用03 使用PyQT制作视频播放器实例

    Python应用03 使用PyQT制作视频播放器实例

    最近研究了Python的两个GUI包,Tkinter和PyQT.这两个GUI包的底层分别是Tcl/Tk和QT.相比之下,我觉得PyQT使用起来更加方便,功能也相对丰富.这一篇用PyQT实现一个视频播放器,并借此来说明PyQT的基本用法.  视频播放器 先把已经完成的代码放出来.代码基于Python 3.5: import time import sys from PyQt4 import QtGui, QtCore from PyQt4.phonon import Phonon class Po

  • python爬取m3u8连接的视频

    本文为大家分享了python爬取m3u8连接的视频方法,供大家参考,具体内容如下 要求:输入m3u8所在url,且ts视频与其在同一路径下 #!/usr/bin/env/python #_*_coding:utf-8_*_ #Data:17-10-08 #Auther:苏莫 #Link:http://blog.csdn.net/lingluofengzang #PythonVersion:python2.7 #filename:download_movie.py import os import

  • Python实现Youku视频批量下载功能

    Python实现Youku视频批量下载功能

    前段时间由于收集视频数据的需要,自己捣鼓了一个YouKu视频批量下载的程序.东西虽然简单,但还挺实用的,拿出来分享给大家. 版本:Python2.7+BeautifulSoup3.2.1 import urllib,urllib2,sys,os from BeautifulSoup import BeautifulSoup import itertools,re url_i =1 pic_num = 1 #自己定义的引号格式转换函数 def _en_to_cn(str): obj = itert

  • python+ffmpeg视频并发直播压力测试

    通过python与ffmpeg结合使用,可生成进行视频点播.直播的压力测试脚本.可支持不同类型的视频流,比如rtmp或者hls形式. 通过如下方式执行脚本:python multiRealPlay.py [rtmp|http] [thread counts] [interval Time] [rtmp | http]:视频播放的不同形式 [thread counts]:并发线程数 [interval Time]:启动每个线程的间隔时间 代码: #!/usr/bin/python # -*- co

  • Python实现的多线程http压力测试代码

    本文实例讲述了Python实现的多线程http压力测试代码.分享给大家供大家参考,具体如下: # Python version 3.3 __author__ = 'Toil' import sys, getopt import threading def httpGet(url, file): import http.client conn = http.client.HTTPConnection(url) conn.request("GET", file) r = conn.getr

  • python实现websocket的客户端压力测试

    使用python进行websocket的客户端压力测试,这个代码是从github上 找到.然后简单修改了下.大神运用了进程池,以及线程池的内容.所以保存下来,学习学习 然后需要说明的是:本次用的python2.7,也尝试用python3.6,但是老实出现websocket-client包和python3不能兼容的情况,提示没有相关的方法.所以不得已最后又采用了python2 # -*- coding:utf-8 -*- # __author__ == 'chenmingle' import we

  • Apache ab并发负载压力测试实现方法

    ab命令原理 Apache的ab命令模拟多线程并发请求,测试服务器负载压力,也可以测试nginx.lighthttp.IIS等其它Web服务器的压力. ab命令对发出负载的计算机要求很低,既不会占用很多CPU,也不会占用太多的内存,但却会给目标服务器造成巨大的负载,因此是某些DDOS攻击之必备良药,老少皆宜.自己使用也须谨慎.否则一次上太多的负载,造成目标服务器直接因内存耗光死机,而不得不硬重启,得不偿失. 在带宽不足的情况下,最好是本机进行测试,建议使用内网的另一台或者多台服务器通过内网进行测

  • 并发数据库压力测试的shell脚本代码

    复制代码 代码如下: #!/bin/bash#********************************##并发后台运行fun                 ##for wanggy 2012-01-25           ##note:                           ##fun_num fun函数后台运行次数     ##sql_num 每个函数sql运行次数     ##********************************##数据库变量设置dbho

  • 使用Apache ab工具对Apache服务器进行简单的压力测试

    1.安裝ab命令 sudo apt-get install apache2-utils 2.ab命令参数说明 Usage: ab [options] [http[s]://]hostname[:port]/path  Options are: //总的请求数 -n requests Number of requests to perform //一次同时并发的请求数 总的请求数(n)=次数*一次并发数(c) -c concurrency Number of multiple requests t

  • 使用Python的Treq on Twisted来进行HTTP压力测试

    从事API相关的工作很有挑战性,在高峰期保持系统的稳定及健壮性就是其中之一,这也是我们在Mailgun做很多压力测试的原因. 这么久以来,我们已经尝试了很多种方法,从简单的ApacheBench到复杂些的自定义测试套.但是本贴讲述的,是一种使用python进行"快速粗糙"却非常灵活的压力测试的方法. 使用python写HTTP客户端的时候,我们都很喜欢用 Requests library.这也是我们向我们的API用户们推荐的.Requests 很强大,但有一个缺点,它是一个模块化的每线

  • python多线程http压力测试脚本

    本文实例为大家分享了python多线程http压力测试的具体代码,供大家参考,具体内容如下 #coding=utf-8 import sys import time import thread import httplib, urllib import random import uuid import logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d

  • Python+Pytest实现压力测试详解

    目录 1.程序说明 1.1 设置测试参数 1.2 初始化测试结果 1.3 定义测试函数 1.4 创建线程.执行线程.等待 1.5 计算测试结果 1.6 将测试结果写入文件 2.程序执行 2.1 直接执行 2.2 加个装饰器然后出报告 3.案例缺陷 4 完整源码 在现代Web应用程序中,性能是至关重要的.为了确保应用程序能够在高负载下正常运行,我们需要进行性能测试. 今天,应小伙伴的提问, 田辛老师来写一个Pytest进行压力测试的简单案例. 这个案例的测试网站我们就隐藏了,不过网站的基本情况是:

  • python 实现手机自动拨打电话的方法(通话压力测试)

    现在能用自动化实现的,尽量使用自动化程序去操作,代替人工去操作,更有效率. 今天说下用python结合adb命令去实现安卓手机端的通话压力测试. #操作前先在设置里打开power键可以结束通话按钮,否则会导致代码报错 from time import sleep import os def test_call(number1,number2,number3,number4,number5): #拨打电话 call = os.popen('adb shell am start -a android

随机推荐