1

ZeroMQ をテストしていますが、1 秒あたり約 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 に答える