I am working on multithreading project and I am using ZeroMQ socket to send the data as a byte array. Below is my example which works fine. I'm opting for code review to see whether it can be improved by adding a wrapper class for using ZeroMQ socket as some sort of socket manager.
public class TestDriver {
private static final int THREADS = 5;
private static final long DURATION_OF_RUN = 10;
private static final ExecutorService executor = Executors.newFixedThreadPool(THREADS);
public static void main(String[] args) {
try {
long startTime = System.currentTimeMillis();
long endTime = startTime + (DURATION_OF_RUN * 60 * 1000);
for (int i = 0; i < THREADS; i++) {
executor.submit(new ClientTask(endTime));
}
// wait for termination
executor.shutdown();
executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
And below is ClientTask class -
public class ClientTask implements Runnable {
private static final Random random = new Random();
private static final String CHARACTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
private final long endTime;
public ClientTask(long endTime) {
this.endTime = endTime;
}
@Override
public void run() {
while (System.currentTimeMillis() <= endTime) {
// does the way I am using ZeroMQ socket here is efficient?
// or we should make connection pooling of sockets using zeromq
// inside any singleton wrapper class?
ZMQ.Socket socket = SocketsManager.getInstance().getSockets();
byte[] byteArray = generateRandomStringAsBytes(20);
ZMsg req = new ZMsg();
req.add(byteArray);
req.send(socket);
socket.close();
context.term();
}
}
/**
* A simple method which will generate random string and return as bytes
*
* @param length
* @return
*/
public static byte[] generateRandomStringAsBytes(int length) {
return generateRandomString(length).getBytes(Charset.forName("UTF-8"));
}
}
And below is my SocketManager class -
public class SocketsManager {
private ZMQ.Context context = ZMQ.context(1);
private static Random rand = new Random(System.nanoTime());
private static class SocketsHolder {
static final SocketsManager INSTANCE = new SocketsManager();
}
public static SocketsManager getInstance() {
return SocketsHolder.INSTANCE;
}
private SocketsManager() {
// should I do anything here?
}
// does below method is right?
public Socket getSockets() {
ZMQ.Socket socket = context.socket(ZMQ.PUSH);
String identity = String.format("%04X-%04X", rand.nextInt(), rand.nextInt());
socket.setIdentity(identity.getBytes());
socket.connect("tcp://localhost:5700");
return socket;
}
}
Is the way I am using ZeroMQ socket is efficient here? Or do I need to create connection pool of sockets?
As far as I know ZeroMQ sockets must be used on the threads where it has been created and I believe one ZeroMQ context is usually enough for the application? I read the ZeroMq guide and I stumbled upon the following:
First, do not try to use the same socket from multiple threads. Please don't explain why you think this would be excellent fun, just please don't do it. Next, you need to shut down each socket that has ongoing requests. The proper way is to set a low LINGER value (1 second), and then close the socket. If your language binding doesn't do this for you automatically when you destroy a context, I'd suggest sending a patch.
Finally, destroy the context. This will cause any blocking receives or polls or sends in attached threads (i.e., which share the same context) to return with an error. Catch that error, and then set linger on, and close sockets in that thread, and exit. Do not destroy the same context twice. The zmq_ctx_destroy in the main thread will block until all sockets it knows about are safely closed.
Am I following the above rules in my above code? If not, then how can I achieve the above thing in my code? Since this is my first time working with sockets directly so not able to understand what is the right way to add a wrapper around ZeroMQ socket.
Do I need to create connection pool of sockets? As in this example they are using same socket connection for same thread to send the data.
