我如何使用Python和MutlipRocessing打开程序，并从主过程中发送字符串

Question

我有一个程序，将字符串作为字符串发送到ffmpeg，使用类似的内容：

工作脚本，该脚本在不使用ubuntu

上使用多处理模块的情况

#!/usr/bin/python
import sys, os
import subprocess as sp
import pygame
from pygame.locals import QUIT, KEYUP, K_ESCAPE
import pygame.display
pygame.init()
os.environ['SDL_VIDEODRIVER'] = 'dummy'
pygame.display.init()
Display_Surface = pygame.display.set_mode([1280,720], 0, 32)
# FFMPEG command and settings
command = ['ffmpeg', '-framerate', '25', '-s', '1280x720', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
           '-f', 'lavfi', '-i', 'anullsrc=cl=mono',
           '-pix_fmt', 'yuv420p','-s', 'hd720', '-r', '25', '-g', '50',
           '-f', 'flv', 'rtmp://a.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx']
pipe = sp.Popen(command, bufsize=0, stdin=sp.PIPE)
while True:
    # Quit event handling
    for event in pygame.event.get():
        if event.type == QUIT or (event.type == KEYUP and event.key == K_ESCAPE):
            pygame.quit()
            sys.exit()
    pipe.stdin.write(pygame.image.tostring(Display_Surface, "RGBA"))
pipe.stdin.close()
pygame.display.quit()
os._exit()

这可以正常工作，除了它正在杀死我的CPU，这反过来又导致我的现场流经常冻结。愚蠢的吉尔不会让ffmpeg在另一个CPU/核心上运行，而我有3个完美的核心无济于事。

我只是在另一个过程中打开一些代码以打开FFMPEG。(顺便说一句，我熟悉螺纹。线程，但不熟悉多处理(。

import os
import subprocess as sp
import multiprocessing
class FFMPEG_Consumer():
    def __init__(self):
        proc = multiprocessing.Process(target=self.start_ffmpeg)
        proc.start()
    def start_ffmpeg(self):
        command = ['ffmpeg','-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
                   '-f, 'lavfi', '-i', 'anullsrc=channel_layout=stereo:sample_rate=44100',
                   '-pix_fmt', 'yuv420p','-s', 'hd720', '-f', 'flv', 'rtmp://example.com']
        pipe = sp.Popen(command, bufsize=-1, stdin=sp.PIPE)
    def send_down_the_pipe(self, frame):
        pipe.stdin.write(frame)
ffmpeg = FFMPEG_Consumer()

对于任何知道如何使用多处理的人，我敢肯定，您会立即看到这不起作用，因为我不能在过程中以这种方式共享变量。但是，它确实在另一个核心上打开FFMPEG。

大多数在线教程和资源集中于创建工人和队列的库，以向这些工人发送一些要处理的东西，直到完成作业为止。但是，我正在尝试通过每次迭代反复将新字符串发送给FFMPEG。

如何将字符串插入FFMPEG的该过程/实例？

还是我想做的是不可能的？

这是工作解决方案(带有愚蠢的FFMPEG设置(：

#!/usr/bin/python
import sys, os, multiprocessing
import subprocess as sp
import pygame
from pygame.locals import QUIT, KEYUP, K_ESCAPE
import pygame.display
pygame.init()
os.environ['SDL_VIDEODRIVER'] = 'dummy'
pygame.display.init()
Display_Surface = pygame.display.set_mode([1280,720], 0, 32)
class FFMPEGConsumer(object):
    def __init__(self):
        self._r, self._w = multiprocessing.Pipe()
        self.reader = os.fdopen(self._r.fileno(), 'r')
        self.writer = os.fdopen(self._w.fileno(), 'w', 0)
        self.proc = None
    def start_ffmpeg(self):
        command = ['ffmpeg', '-framerate', '25', '-s', '1280x720', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-',
           '-f', 'lavfi', '-i', 'anullsrc=cl=mono',
           '-pix_fmt', 'yuv420p','-s', 'hd720', '-r', '25', '-g', '50',
           '-f', 'flv', 'rtmp://a.rtmp.youtube.com/live2/xxxx-xxxx-xxxx-xxxx']
        
        self.proc = sp.Popen(command, bufsize=-1, stdin=self.reader)
    def send_down_the_pipe(self, frame):
        self.writer.write(frame)
        #print self._stdin.read()
    def __del__(self):
        self.reader.close()
        self.writer.close()
ffmpeg = FFMPEGConsumer()
while True:
    # Quit event handling
    for event in pygame.event.get():
        if event.type == QUIT or (event.type == KEYUP and event.key == K_ESCAPE):
            pygame.quit()
            sys.exit()
    ffmpeg.send_down_the_pipe(pygame.image.tostring(Display_Surface, "RGBA"))
    proc.join()
pipe.stdin.close()
pygame.display.quit()
os._exit()

所有核心都在射击，到目前为止没有滞后！

Answer 1

您可以使用的是multiprocessing.Pipe用于与子过程交流。

multiprocess.Pipe为您提供了两端的管道，其中一个可用于写作，另一端用于阅读，反之亦然(如果将双工标志设置为true(。这两个末端的事情是这些只是文件描述符。这就是为什么您需要像文件一样打开它们，然后读/写。要打开它们，您必须使用os.fdopen。从那以后，它就像阅读/写作到文件一样。

这样的事情应该有帮助：

import os
import subprocess as sp
import multiprocessing

class FFMPEG_Consumer():
    def __init__(self):
        r, w = multiprocessing.Pipe()
        self._stdin = os.fdopen(r.fileno(), 'r')
        self._stdout = os.fdopen(w.fileno(), 'w')
        proc = multiprocessing.Process(target=self.start_ffmpeg)
        proc.start()
    def start_ffmpeg(self):
        command = ['ffmpeg', '-pix_fmt', 'rgba', '-f', 'rawvideo', '-i', '-', '-f',
                   'lavfi', '-i', 'anullsrc=channel_layout=stereo:sample_rate=44100',
                   '-pix_fmt', 'yuv420p', '-s', 'hd720', '-f', 'flv', 'rtmp://example.com']
        sp.Popen(command, bufsize=-1, stdin=self._stdin)
    def send_down_the_pipe(self, frame):
        self._stdout.write(frame)

ffmpeg = FFMPEG_Consumer()
ffmpeg.send_down_the_pipe(frame)

编辑1。

您可以调整缓冲以更好地适合您的需求。当您使用os.fdopen时，第三个参数用于缓冲区大小。 0是在不缓冲的情况下打开它。

编辑2。

我不想出于教育目的删除旧版本。我以前的脚本有什么问题？

class FFMPEG_Consumer():
    def __init__(self):
        r, w = multiprocessing.Pipe()
        self._stdin = os.fdopen(r.fileno(), 'r')
        self._stdout = os.fdopen(w.fileno(), 'w')
        ...

执行__init__方法后，收集了r和w文件描述符。现在，您拥有仍然存在的self._stdin和self._stdout，并指向非现有文件描述符。实例化FFMPEG_Consumer后，您会得到以下错误：

close failed in file object destructor:
IOError: [Errno 9] Bad file descriptor
close failed in file object destructor:
IOError: [Errno 9] Bad file descriptor

如何解决这个问题？您必须在实例化FFMPEG_Consumer后确保保留文件描述符。

要示例脚本显示如何解决此问题，让我给您几个建议：

1. 您说您使用Python2.7。这就是为什么您可能需要在定义类时继承object的原因。您将在示例脚本中看到它。
2. 命名课时，惯例是使用骆驼盒。因此FFMPEGConsumer比FFMPEG_Consumer好。

现在进入工作示例。

工作示例

这是我使用的两个脚本，可以使任何人在不使用ffmpeg的情况下测试它的工作方式。注意：这是Python2.7版本。

需要用户输入的脚本so6_reader.py：

#!/usr/bin/env python2.7
if __name__ == "__main__":
    for i in range(2):
        a = raw_input('Enter the string: ')
        print 'You entered %s' % a

启动启动so6_reader.py并写入其标准输入的子进程的脚本。

#!/usr/bin/env python2.7
import os
import multiprocessing
import subprocess as sp

class FFMPEGConsumer(object):
    def __init__(self):
        self._r, self._w = multiprocessing.Pipe()
        self.reader = os.fdopen(self._r.fileno(), 'r')
        self.writer = os.fdopen(self._w.fileno(), 'w', 0)
        self.proc = None
    def start_ffmpeg(self):
        command = ['python', 'so6_reader.py']
        self.proc = sp.Popen(command, bufsize=0, stdin=self.reader)
    def send_down_the_pipe(self, frame):
        self.writer.write(frame)
    def __del__(self):
        self.reader.close()
        self.writer.close()

ffmpeg = FFMPEGConsumer()
proc = multiprocessing.Process(target=ffmpeg.start_ffmpeg)
proc.start()
ffmpeg.send_down_the_pipe('test 001n')
ffmpeg.send_down_the_pipe('test 002n')
proc.join()