在gtk应用程序中，所有的执行都发生在gtk_main函数内部。其他图形框架作品具有类似的事件循环，app.exec例如QT和Clutter。然而， ZeroMQ是基于这样一个假设，即它被插入到一个循环中（例如，请参见此处的示例）。clutter_mainwhile (1) ...
您如何结合这两种执行策略？
我目前想在用 C 编写的杂乱应用程序中使用 zeromq，所以我当然希望直接回答这个问题，但也请为其他变体添加答案。
问问题
3504 次
5 回答
13
结合 zmq 和 gtk 或混乱的正确方法是将 zmq 队列的文件描述符连接到主事件循环。fd 可以通过使用来检索
int fd;
size_t sizeof_fd = sizeof(fd);
if(zmq_getsockopt(socket, ZMQ_FD, &fd, &sizeof_fd))
perror("retrieving zmq fd");
将其连接到主循环是使用 io_add_watch 的问题：
GIOChannel* channel = g_io_channel_unix_new(fd);
g_io_add_watch(channel, G_IO_IN|G_IO_ERR|G_IO_HUP, callback_func, NULL);
在回调函数中，需要先检查是否真的有东西要读，然后再读。否则，该函数可能会阻塞等待 IO。
gboolean callback_func(GIOChannel *source, GIOCondition condition,gpointer data)
{
uint32_t status;
size_t sizeof_status = sizeof(status);
while (1){
if (zmq_getsockopt(socket, ZMQ_EVENTS, &status, &sizeof_status)) {
perror("retrieving event status");
return 0; // this just removes the callback, but probably
// different error handling should be implemented
}
if (status & ZMQ_POLLIN == 0) {
break;
}
// retrieve one message here
}
return 1; // keep the callback active
}
请注意：这实际上没有经过测试，我从 Python+Clutter 进行了翻译，这是我使用的，但我很确定它会起作用。作为参考，下面是实际有效的完整 Python+Clutter 代码。
import sys
from gi.repository import Clutter, GObject
import zmq
def Stage():
"A Stage with a red spinning rectangle"
stage = Clutter.Stage()
stage.set_size(400, 400)
rect = Clutter.Rectangle()
color = Clutter.Color()
color.from_string('red')
rect.set_color(color)
rect.set_size(100, 100)
rect.set_position(150, 150)
timeline = Clutter.Timeline.new(3000)
timeline.set_loop(True)
alpha = Clutter.Alpha.new_full(timeline, Clutter.AnimationMode.EASE_IN_OUT_SINE)
rotate_behaviour = Clutter.BehaviourRotate.new(
alpha,
Clutter.RotateAxis.Z_AXIS,
Clutter.RotateDirection.CW,
0.0, 359.0)
rotate_behaviour.apply(rect)
timeline.start()
stage.add_actor(rect)
stage.show_all()
stage.connect('destroy', lambda stage: Clutter.main_quit())
return stage, rotate_behaviour
def Socket(address):
ctx = zmq.Context()
sock = ctx.socket(zmq.SUB)
sock.setsockopt(zmq.SUBSCRIBE, "")
sock.connect(address)
return sock
def zmq_callback(queue, condition, sock):
print 'zmq_callback', queue, condition, sock
while sock.getsockopt(zmq.EVENTS) & zmq.POLLIN:
observed = sock.recv()
print observed
return True
def main():
res, args = Clutter.init(sys.argv)
if res != Clutter.InitError.SUCCESS:
return 1
stage, rotate_behaviour = Stage()
sock = Socket(sys.argv[2])
zmq_fd = sock.getsockopt(zmq.FD)
GObject.io_add_watch(zmq_fd,
GObject.IO_IN|GObject.IO_ERR|GObject.IO_HUP,
zmq_callback, sock)
return Clutter.main()
if __name__ == '__main__':
sys.exit(main())
于 2012-06-02T10:17:34.627 回答
6
听起来 ZeroMQ 代码只想尽可能频繁地反复执行。最简单的方法是将 ZeroMQ 代码放入空闲函数或超时函数中，并使用这些函数的非阻塞版本（如果存在）。
对于杂波，您将使用clutter_threads_add_idle()or clutter_threads_add_timeout()。对于 GTK，您将使用g_idle_add()or g_timeout_add()。
更困难但可能更好的方法是使用 为 ZeroMQ 代码创建一个单独的线程g_thread_create()，并while(1)按照他们的建议使用带有阻塞函数的构造。如果你这样做了，你还必须找到一些方法让线程相互通信——GLib 的互斥锁和异步队列通常都很好。
于 2011-06-23T11:53:50.467 回答
2
我发现有一个名为Zeromqt的 QT 集成库。看源码，集成的核心如下：
ZmqSocket::ZmqSocket(int type, QObject *parent) : QObject(parent)
{
...
notifier_ = new QSocketNotifier(fd, QSocketNotifier::Read, this);
connect(notifier_, SIGNAL(activated(int)), this, SLOT(activity()));
}
...
void ZmqSocket::activity()
{
uint32_t flags;
size_t size = sizeof(flags);
if(!getOpt(ZMQ_EVENTS, &flags, &size)) {
qWarning("Error reading ZMQ_EVENTS in ZMQSocket::activity");
return;
}
if(flags & ZMQ_POLLIN) {
emit readyRead();
}
if(flags & ZMQ_POLLOUT) {
emit readyWrite();
}
...
}
因此，它依赖于 QT 的集成套接字处理，而 Clutter 不会有类似的东西。
于 2011-06-23T12:48:44.303 回答
2
您可以获得 0MQ 套接字的文件描述符（ZMQ_FD 选项）并将其与您的事件循环集成。我认为 gtk 有一些处理套接字的机制。
于 2011-07-27T06:30:03.730 回答
1
这是 Python 中的一个示例，使用 PyQt4。它来自一个工作应用程序。
import zmq
from PyQt4 import QtCore, QtGui
class QZmqSocketNotifier( QtCore.QSocketNotifier ):
""" Provides Qt event notifier for ZMQ socket events """
def __init__( self, zmq_sock, event_type, parent=None ):
"""
Parameters:
----------
zmq_sock : zmq.Socket
The ZMQ socket to listen on. Must already be connected or bound to a socket address.
event_type : QtSocketNotifier.Type
Event type to listen for, as described in documentation for QtSocketNotifier.
"""
super( QZmqSocketNotifier, self ).__init__( zmq_sock.getsockopt(zmq.FD), event_type, parent )
class Server(QtGui.QFrame):
def __init__(self, topics, port, mainwindow, parent=None):
super(Server, self).__init__(parent)
self._PORT = port
# Create notifier to handle ZMQ socket events coming from client
self._zmq_context = zmq.Context()
self._zmq_sock = self._zmq_context.socket( zmq.SUB )
self._zmq_sock.bind( "tcp://*:" + self._PORT )
for topic in topics:
self._zmq_sock.setsockopt( zmq.SUBSCRIBE, topic )
self._zmq_notifier = QZmqSocketNotifier( self._zmq_sock, QtCore.QSocketNotifier.Read )
# connect signals and slots
self._zmq_notifier.activated.connect( self._onZmqMsgRecv )
mainwindow.quit.connect( self._onQuit )
@QtCore.pyqtSlot()
def _onZmqMsgRecv():
self._test_info_notifier.setEnabled(False)
# Verify that there's data in the stream
sock_status = self._zmq_sock.getsockopt( zmq.EVENTS )
if sock_status == zmq.POLLIN:
msg = self._zmq_sock.recv_multipart()
topic = msg[0]
callback = self._topic_map[ topic ]
callback( msg )
self._zmq_notifier.setEnabled(True)
self._zmq_sock.getsockopt(zmq.EVENTS)
def _onQuit(self):
self._zmq_notifier.activated.disconnect( self._onZmqMsgRecv )
self._zmq_notifier.setEnabled(False)
del self._zmq_notifier
self._zmq_context.destroy(0)
根据QSocketNotifier_on_ZmqMsgRecv的文档，禁用然后重新启用通知程序。
出于某种原因，最后一次调用 getsockopt 是必要的。否则，通知器在第一个事件后停止工作。我实际上打算为此发布一个新问题。有谁知道为什么需要这个？
请注意，如果您没有在 ZMQ 上下文之前销毁通知程序，那么您在退出应用程序时可能会收到如下错误：
QSocketNotifier: Invalid socket 16 and type 'Read', disabling...
于 2015-08-17T13:18:23.613 回答