我的当前解决方案是设置一个IPython ZMQ内核。我不只是使用
IPython.embed_kernel()
因为这会产生许多副作用,例如干扰
sys.stdout
、
sys.stderr
、
sys.excepthook
、
signal.signal
等,我不想要这些副作用。另外,
embed_kernel()
是阻塞的,在单独的线程中并不能真正地开箱即用(请参见
此处)。所以,我想出了下面这个代码,但我认为它太过复杂了(这也是我在
这里创建功能请求的原因)。
def initIPythonKernel():
try:
import IPython.kernel.zmq.ipkernel
from IPython.kernel.zmq.ipkernel import Kernel
from IPython.kernel.zmq.heartbeat import Heartbeat
from IPython.kernel.zmq.session import Session
from IPython.kernel import write_connection_file
import zmq
from zmq.eventloop import ioloop
from zmq.eventloop.zmqstream import ZMQStream
IPython.kernel.zmq.ipkernel.signal = lambda sig, f: None
except ImportError, e:
print "IPython import error, cannot start IPython kernel. %s" % e
return
import atexit
import socket
import logging
import threading
assert isinstance(threading.currentThread(), threading._MainThread)
try:
connection_file = "kernel-%s.json" % os.getpid()
def cleanup_connection_file():
try:
os.remove(connection_file)
except (IOError, OSError):
pass
atexit.register(cleanup_connection_file)
logger = logging.Logger("IPython")
logger.addHandler(logging.NullHandler())
session = Session(username=u'kernel')
context = zmq.Context.instance()
ip = socket.gethostbyname(socket.gethostname())
transport = "tcp"
addr = "%s://%s" % (transport, ip)
shell_socket = context.socket(zmq.ROUTER)
shell_port = shell_socket.bind_to_random_port(addr)
iopub_socket = context.socket(zmq.PUB)
iopub_port = iopub_socket.bind_to_random_port(addr)
control_socket = context.socket(zmq.ROUTER)
control_port = control_socket.bind_to_random_port(addr)
hb_ctx = zmq.Context()
heartbeat = Heartbeat(hb_ctx, (transport, ip, 0))
hb_port = heartbeat.port
heartbeat.start()
shell_stream = ZMQStream(shell_socket)
control_stream = ZMQStream(control_socket)
kernel = Kernel(session=session,
shell_streams=[shell_stream, control_stream],
iopub_socket=iopub_socket,
log=logger)
write_connection_file(connection_file,
shell_port=shell_port, iopub_port=iopub_port, control_port=control_port, hb_port=hb_port,
ip=ip)
print "To connect another client to this IPython kernel, use:", \
"ipython console --existing %s" % connection_file
except Exception, e:
print "Exception while initializing IPython ZMQ kernel. %s" % e
return
def ipython_thread():
kernel.start()
try:
ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
pass
thread = threading.Thread(target=ipython_thread, name="IPython kernel")
thread.daemon = True
thread.start()
请注意,这段代码现已过时。我制作了一个
包,其中应该包含更近期的版本,并且可以通过pip进行安装。
其他附加到运行中的CPython进程而不需要事先准备的替代方法。它们通常使用操作系统的调试功能(或使用gdb/lldb)附加到本地CPython进程,然后注入一些代码或仅分析本地CPython线程堆栈。
以下是其他替代方法,您可以事先准备好Python脚本来侦听某个(tcp/file)套接字,以提供远程调试和/或Python shell / REPL界面。
一些概述和收集的代码示例:
(此概述来自这里。)