我认为你需要类似于Python subprocess 的东西。在该示例中，将dmesg 视为您的号码生成器，将grep 视为您的套接字服务器；但是，您只需在套接字服务器中打开一个PIPE，而不是生成一个 shell 命令。
假设您有两个进程，talker 和 server（在 Python 2.x 中）...
talker.py
>
"""
Call as talker.py <INTEGER_DELAY>
"""
import time
import sys
assert sys.version_info < (3, 0, 0)
DELAY = int(sys.argv[1])
ii = 1
while True:
sys.stdout.write("Talk every %i seconds, blabbed %i times\n" % (DELAY, ii))
sys.stdout.flush()
ii += 1
time.sleep(DELAY)
server.py
>
#!/usr/bin/env python
from subprocess import Popen, PIPE
from os import kill
import signal
import sys
assert sys.version_info < (3, 0, 0)
talkpipe = Popen(['python', 'talker.py', '3'],
shell=False, stdout=PIPE)
try:
while True:
line = talkpipe.stdout.readline()
print "SERVER HEARD", line.strip()
except KeyboardInterrupt:
print "Killing child..."
kill(talkpipe.pid, signal.SIGTERM)
使用./server.py 运行它会产生...
[mpenning@Bucksnort tmp]$ python server.py
SERVER HEARD Talk every 3 seconds, blabbed 1 times
SERVER HEARD Talk every 3 seconds, blabbed 2 times
SERVER HEARD Talk every 3 seconds, blabbed 3 times
SERVER HEARD Talk every 3 seconds, blabbed 4 times
^CTraceback (most recent call last):
File "talker.py", line 11, in <module>
time.sleep(DELAY)
KeyboardInterrupt
Killing child...
[mpenning@Bucksnort tmp]$
这些脚本的（更强大的）Python3 版本是...
talker.py
>
"""
talker.py: run talker.py with continuous stdout stream and control-C handling.
talker.py was written for Python3, but it should be fine under Python 2.X...
- To run under python 2.x, set `ENFORCE_PYTHON_3 = 0`
Execute with: 'python talker.py <DELAY_INTEGER>'
"""
import signal
import types
import time
import sys
import os
CATCH_CONTROL_C = 1
ENFORCE_PYTHON_3 = 1
BOUNDARY_CONDITIONS_SET = 0
def custom_signal_handler(signum, stackframe):
assert isinstance(signum, int)
assert isinstance(stackframe, types.FrameType)
assert isinstance(BOUNDARY_CONDITIONS_SET, int) and BOUNDARY_CONDITIONS_SET >= 1
print("---> custom_signal_handler() got signal: %s <---" % signum)
if signum == 0:
# signal.SIG_DFL
print(repr(stackframe))
elif signum == 2:
# signal.SIGINT
assert CATCH_CONTROL_C >= 1
raise KeyboardInterrupt(repr(stackframe))
else:
raise NotImplementedError("Cannot handle signal %s" % signal.signum)
def set_boundary_conditions():
global BOUNDARY_CONDITIONS_SET
global DELAY_INTEGER
assert isinstance(BOUNDARY_CONDITIONS_SET, int)
assert BOUNDARY_CONDITIONS_SET <= 0
if ENFORCE_PYTHON_3 == 1:
assert tuple(sys.version_info) >= (3, 0, 0)
# len(sys.argv) will be 1 if this script is imported into python
# len(sys.argv) will be >= 2 if this script was called from CLI w/ args
if len(sys.argv) == 1:
# Called from an import statement...
assert sys.argv[0] == ''
DELAY_INTEGER = 1
elif len(sys.argv) >= 2:
# Called from the command-line... sys.argv[0] is the file name...
assert sys.argv[0] == os.path.basename(__file__).strip()
print("FOO", sys.argv[0])
DELAY_INTEGER = int(sys.argv[1])
assert DELAY_INTEGER > 0
else:
raise ValueError("DELAY_INTEGER was not specified correctly in the CLI")
if CATCH_CONTROL_C >= 1:
assert custom_signal_handler
# Send control-C to the custom signal handler...
signal.signal(signal.SIGINT, custom_signal_handler)
print(" The current handler for signal.SIGINT is %s" % custom_signal_handler)
else:
assert signal.SIG_DFL
# signal.SIGINT (control-C) is 2...
signal.signal(signal.SIGINT, signal.SIG_DFL)
print("The current handler for signal.SIGINT is signal.SIG_DFL")
BOUNDARY_CONDITIONS_SET = 1
def loop_talking_over_stdout():
"""
Write messages to stdout with a variable delay...
This function (loop_talking_over_stdout()) is a blocking call...
"""
assert BOUNDARY_CONDITIONS_SET >= 1
ii = 1
while True:
sys.stdout.write("Talk every %i seconds, blabbed %i times\n" % (
DELAY_INTEGER, ii))
sys.stdout.flush()
ii += 1
time.sleep(DELAY_INTEGER)
if __name__=="__main__":
print("running python %s" % list(sys.version_info))
set_boundary_conditions()
loop_talking_over_stdout()
server.py
>
#!/usr/bin/env python3
from subprocess import Popen, PIPE
from os import kill
import signal
import shlex
import sys
import os
assert sys.version_info >= (3, 0, 0,)
def custom_signal_handler(signum, stackframe):
assert isinstance(signum, int)
print("Got signal: %s" % signum)
raise KeyboardInterrupt(repr(stackframe))
# Send control-C to the custom signal handler...
signal.signal(signal.SIGINT, custom_signal_handler)
TIMEOUT = 3
while True:
filepath = os.sep.join((os.getcwd(), 'talker.py'))
cmd = 'python3 %s %s' % (filepath, TIMEOUT)
talkpipe = Popen(shlex.split(cmd), shell=False,
stdin=PIPE, stdout=PIPE, stderr=PIPE)
stdout = talkpipe.stdout
stderr = talkpipe.stderr
stdin = talkpipe.stdin
for line_out in iter(stdout.readline, b''):
print(b'out>>> ' + line_out.rstrip())
for line_err in iter(stderr.readline, b''):
print(b'err-->' + line_err)