python - 如何通过在后台运行的程序读取参数？

Question

示例:每 10 秒打印一次列表值的简单程序

import argparse
import time
import sys

myList = []

def parseArguments():
    parser = argparse.ArgumentParser(description="example")
    parser.add_argument('-a', '--addElement', help='adds an element to the list')

    args = parser.parse_args()

    if args.addElement:
        myList.append(args.addElement)

def main():
    parseArguments()

    while(True):
        print(myList)
        time.sleep(10)

问题是程序只读取开始时传递的参数，我希望它在运行时随时读取传递的参数。

我想像服务一样在后台运行程序，并每隔一段时间向程序传递参数。

Answer 1

我知道您要求的看起来像是能够接受异步命令的服务(或守护进程)。

外部接口(interface):

程序富

=> ok 重复打印 ['foo']

后来:

进度条

=> 第二个实例退出，第一个实例重复打印 ['foo', 'bar']

内部设计

这绝非易事!您需要设置一个 IPC 机制以允许第二个实例与第一个实例进行通信，在第一个实例中使用非阻塞 IO(或多线程)。在 Unix 下，您可以使用 os.mkfifo，但如果您想要一个可移植的解决方案，则必须在本地主机上使用 IP 套接字

高级伪代码结构

get argument via argparse
bind to a fix port on localhost, in UDP protocol
if success:
    # ok it is the first prog
    initialize list from argument
    loop:
        get command from UDP socket, with timeout = 10s
        if cmd is add param:
            add parameter to list
        elif cmd is exit:   # not asked in question but should exist
            exit
        print list
else:
    # another prog has taken the socket, pass it the arg
    send the arg to the UDP port with proper protocol

关于这个简单设计的注意事项:存在一个竞争条件，即在第一次尝试绑定(bind)和发送之间退出的套接字上已经有一个程序在等待。要解决这个问题，您应该使用 TCP 协议(protocol)，在监听套接字上使用带有超时的 select，并正常关闭以确保在另一端接收到消息。如果出现错误，您将迭代(最大次数)，因为第一台服务器可能已经退出。

这是一个实现示例:

import socket
import select
import argparse
import time
import sys

TIMEOUT=10
IFACE='127.0.0.1'
PORT=4000
DEBUG=False

myList = []
old = ""

def parseArguments():
    parser = argparse.ArgumentParser(description="example")
    parser.add_argument('-a', '--addElement',
                        help='adds an element to the list')
    parser.add_argument('-q', '--quit', action='store_true',
                        help='closes main service')
    parser.add_argument('-d', '--debug', action='store_true',
                        help='display debug information')

    args = parser.parse_args()

    if args.quit:
        senddata("QUIT\n")
        sys.exit(0)

    if args.debug:
        DEBUG=True

    if args.addElement:
        myList.append(args.addElement)

def read(s):
    global old
    data = old
    while True:
        block = s.recv(1024)
        if len(block) == 0: return data
        if b'\n' in block:
            block,o = block.split(b'\n', 1)
            old = o.decode()
            data += block.decode()
            return data
        data += block.decode()

def gracefulclose(s, msg):
    s.send(msg.encode())
    s.shutdown(socket.SHUT_WR)
    try:
        read(s)
    finally:
        s.close()

def server(s):
    if DEBUG:
        print("SERVER")
    s.listen(5)
    while True:
        sl = select.select([s], [], [], TIMEOUT)
        if len(sl[0]) > 0:
            s2, peer = s.accept()
            try:
                data = read(s2)
                print(data)
                gracefulclose(s2, "OK")
            finally:
                s2.close()
            if data.startswith("QUIT"):
                return
            elif data.startswith("DATA:"):
                myList.append(data[5:])
        print(myList)

def senddata(data):
    s = socket.socket(socket.AF_INET)
    try:
        s.connect((IFACE, PORT))
        s.send(data.encode())
        data = read(s)
        if (data.startswith("OK")):
            return True
    except:
        pass
    finally:
        s.close()
    return False
        
def client():
    return senddata("DATA:" + myList[0] + "\n")
        
def main():
    end = False
    MAX = 5
    while not end and MAX > 0:
        s = socket.socket(socket.AF_INET)
        try:
            s.bind((IFACE, PORT))
        except Exception:
            s.close()
            s = None
        if s:
            try:
                server(s)
            finally:
                s.close()
                return
        else:
            if DEBUG:
                print("CLIENT", " ", 6 - MAX)
            end = client()
        MAX -= 1
        time.sleep(1)

if __name__ == "__main__":
    parseArguments()
    main()