0

Java アプリケーションで ZMQ を使用しています。私はそれが不均一に動作することを発見しました.1つの消費者が1秒かかると言う100メッセージを送信すると、消費者を増やし続けると所要時間が2,1.5,3になります。段階的な増減はありません。どうすればこれを修正できますか。以下の私のコードを探してください

// Broker

import org.zeromq.ZMQ; import org.zeromq.ZMQ.Context; import org.zeromq.ZMQ.Socket; import org.zeromq.ZMQStreamer;

public class Broker {

/**
 * @param args
 */
public static void main(String[] args) 
{
    Context context = ZMQ.context(1);

    Socket frontEnd = context.socket(ZMQ.PULL);
    frontEnd.bind("tcp://*:5555");

    Socket backEnd = context.socket(ZMQ.PUSH);
    backEnd.bind("tcp://*:5560");

    ZMQStreamer zmqStreamer = new ZMQStreamer(context, frontEnd, backEnd);
    zmqStreamer.run();
}

}

// Producer

import org.zeromq.ZMQ; import org.zeromq.ZMQ.Socket;

public class Producer {

public void init()
{
    ZMQ.Context context = ZMQ.context(1);
    socket = context.socket(ZMQ.PUSH);
    socket.connect("tcp://localhost:5555");
}

public void initMessage(String message)
{
    this.message = message;
}

public void sendMessage()
{
    String sendMessage = System.nanoTime() +"#"+ message;
    socket.send(sendMessage.getBytes(), 0);
}
/**
 * @param args
 */
public static void main(String[] args) 
{
    Producer producer = new Producer();
    producer.init();
    byte[] message = new byte[Integer.parseInt(args[0])];
    //message = "Hello".getBytes();
    producer.initMessage(new String(message));
    for(int i=0;i<100;i++)
    {
        producer.sendMessage();
    }
}

private Socket socket = null;
private String message;
}

//Consumer

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;

public class Consumer 
{

public void init()
{
    ZMQ.Context context = ZMQ.context(1);
    socket = context.socket(ZMQ.PULL);
    socket.connect("tcp://localhost:5560");
}

public void reciveMessage()
{
    byte[] recived = socket.recv(0);
    //System.out.println(recived.length);
    long recivedTime = System.nanoTime();
    String message = new String(recived);
    String[] splitMessage = message.split("#");
    long sendTime = Long.parseLong(splitMessage[0]);
    System.out.println("Send Time " + sendTime + " RecivedTime "
            + recivedTime + " Time taken " + (recivedTime - sendTime)
            + " Message " + message);
}
/**
 * @param args
 */
public static void main(String[] args) 
{
    Consumer consumer = new Consumer();
    consumer.init();
    for (int i=0;i<100;i++)
    {
        consumer.reciveMessage();
    }
}
private Socket socket = null;
}
4

1 に答える 1

1

マルチスレッド コードの一部の時間を確実に計るには、コレクター/シンクで開始時刻と終了時刻を同期する何らかの方法が必要になります (現在、プログラムされていることは示されていません)。

ZMQ ガイドのこの例を確認してください。この例では、データセットを分割する ZMQ の正しい方法の 1 つとして次のプロセスが述べられています。

...

私たちのスーパーコンピューティング アプリケーションは、かなり典型的な並列処理モデルです。

We have a ventilator that produces tasks that can be done in parallel.
We have a set of workers that process tasks.
We have a sink that collects results back from the worker processes.

...

于 2012-08-07T11:47:46.557 に答える