python - 如何避免高CPU使用率？

Question

我创建了一个单独运行的 zmq_forwarder.py 并将消息从应用程序传递到 sockJS 连接，我目前正在研究 Flask 应用程序如何从通过 zmq 的 sockJS。我正在粘贴我的 zmq_forwarder.py 的内容。我是 ZMQ 的新手，我不知道为什么每次运行它时，它都会占用 100% 的 CPU 负载。

import zmq

# Prepare our context and sockets
context = zmq.Context()

receiver_from_server = context.socket(zmq.PULL)
receiver_from_server.bind("tcp://*:5561")

forwarder_to_server = context.socket(zmq.PUSH)
forwarder_to_server.bind("tcp://*:5562")

receiver_from_websocket = context.socket(zmq.PULL)
receiver_from_websocket.bind("tcp://*:5563")

forwarder_to_websocket = context.socket(zmq.PUSH)
forwarder_to_websocket.bind("tcp://*:5564")

# Process messages from both sockets
# We prioritize traffic from the server
while True:

    # forward messages from the server
    while True:
        try:
            message = receiver_from_server.recv(zmq.DONTWAIT)
        except zmq.Again:
            break

        print "Received from server: ", message
        forwarder_to_websocket.send_string(message)

    # forward messages from the websocket
    while True:
        try:
            message = receiver_from_websocket.recv(zmq.DONTWAIT)
        except zmq.Again:
            break

        print "Received from websocket: ", message
        forwarder_to_server.send_string(message)

如您所见，我设置了 4 个 socket 。该应用程序连接到端口 5561 以将数据推送到 zmq，并连接到端口 5562 以从 zmq 接收(尽管我仍在弄清楚如何实际设置它以监听 zmq 发送的消息)。另一方面，sockjs 在端口 5564 上从 zmq 接收数据并在端口 5563 上向其发送数据。

我读过 zmq.DONTWAIT 使消息接收异步和非阻塞，所以我添加了它。

有没有办法改进代码，以免 CPU 过载？目标是能够使用 zmq 在 flask 应用程序和 websocket 之间传递消息。

Answer 1

您正在一个紧凑的循环中轮询您的两个接收器套接字，没有任何阻塞(zmq.DONTWAIT)，这将不可避免地使 CPU 达到最大。

请注意，ZMQ 支持在单个线程中轮询多个套接字 - 参见 this answer .我认为您可以在 poller.poll(millis) 中调整超时，以便您的代码仅在有大量传入消息时使用大量 CPU，否则空闲。

您的另一个选择是使用 ZMQ 事件循环，使用回调异步响应传入消息。查看PyZMQ documentation关于这个主题，从中改编了以下“echo”示例:

# set up the socket, and a stream wrapped around the socket
s = ctx.socket(zmq.REP)
s.bind('tcp://localhost:12345')
stream = ZMQStream(s)

# Define a callback to handle incoming messages
def echo(msg):
    # in this case, just echo the message back again
    stream.send_multipart(msg)

# register the callback
stream.on_recv(echo)

# start the ioloop to start waiting for messages
ioloop.IOLoop.instance().start()