Return to Question

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.

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.

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.

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.Context context = ZMQ.context(1);
 ZMQ.Socket socket = context.socket(ZMQ.PUSH);
 // in general, I will have more connection here, 
 // as of now I have only one
 socket.connect("tcp://ip_address_1:10000");

 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"));
 }
}

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?

How do I provide a wrapper class around ZeroMQ sockets so that it is efficient and thread safe as well? In general what I have seen is we always have a separate class which makes the connection and then you get the instance of that class and start using it in other class. How can I do the same thing with ZeroMQ socket as well?

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.

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;
 }
}

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.