python实现直播服务非rtmp版本(非常简单)

主程序

使用说明

"""执行主程序
执行直播程序
执行用户程序"""

import multiprocessing.managers
from multiprocessing import Queue
# 任务队列
task_queue=Queue()
# 结果队列
result_queue=Queue()
result_queue1=Queue()
# 提供给从机调用后得到队列中的数据
# 就是接口
class QueueManager(multiprocessing.managers.BaseManager):pass
class StartMain():def return_task(self):return task_queuedef return_result(self):return result_queuedef return_result1(self):return result_queue1# 开启分布式支持def start_main(self):multiprocessing.freeze_support()QueueManager.register('get_task', callable=self.return_task)QueueManager.register('get_result', callable=self.return_result)QueueManager.register('get_result1', callable=self.return_result1)self.manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))self.manager.start()self.task =self.manager.get_task()self.result = self.manager.get_result()self.result1 = self.manager.get_result1()return self.task,self.result,self.result1def stop_mian(self):self.manager.shutdown()
if __name__=="__main__":s=StartMain()req=s.start_main()while True:user_list=[]if req[2].qsize()==0:while req[1].qsize()>0:user_list.append(req[1].get())data=req[0].get()for user in set(user_list):req[2].put(data)s.stop_mian()

直播程序

import multiprocessing.managers
import timeclass QueueManager(multiprocessing.managers.BaseManager):passclass StartCelent():def person_live_celent(self, data):multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task')QueueManager.register('get_result')manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))manager.connect()task = manager.get_task()if task.qsize()>=1:passelse:task.put(data)if __name__ == "__main__":# 使用opencv打开摄像头进行截取使用使用time.time()验证while True:StartCelent().person_live_celent(time.time())

观看程序

import multiprocessing.managersclass QueueManager(multiprocessing.managers.BaseManager):passclass StartCelent():def watch_live_celent(self):multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task')QueueManager.register('get_result1')QueueManager.register('get_result')manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))manager.connect()result = manager.get_result()result1 = manager.get_result1()result.put("信号")res = result1.get()return resif __name__ == "__main__":for _ in range(100):# 这里还要添加显示功能使用opencv就好print(StartCelent().watch_live_celent())

下面是完整代码

主程序

import multiprocessing.managers
from multiprocessing import Queue
# 任务队列
task_queue=Queue()
# 结果队列
result_queue=Queue()
result_queue1=Queue()
# 提供给从机调用后得到队列中的数据
# 就是接口
class QueueManager(multiprocessing.managers.BaseManager):pass
class StartMain():def return_task(self):return task_queuedef return_result(self):return result_queuedef return_result1(self):return result_queue1# 开启分布式支持def start_main(self):multiprocessing.freeze_support()QueueManager.register('get_task', callable=self.return_task)QueueManager.register('get_result', callable=self.return_result)QueueManager.register('get_result1', callable=self.return_result1)self.manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))self.manager.start()self.task =self.manager.get_task()self.result = self.manager.get_result()self.result1 = self.manager.get_result1()return self.task,self.result,self.result1def stop_mian(self):self.manager.shutdown()
if __name__=="__main__":s=StartMain()req=s.start_main()while True:user_list=[]if req[2].qsize()==0:while req[1].qsize()>0:user_list.append(req[1].get())data=req[0].get()for user in set(user_list):req[2].put(data)s.stop_mian()

直播程序

import multiprocessing.managers
import timeclass QueueManager(multiprocessing.managers.BaseManager):passclass StartCelent():def person_live_celent(self, data):multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task')QueueManager.register('get_result')manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))manager.connect()task = manager.get_task()if task.qsize()>=1:passelse:task.put(data)if __name__ == "__main__":# 使用opencv打开摄像头进行截取使用使用time.time()验证import cv2"""函数名：cv2.VideoCapture()功  能：通过摄像头捕获实时图像数据返回值：有参数一：摄像头代号，0为默认摄像头，笔记本内建摄像头一般为 0或者填写视频名称直接加载本地视频文件"""cap = cv2.VideoCapture(0)  # 创建一个 VideoCapture 对象"""函数名：cap.set( propId , value )功  能：设置视频参数。设置视频的宽高值和摄像头有关使用笔记本内置摄像头时只能设置为 1280*720 以及 640*480，哪怕设置值不同时都会自动校正过来，并且仍然返回 True返回值：布尔值参数一：需要设置的视频参数参数二：设置的参数值"""cap.set(3, 480)cap.set(4, 320)flag = 1;  # 设置一个标志，用来输出视频信息"""函数名：cv2.isOpened()功  能：返回一个布尔值（ True / False ），检查是否初始化成功，成功返回 True返回值：布尔值"""while (cap.isOpened()):  # 循环读取每一帧"""函数名：cap.read()功  能：返回两个值先返回一个布尔值，如果视频读取正确，则为 True，如果错误，则为 False，也可用来判断是否到视频末尾再返回一个值，为每一帧的图像，该值是一个三维矩阵通用接收方法为：ret,frame = cap.read();这样 ret 存储布尔值，frame 存储图像若使用一个变量来接收两个值，如frame = cap.read()则 frame 为一个元组，原来使用 frame 处需更改为 frame[1]返回值：R1：布尔值R2：图像的三维矩阵"""ret_flag, Vshow = cap.read()StartCelent().person_live_celent(Vshow)cv2.imshow("Capture_Test", Vshow)  # 窗口显示，显示名为 Capture_Testk = cv2.waitKey(1) & 0xFF  # 每帧数据延时 1ms，延时不能为 0，否则读取的结果会是静态帧if k == ord('s'):  # 若检测到按键 ‘s’，打印字符串"""函数名：cap.get( propId )功  能：查询视频的参数返回值：无参数一：查询的视频参数，其中部分值可以使用 cap.set() 进行修改"""print(cap.get(3))print(cap.get(4))elif k == ord('q'):  # 若检测到按键 ‘q’，退出breakcap.release()  # 释放摄像头cv2.destroyAllWindows()  # 删除建立的全部窗口

用户程序

import multiprocessing.managers
class QueueManager(multiprocessing.managers.BaseManager):pass
class StartCelent():def watch_live_celent(self):multiprocessing.freeze_support()# 注册可以访问队列并得到结果的函数QueueManager.register('get_task')QueueManager.register('get_result1')QueueManager.register('get_result')manager = QueueManager(address=('192.168.0.123', 8888), authkey='password'.encode('utf-8'))manager.connect()result = manager.get_result()result1 = manager.get_result1()result.put("信号")res = result1.get()return resif __name__ == "__main__":import matplotlib.pyplot as pltfrom matplotlib import animationfig = plt.figure()def update(i):image_l=StartCelent().watch_live_celent()s=image_l.copy()r=image_l[:, :, :1]g=image_l[:, :, 1:2]b=image_l[:, :, 2:3]s[:, :, :1]=bs[:, :, 1:2]=gs[:, :, 2:3]=rreturn plt.imshow(s)ani = animation.FuncAnimation(fig, update)plt.show()

配合多进程就可以实现一个软件多进程服务

def f():print("你需要同时启动的程序")if __name__ == '__main__':p = Process(target=f)p.start()

python实现直播服务非rtmp版本(非常简单)相关推荐

0基础搭建视频直播服务（RTMP推流云直播）
本文主要通过对直播工具和云直播网页控制台操作,从全局角度快速上手云直播服务的基本功能和接入使用.详细流程如下: 备注:本文不包含频道管理.直播处理.统计分析部分内容,主要为了用最快速的方式测试直播视频 ...
python写rest服务_Python 如何创建一个简单的REST接口
问题你想使用一个简单的REST接口通过网络远程控制或访问你的应用程序,但是你又不想自己去安装一个完整的web框架. 解决方案构建一个REST风格的接口最简单的方法是创建一个基于WSGI标准(PEP ...
LeetCode：278（Python）—— 第一个错误的版本（简单）
第一个错误的版本概述:你是产品经理,目前正在带领一个团队开发新的产品.不幸的是,你的产品的最新版本没有通过质量检测.由于每个版本都是基于之前的版本开发的,所以错误的版本之后的所有版本都是错的.假设你 ...
如何配置Flash Media Live Encoder (FMLE)从而使用Azure直播服务
Azure媒体服务中的直播服务已经在中国Azure开始公共预览.通过这篇英文博客,您可以了解到直播服务对RTMP协议的支持.以及多种客户端编码器的配置. http://blogs.msdn.com/b ...
windows下ffmpeg+nginx-rtmp环境搭建及opencv+ffmpeg+python实现拉流和rtmp直播推流
由于工作需要最近在研究rtsp拉流与直播推流的问题,目前先在我本机上进行了实验,本博客记录学习的内容,包含windows下ffmpeg.nginx-rtmp环境搭建及opencv+ffmpeg+pyt ...
服务器搭建和使用 ubuntu_Ubuntu中使用Nginx+rtmp搭建流媒体直播服务
一.背景本篇文章是继上一篇文章<Ubuntu中使用Nginx+rtmp模块搭建流媒体视频点播服务>文章而写,在上一篇文章中我们搭建了一个点播服务器,在此基础上我们再搭建一个直播服务器, ...
Windows远程桌面实现之十一：桌面屏幕通过各种直播服务端直播（RTSP, RTMP, HTTP-FLV, HLS)
by fanxiushu 2020-01-23 转载或引用请注明原始作者. 此文还是基于xdisp_virt远程项目中的一个子功能.在把xdisp_virt移植到各种平台之后,就想着再做点什么新功能, ...
【直播资料下载】Python 系列直播——深入Python与日志服务，玩转大规模数据分析处理实战第二讲...
流畅的Python数据处理及大数据处理ETL Python 系列直播--深入Python与日志服务,玩转大规模数据分析处理实战直播讲师:丁来强(成喆)--阿里高级技术专家,从事阿里云日志服务相关的产 ...
【直播资料下载】Python 系列直播——深入Python与日志服务，玩转大规模数据分析处理实战第三讲...
一站式开发者服务,海量学习资源0元起,尽在开发者分会场 Python3舒适性编程与兼容Py2/3实践 Python 系列直播--深入Python与日志服务,玩转大规模数据分析处理实战直播时间:3月1 ...

python实现直播服务非rtmp版本(非常简单)

主程序

直播程序

观看程序

下面是完整代码

python实现直播服务非rtmp版本(非常简单)相关推荐

最新文章

热门文章