1

我正在测试 ZeroMQ，我每秒只能收到大约 1227 - 1276 条消息。然而，我读过这些应该超过这个数量的 100 倍。

我究竟做错了什么？我可以指定一些配置来解决这个问题吗？

我正在使用以下功能：

public static final String SERVER_LOCATION = "127.0.0.1";
public static final int SERVER_BIND_PORT = 5570;

public static void receiveMessages() throws InvalidProtocolBufferException, FileNotFoundException, UnsupportedEncodingException{
    ZContext ctx = new ZContext();

    Socket frontend = ctx.createSocket(ZMQ.PULL);
    frontend.bind("tcp://*:"+SERVER_BIND_PORT);

    int i = 1;
    do{
        ZMsg msg = ZMsg.recvMsg(frontend);
        ZFrame content = msg.pop();
        if(content!= null){
            msg.destroy();
            System.out.println("Received: "+i);
            i++;
            content.destroy();
        }
    }while(true);
}

public static void sendMessages() throws FileNotFoundException, UnsupportedEncodingException{
    ZContext ctx = new ZContext();
    Socket client = ctx.createSocket(ZMQ.PUSH);

    client.setIdentity("i".getBytes());
    client.connect("tcp://"+SERVER_LOCATION+":"+SERVER_BIND_PORT);

    PollItem[] items = new PollItem[] { new PollItem(client, Poller.POLLIN) };
    int i = 1;
    Timer t = new Timer(timeToSpendSending);
    t.start();
    do{
        client.send(/* object to send*/ , 0);
        i++;
    }while(!t.isDone());

    System.out.println("Done with "+i);
}

用于限制程序运行时间的计时器类：

class Timer extends Thread{
    int time;
    boolean done;
    public Timer(int time){
        this.time = time;
        done = false;
    }
    public void run(){
        try {
            this.sleep(time);
            done = true;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public boolean isDone(){
        return done;
    }
}

编辑：我正在使用 jeroMQ

<dependency>
    <groupId>org.zeromq</groupId>
    <artifactId>jeromq</artifactId>
    <version>0.3.4</version>
</dependency>
4

1 回答 1

0

我不得不更换连接方法并删除了高水位标记（设置为 0 表示内存中的无限消息）

代码如下：

public static final String SERVER_LOCATION = "127.0.0.1";
public static final int SERVER_BIND_PORT = 5570;
public static final String TOPIC = "topic1";

public static void receiveMessages() throws InvalidProtocolBufferException, FileNotFoundException, UnsupportedEncodingException{
    // Prepare our context and subscribe
       Context context = ZMQ.context(1);
       Socket subscriber = context.socket(ZMQ.SUB);

       subscriber.connect("tcp://"+SERVER_LOCATION+":"+SERVER_BIND_PORT);
       subscriber.setRcvHWM(0);
       subscriber.subscribe(TOPIC.getBytes());
       System.out.println("subscribed to  "+TOPIC);
       int i = 1;
       boolean started = false;
       Timer t = new Timer(timeToSpendSending);
       do{
           String msg = subscriber.recvStr();
           if(!TOPIC.equals(msg)){
               if(!started){
                   t.start();
                   started = true;
               }
               i++;
           }
       }while(!t.isDone());
       System.out.println("Done with: "+i);
       subscriber.close();
       context.term();
   }
   public static void sendMessages() throws FileNotFoundException, UnsupportedEncodingException{
       Context context = ZMQ.context(1);
       Socket publisher = context.socket(ZMQ.PUSH);
       publisher.bind("tcp://"+SERVER_LOCATION+":"+SERVER_BIND_PORT);
       publisher.setHWM(0);
       publisher.setSndHWM(0);

       int i = 1;
       Timer t = new Timer(timeToSpendSending);
       t.start();
       do{
          publisher.sendMore(TOPIC);
          publisher.send("Test Data number "+i);
          i++;
      }while(!t.isDone());
      System.out.println("Done with: "+i);
      publisher.close();
      context.term();
   }

像这样，我得到的消息计数范围在发送时每秒 250,000 条和接收时每秒 145,000 条。

于 2014-09-17T10:02:20.767 回答