zeromq 相关问题

如何使用ZeroMQ通过TCP套接字发送PNG图像？

我有一个在Python上运行的Keras模型，我想从Unity 3D摄像机向该模型发送帧。我可以通过使用外部库轻松地在它们之间传输字符串。此外部库发送...

为什么我不能为zmq编译基本的发送和接收模拟

我正在zmq.hpp（仅zmq c ++标头的活页夹）中为socket_t :: send和socket_t :: recv写一个gmock。 >

是什么原因导致ZeroMQ延迟高的情况以及如何避免'em？

'__ u32 __fswab32（__ u32）'不能出现在netinet / tcp_md5.h中的常量表达式中。

，因此我面临以下错误。 ==========错误======== /auto/andpkg/rep_cache//wr-x86-xeon/5.0/sysroot/intel_xeon_32_core-wrs-linux/usr/include / ...

将clustermq R软件包用作foreach的并行后端

我已经开始将clustermq软件包用作drake管道的并行后端，并且对我观察到的性能改进印象非常深刻。我有兴趣评估...

ZeroMQ：等待消息而不消耗线程

我有一些简单的python代码，等待主题消息。但是...运行此程序时，python进程将占用CPU。我知道其他语言，例如套接字，有一种方法可以...

ZeroMQ：等待消息而不消耗线程

我有一些简单的python代码，等待主题消息。但是...运行此程序时，python进程将占用cpu。我知道其他语言，例如套接字，有一种方法可以...

ZeroMQ：推/拉时延迟非常高

总结，我发现使用zmq.PUSH / PULL套接字的接收器的延迟很高。详细信息我是ZeroMQ的新手，正在尝试从发送方向接收方发送几兆字节（我的最终目标是发送...

如何在MFC应用程序中启动ZeroMQ线程？

（Visual Studio 2017，C ++ 17，cppzmq 4.6.0）我是MFC编程的新手。我只是想在一个线程上创建一个ZeroMQ答复服务器，该服务器将随时准备接收用户请求并运行某些...

在这种情况下我应该使用哪个ZMQ模式？

我是ZMQ和多线程设计的新手，所以我希望对我的场景的设计模式有所了解：我有一个充当中间件的服务器，它将从...接收外部数据。

ZeroMQ：如何发布到多个接收者，但其中一个除外

zeromq：发布到多个接收者，但其中一个除外

我有N个发布者订阅者。该消息是一个简单的布尔值。消息传递模式与常规PubSub略有不同：当一个订户收到一个true时，其他所有...

ZeroMQ多主题过滤用法中一些丢失的消息

我有一个使用JeroMQ 0.5.2用Java编写的ZeroMQ订户应用程序。在代码中，我订阅了多个这样的主题：套接字订阅者= context.createSocket（SocketType.SUB）;订户...。

pyzmq：套接字因“ ZMQError：无法在当前状态下完成操作”而中断]

对于ZMQ而言，我还很陌生，并且试图构建非常基本的消息传递系统代码完全是基于此处的示例，出于某种原因，在最后一条消息到达前端之后，由于某种原因而有些曲折...

zmq为什么将多个消息打包到一个TCP帧？

我使用带有ZMQ_DONTWAIT标志的ZMQ_PUSH套接字发送消息。每个消息间隔300毫秒（因此，我严重怀疑我的线程调度是否如此）。但是不时有几个（最多7个...

ZeroMQ如何使用C 我是ZMQ的新手。我有多个发布商和一个客户。寻求建议以最佳方式实施它。当前，它利用单个客户端和服务器的回复-请求模式；必须将其扩展到多个发布者和一个订户。 此应用程序将在不支持C11的QNX系统上运行，因此zmq::mutlipart_t没有帮助。 void TransportLayer::Init() { socket.bind( "tcp://*:5555" ); } void TransportLayer::Receive() { while ( true ) { zmq::message_t request; string protoBuf; socket.recv( &request ); uint16_t id = *( (uint16_t*)request.data() ); protoBuf = std::string( static_cast<char*>( request.data() + sizeof( uint16_t ) ), request.size() - sizeof( uint16_t ) ); InterfaceLayer::getInstance()->ParseProtoBufTable( protoBuf ); } Send(); usleep( 1 ); } void TransportLayer::Send() { zmq::message_t reply( 1 ); memcpy( reply.data(), "#", 1 ); socket.send( reply ); } 这是我编写的代码，最初旨在仅侦听一个客户端，现在我必须扩展它以侦听多个客户端。 我尝试使用zmq::multipart_t，但这需要C11支持，但我们使用的QNX版本不支持C11。 我尝试实施建议的解决方案。我创建了2个发布商，它们连接到相同的静态位置。 观察： I]执行顺序：1.启动订户2.启动Publisher1（它仅发布一个数据值） 订户未收到此数据。 II]修改了Publisher1以在while循环中发送相同的数据执行顺序：1.启动订户2.启动Publisher1 3.启动Publsiher2。 现在我看到订户正在从两个发布者处接收数据。 这表明存在数据丢失的可能性。 我如何确保绝对不会丢失数据？ 这是我的源代码： 发布者2： dummyFrontEnd::dummyFrontEnd():context(1),socket(context,ZMQ_PUB) { } void dummyFrontEnd::Init() { socket.connect("tcp://127.0.0.1:5555"); cout << "Connecting .... " << endl; } void dummyFrontEnd::SendData() { while ( std::getline(file, line_str) ) { std::stringstream ss(line_str); std::string direction; double tdiff; int i, _1939, pgn, priority, source, length, data[8]; char J, p, _0, dash, d; ss >> tdiff >> i >> J >> _1939 >> pgn >> p >> priority >> _0 >> source >> dash >> direction >> d >> length >> data[0] >> data[1] >> data[2] >> data[3] >> data[4] >> data[5] >> data[6] >> data[7]; timestamp += tdiff; while ( gcl_get_time_ms() - start_time < uint64_t(timestamp * 1000.0) - first_time ) { usleep(1); } if (arguments.verbose) { std::cout << timestamp << " " << i << " " << J << " " << _1939 << " " << pgn << " " << p << " " << priority << " " << _0 << " " << source << " " << dash << " " << direction << " " << d << " " << length << " " << data[0] << " " << data[1] << " " << data[2] << " " << data[3] << " " << data[4] << " " << data[5] << " " << data[6] << " " << data[7] << std::endl; } uint64_t timestamp_ms = (uint64_t)(timestamp * 1000.0); protoTable.add_columnvalues(uint64ToString(timestamp_ms)); /* timestamp */ protoTable.add_columnvalues(intToString(pgn)); /* PGN */ protoTable.add_columnvalues(intToString(priority)); /* Priority */ protoTable.add_columnvalues(intToString(source)); /* Source */ protoTable.add_columnvalues(direction); /* Direction */ protoTable.add_columnvalues(intToString(length)); /* Length */ protoTable.add_columnvalues(intToString(data[0])); /* data1 */ protoTable.add_columnvalues(intToString(data[1])); /* data2 */ protoTable.add_columnvalues(intToString(data[2])); /* data3 */ protoTable.add_columnvalues(intToString(data[3])); /* data4 */ protoTable.add_columnvalues(intToString(data[4])); /* data5 */ protoTable.add_columnvalues(intToString(data[5])); /* data6 */ protoTable.add_columnvalues(intToString(data[6])); /* data7 */ protoTable.add_columnvalues(intToString(data[7])); /* data8 */ zmq::message_t create_values(protoTable.ByteSizeLong()+sizeof(uint16_t)); *((uint16_t*)create_values.data()) = TABLEMSG_ID; // ID protoTable.SerializeToArray(create_values.data()+sizeof(uint16_t), protoTable.ByteSizeLong()); socket.send(create_values); protoTable.clear_columnvalues(); usleep(1); } } 发布者1： dummyFrontEnd::dummyFrontEnd():context(1),socket(context,ZMQ_PUB) { } void dummyFrontEnd::Init() { socket.connect("tcp://127.0.0.1:5555"); cout << "Connecting .... " << endl; } void dummyFrontEnd::SendData() { cout << "In SendData" << endl; while(1) { canlogreq canLogObj = canlogreq::default_instance(); canLogObj.set_fromhours(11); canLogObj.set_fromminutes(7); canLogObj.set_fromseconds(2); canLogObj.set_fromday(16); canLogObj.set_frommonth(5); canLogObj.set_fromyear(2020); canLogObj.set_tohours(12); canLogObj.set_tominutes(7); canLogObj.set_toseconds(4); canLogObj.set_today(17); canLogObj.set_tomonth(5); canLogObj.set_toyear(2020); zmq::message_t logsnippetmsg(canLogObj.ByteSizeLong() + sizeof(uint16_t)); *((uint16_t*)logsnippetmsg.data()) = 20; canLogObj.SerializeToArray(logsnippetmsg.data()+sizeof(uint16_t), canLogObj.ByteSizeLong()); socket.send(logsnippetmsg); usleep(1); canLogObj.clear_fromhours(); canLogObj.clear_fromminutes(); canLogObj.clear_fromseconds(); canLogObj.clear_fromday(); canLogObj.clear_frommonth(); canLogObj.clear_fromyear(); canLogObj.clear_tohours(); canLogObj.clear_tominutes(); canLogObj.clear_toseconds(); canLogObj.clear_today(); canLogObj.clear_tomonth(); canLogObj.clear_toyear(); } } 订户： TransportLayer::TransportLayer():context(1),socket(context,ZMQ_SUB){ } void TransportLayer::Init() { socket.bind("tcp://*:5555"); socket.setsockopt(ZMQ_SUBSCRIBE, "", 0); } void TransportLayer::Receive() { cout << "TransportLayer::Receive " << " I am in server " << endl; static int count = 1; // Producer thread. while ( true ){ zmq::message_t request; string protoBuf; socket.recv(&request); uint16_t id = *((uint16_t*)request.data()); cout << "TransportLayer : " << "request.data: " << request.data() << endl; cout << "TransportLayer : count " << count << endl; count = count + 1; cout << "TransportLayer : request.data.size " << request.size() << endl; protoBuf = std::string(static_cast<char*>(request.data() + sizeof(uint16_t)), request.size() - sizeof(uint16_t)); cout << "ProtoBuf : " << protoBuf << endl; InterfaceLayer *interfaceLayObj = InterfaceLayer::getInstance(); switch(id) { case TABLEMSG_ID: cout << "Canlyser" << endl; interfaceLayObj->ParseProtoBufTable(protoBuf); break; case LOGSNIPPET_ID: cout << "LogSnip" << endl; interfaceLayObj->ParseProtoBufLogSnippet(protoBuf); interfaceLayObj->logsnippetSignal(); // publish the signal break; default: break; } usleep(1); } } 我是ZMQ的新手。我有多个发布商和一个客户。寻求建议以最佳方式实施它。当前，它利用单个客户端和服务器的回复-请求模式； ...

我是ZMQ的新手。我有多个发布商和一个客户。寻求建议以最佳方式实施它。当前，它利用单个客户端和服务器的回复-请求模式； ...

为什么ZeroMQ服务器没有收到来自客户端的任何请求？

我有一个使用MQL4（类似于c ++的语言）设置的ZeroMQ服务器，以及一个使用Python设置的ZeroMQ客户端。由PUSH / PULL-sockets组成的连接效果很好，但是当我尝试使用REQ -...] >>

如何使Android使用ZeroMQ libzmq？

在Fargate中使用ZeroMQ Golang

我正在尝试在awsvpc模式下在Fargate上运行的ECS中使用ZeroMQ。我有2个不同的服务，每个服务都运行自己的任务并启用了服务发现功能。我在...

用户上的ZMQ接收不遵守设置的超时时间

我正在尝试将ZMQ通信集成到测试用例中，但是即使我为操作设置了超时500ms，zmq :: socket_t :: recv也存在无限期阻塞的问题。 ...