1

所以我正在使用 jzmq GIT master 分支和 ZeroMQ 3.2.3 编写自己的东西。

安装后，我尝试测试以下简单PUB/SUB程序，其中发布者和订阅者在一个进程中交谈。由于测试是在windows下，所以我使用了TCP。

 public class ZMQReadynessTest {

     private ZMQ.Context context;

     @Before
     public void setUp() {
         context = ZMQ.context(1);
     }

     @Test
     public void testSimpleMessage() {
         String topic = "tcp://127.0.0.1:31216";
         final AtomicInteger counter = new AtomicInteger();

      // _____________________________________ create a simple subscriber

         final ZMQ.Socket subscribeSocket = context.socket(ZMQ.SUB);
         subscribeSocket.connect(topic);
         subscribeSocket.subscribe("TestTopic".getBytes());

         Thread subThread = new Thread() {

             @Override
             public void run() {
                 while (true) {
                     String value = null;

                  // This would result in trouble /\/\/\/\/\/\/\/\/\
                     {
                         ByteBuffer buffer = ByteBuffer.allocateDirect(100);
                         if (subscribeSocket.recvZeroCopy( buffer,
                                                           buffer.remaining(),
                                                           ZMQ.DONTWAIT
                                                           ) > 0 ) {

                             buffer.flip();

                             value = buffer.asCharBuffer().toString();
                             System.out.println(buffer.asCharBuffer().toString());
                         }
                     }

                  // This works perfectly + + + + + + + + + + + + + 
                     /*
                     {
                         byte[] bytes = subscribeSocket.recv(ZMQ.DONTWAIT);
                         if (bytes == null || bytes.length == 0) {
                             continue;
                         }

                         value = new String(bytes);
                     }
                      */

                     if (value != null && value.length() > 0) {
                         counter.incrementAndGet();
                         System.out.println(value);
                         break;
                     }
                 }
             }
         };
         subThread.start();

      // _____________________________ create a simple publisher

         ZMQ.Socket publishSocket = context.socket(ZMQ.PUB);
         publishSocket.bind("tcp://*:31216");

         try {
             Thread.sleep(3000); //  + wait 3 sec to make sure its ready
         } catch (InterruptedException e) {
             e.printStackTrace();
             fail();
         }

      // publish a sample message
         try {
             publishSocket.send("TestTopic".getBytes(), ZMQ.SNDMORE);
             publishSocket.send("This is test string".getBytes(), 0);
             subThread.join(100);
         } catch (InterruptedException e) {
             e.printStackTrace();
             fail();
         }

         assertTrue(counter.get() > 0);
         System.out.println(counter.get());
     }
 }

现在如您所见，在订阅者中，如果我使用一种简单的.recv(ZMQ.DONTWAIT)方法，它就可以完美运行。但是，如果我使用的是直接字节缓冲区，我什么也没有返回 - 我得到了以下异常，似乎在程序退出时：

Exception in thread "Thread-0" org.zeromq.ZMQException: Resource temporarily unavailable(0xb)
at org.zeromq.ZMQ$Socket.recvZeroCopy(Native Method)
at ZMQReadynessTest$1.run(ZMQReadynessTest.java:48)

我还尝试使用一个简单的ByteBuffer（不是直接缓冲区），它不会抛出上面的异常；但也没有回报我。

有人知道如何解决上述问题吗？

我不想byte[]到处创建对象，因为我正在做一些高性能系统。如果这无法解决，我可能会简单地使用 Unsafe 代替。但我真的很想以“假定的方式”工作。

提前致谢。

亚历克斯

4

0 回答 0