如何通过 input() 控制线程？
How to control a Thread via input()?
我想要 运行 一个代码，其进程与我的主代码并行，但也想访问它的参数或 start/stop 通过命令提示符的进程。
我的机器是win7 64位。想到的是：
from multiprocessing import Process
class dllapi():
...
def apiloop(params, args):
apiclient = dllapi(**args)
while True:
apiclient.cycle()
params = [....]
def mainloop(args):
p = Process(target = apiloop, args=(params, args, ))
while True:
cmd = input()
if cmd == 'kill':
p.terminate()
if cmd == 'stop':
pass # no idea
if cmd == 'resume':
pass # no idea
if cmd == 'report':
print (params)
我想简单点。我确实尝试将 apiloop 作为线程，但 input() 可能会冻结程序并停止 apiloop 工作，直到我按下 enter...
为了共享 apiloop 进程的参数，我确实尝试了队列和管道，但在我看来，队列需要 .join 来等待 apiloop 完成并且管道有缓冲区限制。
（实际上我可以每 1 秒制作 apiclient.cycle 运行，但我希望让 apiclient 保持活力）
我想知道是否值得更深入地研究多处理（例如，我也会尝试使用管理器...）或者是否有其他方法更适合我的情况。提前致谢...
* 更新：201809170953*
与经理的一些进展如下：
from multiprocessing import Process, Manager
class dllapi():
...
class webclientapi():
...
def apiloop(args, cmd, params):
apiclient = dllapi(**args)
status = True
while True:
# command from main
if cmd == 'stop':
status = False
elif cmd == 'start':
status = True
cmd = None
# stop or run
if status == True:
apiclient.cycle()
# update parameters
params['status'] = status
def uploadloop(cmds, params):
uploadclient = webclientapi()
status = True
while True:
# command from main
if cmd == 'stop':
status = False
elif cmd == 'start':
status = True
cmd = None
# stop or run
if status == True:
# upload 'status' from apiclient to somewhere
uploadclient.cycle(params['status'])
def mainloop(args):
manager = Manager()
mpcmds = {}
mpparams = {}
mps = {}
mpcmds ['apiloop'] = manager.Value('u', 'start')
mpparams ['apiloop'] = manager.dict()
mps ['apiloop'] = Process(target = apiloop, args=(args, mpcmds['apiloop'], mpparams['apiloop'])
mpcmds ['uploadloop'] = manager.Value('u', 'start')
# mpparams ['uploadloop'] is directly from mpparams ['apiloop']
mps ['uploadloop'] = Process(target = uploadloop, args=(mpcmds['uploadloop'], mpparams['apiloop'])
for key, mp in mps.items():
mp.daemon = True
mp.start()
while True:
cmd = input().split(' ')
# kill daemon process with exit()
if cmd[0] == 'bye':
exit()
# kill individual process
if cmd[0] == 'kill':
mps[cmd[1]].terminate()
# stop individual process via command
if cmd[0] == 'stop':
mpcmds[cmd[1]] = 'stop'
# stop individual process via command
if cmd[0] == 'start':
mpcmds[cmd[1]] = 'start'
# report individual process info via command
if cmd[0] == 'report':
print (mpparams ['apiloop'])
希望这对某人有所帮助。
我正在向您展示如何仅使用线程解决一般问题，因为这是您首先尝试的方法，而您的示例并未提出对子进程的需求。
在下面的示例中，您的 dllapi class 被命名为 Zoo 并且它是 subclassing threading.Thread，添加一些方法以允许执行控制。它在初始化时需要一些 data，它的 cycle 方法简单地重复迭代此数据并计算它看到特定项目的次数。
import time
import logging
from queue import Queue
from threading import Thread
from itertools import count, cycle
class Zoo(Thread):
_ids = count(1)
def __init__(self, cmd_queue, data, *args,
log_level=logging.DEBUG, **kwargs):
super().__init__()
self.name = f'{self.__class__.__name__.lower()}-{next(self._ids)}'
self.data = data
self.log_level = log_level
self.args = args
self.kwargs = kwargs
self.logger = self._init_logging()
self.cmd_queue = cmd_queue
self.data_size = len(data)
self.actual_item = None
self.iter_cnt = 0
self.cnt = count(1)
self.cyc = cycle(self.data)
def cycle(self):
item = next(self.cyc)
if next(self.cnt) % self.data_size == 0: # new iteration round
self.iter_cnt += 1
self.actual_item = f'{item}_{self.iter_cnt}'
def run(self):
"""
Run is the main-function in the new thread. Here we overwrite run
inherited from threading.Thread.
"""
while True:
if self.cmd_queue.empty():
self.cycle()
time.sleep(1) # optional heartbeat
else:
self._get_cmd()
self.cmd_queue.task_done() # unblocks prompter
def stop(self):
self.logger.info(f'stopping with actual item: {self.actual_item}')
# do clean up
raise SystemExit
def pause(self):
self.logger.info(f'pausing with actual item: {self.actual_item}')
self.cmd_queue.task_done() # unblocks producer joining the queue
self._get_cmd() # just wait blockingly until next command
def resume(self):
self.logger.info(f'resuming with actual item: {self.actual_item}')
def report(self):
self.logger.info(f'reporting with actual item: {self.actual_item}')
print(f'completed {self.iter_cnt} iterations over data')
def _init_logging(self):
fmt = '[%(asctime)s %(levelname)-8s %(threadName)s' \
' %(funcName)s()] --- %(message)s'
logging.basicConfig(format=fmt, level=self.log_level)
return logging.getLogger()
def _get_cmd(self):
cmd = self.cmd_queue.get()
try:
self.__class__.__dict__[cmd](self)
except KeyError:
print(f'Command `{cmd}` is unknown.')
input是阻塞函数。您需要将它外包到一个单独的线程中，这样它就不会阻塞您的主线程。在下面的示例中，input 被包裹在 Prompter 中，一个 class subclassing threading.Thread。 Prompter 将输入传递到命令队列。此命令队列由 Zoo.
读取
class Prompter(Thread):
"""Prompt user for command input.
Runs in a separate thread so the main-thread does not block.
"""
def __init__(self, cmd_queue):
super().__init__()
self.cmd_queue = cmd_queue
def run(self):
while True:
cmd = input('prompt> ')
self.cmd_queue.put(cmd)
self.cmd_queue.join() # blocks until consumer calls task_done()
if __name__ == '__main__':
data = ['ape', 'bear', 'cat', 'dog', 'elephant', 'frog']
cmd_queue = Queue()
prompter = Prompter(cmd_queue=cmd_queue)
prompter.daemon = True
zoo = Zoo(cmd_queue=cmd_queue, data=data)
prompter.start()
zoo.start()
终端中的示例会话：
$python control_thread_over_prompt.py
prompt> report
[2018-09-16 17:59:16,856 INFO zoo-1 report()] --- reporting with actual item: dog_0
completed 0 iterations over data
prompt> pause
[2018-09-16 17:59:26,864 INFO zoo-1 pause()] --- pausing with actual item: bear_2
prompt> resume
[2018-09-16 17:59:33,291 INFO zoo-1 resume()] --- resuming with actual item: bear_2
prompt> report
[2018-09-16 17:59:38,296 INFO zoo-1 report()] --- reporting with actual item: ape_3
completed 3 iterations over data
prompt> stop
[2018-09-16 17:59:42,301 INFO zoo-1 stop()] --- stopping with actual item: elephant_3
我想要 运行 一个代码，其进程与我的主代码并行，但也想访问它的参数或 start/stop 通过命令提示符的进程。
我的机器是win7 64位。想到的是：
from multiprocessing import Process
class dllapi():
...
def apiloop(params, args):
apiclient = dllapi(**args)
while True:
apiclient.cycle()
params = [....]
def mainloop(args):
p = Process(target = apiloop, args=(params, args, ))
while True:
cmd = input()
if cmd == 'kill':
p.terminate()
if cmd == 'stop':
pass # no idea
if cmd == 'resume':
pass # no idea
if cmd == 'report':
print (params)
我想简单点。我确实尝试将 apiloop 作为线程，但 input() 可能会冻结程序并停止 apiloop 工作，直到我按下 enter...
为了共享 apiloop 进程的参数，我确实尝试了队列和管道，但在我看来，队列需要 .join 来等待 apiloop 完成并且管道有缓冲区限制。
（实际上我可以每 1 秒制作 apiclient.cycle 运行，但我希望让 apiclient 保持活力）
我想知道是否值得更深入地研究多处理（例如，我也会尝试使用管理器...）或者是否有其他方法更适合我的情况。提前致谢...
* 更新：201809170953*
与经理的一些进展如下：
from multiprocessing import Process, Manager
class dllapi():
...
class webclientapi():
...
def apiloop(args, cmd, params):
apiclient = dllapi(**args)
status = True
while True:
# command from main
if cmd == 'stop':
status = False
elif cmd == 'start':
status = True
cmd = None
# stop or run
if status == True:
apiclient.cycle()
# update parameters
params['status'] = status
def uploadloop(cmds, params):
uploadclient = webclientapi()
status = True
while True:
# command from main
if cmd == 'stop':
status = False
elif cmd == 'start':
status = True
cmd = None
# stop or run
if status == True:
# upload 'status' from apiclient to somewhere
uploadclient.cycle(params['status'])
def mainloop(args):
manager = Manager()
mpcmds = {}
mpparams = {}
mps = {}
mpcmds ['apiloop'] = manager.Value('u', 'start')
mpparams ['apiloop'] = manager.dict()
mps ['apiloop'] = Process(target = apiloop, args=(args, mpcmds['apiloop'], mpparams['apiloop'])
mpcmds ['uploadloop'] = manager.Value('u', 'start')
# mpparams ['uploadloop'] is directly from mpparams ['apiloop']
mps ['uploadloop'] = Process(target = uploadloop, args=(mpcmds['uploadloop'], mpparams['apiloop'])
for key, mp in mps.items():
mp.daemon = True
mp.start()
while True:
cmd = input().split(' ')
# kill daemon process with exit()
if cmd[0] == 'bye':
exit()
# kill individual process
if cmd[0] == 'kill':
mps[cmd[1]].terminate()
# stop individual process via command
if cmd[0] == 'stop':
mpcmds[cmd[1]] = 'stop'
# stop individual process via command
if cmd[0] == 'start':
mpcmds[cmd[1]] = 'start'
# report individual process info via command
if cmd[0] == 'report':
print (mpparams ['apiloop'])
希望这对某人有所帮助。
我正在向您展示如何仅使用线程解决一般问题，因为这是您首先尝试的方法，而您的示例并未提出对子进程的需求。
在下面的示例中，您的 dllapi class 被命名为 Zoo 并且它是 subclassing threading.Thread，添加一些方法以允许执行控制。它在初始化时需要一些 data，它的 cycle 方法简单地重复迭代此数据并计算它看到特定项目的次数。
import time
import logging
from queue import Queue
from threading import Thread
from itertools import count, cycle
class Zoo(Thread):
_ids = count(1)
def __init__(self, cmd_queue, data, *args,
log_level=logging.DEBUG, **kwargs):
super().__init__()
self.name = f'{self.__class__.__name__.lower()}-{next(self._ids)}'
self.data = data
self.log_level = log_level
self.args = args
self.kwargs = kwargs
self.logger = self._init_logging()
self.cmd_queue = cmd_queue
self.data_size = len(data)
self.actual_item = None
self.iter_cnt = 0
self.cnt = count(1)
self.cyc = cycle(self.data)
def cycle(self):
item = next(self.cyc)
if next(self.cnt) % self.data_size == 0: # new iteration round
self.iter_cnt += 1
self.actual_item = f'{item}_{self.iter_cnt}'
def run(self):
"""
Run is the main-function in the new thread. Here we overwrite run
inherited from threading.Thread.
"""
while True:
if self.cmd_queue.empty():
self.cycle()
time.sleep(1) # optional heartbeat
else:
self._get_cmd()
self.cmd_queue.task_done() # unblocks prompter
def stop(self):
self.logger.info(f'stopping with actual item: {self.actual_item}')
# do clean up
raise SystemExit
def pause(self):
self.logger.info(f'pausing with actual item: {self.actual_item}')
self.cmd_queue.task_done() # unblocks producer joining the queue
self._get_cmd() # just wait blockingly until next command
def resume(self):
self.logger.info(f'resuming with actual item: {self.actual_item}')
def report(self):
self.logger.info(f'reporting with actual item: {self.actual_item}')
print(f'completed {self.iter_cnt} iterations over data')
def _init_logging(self):
fmt = '[%(asctime)s %(levelname)-8s %(threadName)s' \
' %(funcName)s()] --- %(message)s'
logging.basicConfig(format=fmt, level=self.log_level)
return logging.getLogger()
def _get_cmd(self):
cmd = self.cmd_queue.get()
try:
self.__class__.__dict__[cmd](self)
except KeyError:
print(f'Command `{cmd}` is unknown.')
input是阻塞函数。您需要将它外包到一个单独的线程中，这样它就不会阻塞您的主线程。在下面的示例中，input 被包裹在 Prompter 中，一个 class subclassing threading.Thread。 Prompter 将输入传递到命令队列。此命令队列由 Zoo.
class Prompter(Thread):
"""Prompt user for command input.
Runs in a separate thread so the main-thread does not block.
"""
def __init__(self, cmd_queue):
super().__init__()
self.cmd_queue = cmd_queue
def run(self):
while True:
cmd = input('prompt> ')
self.cmd_queue.put(cmd)
self.cmd_queue.join() # blocks until consumer calls task_done()
if __name__ == '__main__':
data = ['ape', 'bear', 'cat', 'dog', 'elephant', 'frog']
cmd_queue = Queue()
prompter = Prompter(cmd_queue=cmd_queue)
prompter.daemon = True
zoo = Zoo(cmd_queue=cmd_queue, data=data)
prompter.start()
zoo.start()
终端中的示例会话：
$python control_thread_over_prompt.py
prompt> report
[2018-09-16 17:59:16,856 INFO zoo-1 report()] --- reporting with actual item: dog_0
completed 0 iterations over data
prompt> pause
[2018-09-16 17:59:26,864 INFO zoo-1 pause()] --- pausing with actual item: bear_2
prompt> resume
[2018-09-16 17:59:33,291 INFO zoo-1 resume()] --- resuming with actual item: bear_2
prompt> report
[2018-09-16 17:59:38,296 INFO zoo-1 report()] --- reporting with actual item: ape_3
completed 3 iterations over data
prompt> stop
[2018-09-16 17:59:42,301 INFO zoo-1 stop()] --- stopping with actual item: elephant_3