我在我的 Java 应用程序中使用 ZMQ。我发现它的行为不均匀,即,如果我向一位消费者发送大约 100 条消息,说它需要 1 秒,那么如果我们继续增加消费者,则所用时间变为 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;
}