6
\$\begingroup\$

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.

\$\endgroup\$
3
\$\begingroup\$

You're only sending one ZMsg per connection. Establishing TCP connections is generally considered to be a high-overhead operation. If done over a network, it would involve a three-way handshake to connect and a two-way handshake to disconnect. When done on localhost, the TCP protocol is still somewhat complex, even though the transmission latency is nearly small.

Compare that to this example, which reuses the same connection for multiple messages.

I can understand disconnecting if you have nothing to transmit for a while. However, if you are trying to push as many messages as possible in a tight loop, it makes no sense to make a new connection per message — it defeats the advantage of using ZeroMQ rather than raw TCP in the first place.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.