spotify / netty-zmtp Goto Github PK

A Netty implementation of ZMTP, the ZeroMQ Message Transport Protocol.

License: Apache License 2.0

Java 100.00%

netty-zmtp's Introduction

Netty-zmtp

Note This project has been discontinued.

This is a ZeroMQ codec for Netty that aims to implement ZMTP, the ZeroMQ Message Transport Protocol versions 1.0 and 2.0 as specified in http://rfc.zeromq.org/spec:13 and http://rfc.zeromq.org/spec:15.

This project is hosted on https://github.com/spotify/netty-zmtp/

At Spotify we use ZeroMQ for a lot of the internal communication between services in the backend. As we implement more services on top of the JVM we felt the need for more control over the state of TCP connections, routing, message queue management, etc as well as getting better performance than seems to be currently possible with the JNI based JZMQ library.

This project implements the ZMTP wire protocol but not the ZeroMQ API, meaning that it can be used to communicate with other peers using e.g. ZeroMQ (libzmq) but it's not a drop-in replacement for JZMQ like e.g. JeroMQ attempts to be. For an example of how a ZeroMQ socket equivalent might be implemented using the netty-zmtp codecs, see the ZMTPSocket class in the tests.

We have successfully used these handlers to implement services capable of processing millions of messages per second.

Currently this project targets Java 6+ and Netty 4.x. It does not have any native dependency on e.g. libzmq.

Usage

To use netty-zmtp, insert a ZMTPCodec instance into your channel pipeline.

ch.pipeline().addLast(ZMTPCodec.of(ROUTER));

Upstream handlers will receive ZMTPMessage instances.

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 final ZMTPMessage message = (ZMTPMessage) msg;
 // ...
}

Wait for the ZMTP handshake to complete before sending messages.

@Override
public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt)
    throws Exception {
  if (evt instanceof ZMTPHandshakeSuccess) {
    // ...
  }
}

`pom.xml`

<dependency>
  <groupId>com.spotify</groupId>
  <artifactId>netty4-zmtp</artifactId>
  <version>0.4.1</version>
</dependency>

Performance

Performance seems decent, with the throughput benchmark producing 7M+ messages/s throughput numbers on a recent laptop.

For maximum throughput, look into using the BatchFlusher to opportunistically gather writes into fewer syscalls.

Truly overhead conscientious users might want to look into implementing the ZMTPEncoder and ZMTPDecoder interfaces for eliminating the ZMTPMessage intermediary when reading/writing application messages.

Benchmarks

Preparation

First compile and run tests:

mvn clean test

Fetch dependencies:

mvn dependency:copy-dependencies

Now benchmarks are ready to run.

One-Way Throughput

java -cp 'target/classes:target/test-classes:target/dependency/*' \
     com.spotify.netty4.handler.codec.zmtp.benchmarks.ThroughputBenchmark

1s:    3,393,219 messages/s.    (total:    3,399,036)
2s:    6,595,711 messages/s.    (total:   10,007,250)
3s:    7,158,176 messages/s.    (total:   17,154,830)
4s:    7,313,554 messages/s.    (total:   24,461,521)
5s:    7,294,709 messages/s.    (total:   31,790,260)
6s:    7,282,308 messages/s.    (total:   39,064,152)
7s:    7,294,591 messages/s.    (total:   46,336,682)

Req/Rep Throughput

java -cp 'target/classes:target/test-classes:target/dependency/*' \
     com.spotify.netty4.handler.codec.zmtp.benchmarks.ReqRepBenchmark

1s:      444,848 requests/s.    (total:      446,249)
2s:    1,251,304 requests/s.    (total:    1,699,916)
3s:    1,241,569 requests/s.    (total:    2,941,709)
4s:    1,365,408 requests/s.    (total:    4,307,949)
5s:    1,379,640 requests/s.    (total:    5,681,522)
6s:    1,379,048 requests/s.    (total:    7,064,183)
7s:    1,377,180 requests/s.    (total:    8,438,673)

Req/Rep With Custom Encoder/Decoder Throughput

java -cp 'target/classes:target/test-classes:target/dependency/*' \
     com.spotify.netty4.handler.codec.zmtp.benchmarks.CustomReqRepBenchmark

1s:      443,337 requests/s.         1.470 ms avg latency.    (total:      445,512)
2s:    1,306,539 requests/s.         0.765 ms avg latency.    (total:    1,747,153)
3s:    1,549,594 requests/s.         0.645 ms avg latency.    (total:    3,303,268)
4s:    1,557,397 requests/s.         0.642 ms avg latency.    (total:    4,859,727)
5s:    1,618,137 requests/s.         0.618 ms avg latency.    (total:    6,472,114)
6s:    1,609,406 requests/s.         0.621 ms avg latency.    (total:    8,084,958)
7s:    1,611,349 requests/s.         0.621 ms avg latency.    (total:    9,692,777)
8s:    1,611,672 requests/s.         0.620 ms avg latency.    (total:   11,306,988)

Feedback

There is an open Google group for general development and usage discussion available at https://groups.google.com/group/netty-zmtp

We use the github issue tracker at https://github.com/spotify/netty-zmtp/issues

License

This software is licensed using the Apache 2.0 license. Details in the file LICENSE

netty-zmtp's People

Contributors

Stargazers

Watchers

netty-zmtp's Issues

ZmtpSocketType PUB/SUB Enum is mixed up.

see http://rfc.zeromq.org/spec:15

Support netty 4

More like a question than issue. Is support for netty4 planned?

Does this project still alive?

Does this project still alive and used at Spotify?
Last update was 4 years ago...

Is this project published in Maven Central

A search on Maven central did not provide any similar artifacts - can you please confirm or publish it so that we do not have to maintain local repos?

A complete example

hello, i have studied the netty-zmtp,it's perfect. I was wondering can the project provide a complete example?Thank you!

Question: Can you use communicate with both the server and client as netty-zmtp?

My server pipeline looks like below:

    serverBootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            newCachedThreadPool(), newCachedThreadPool()));
    serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        private final Executor executor = new OrderedMemoryAwareThreadPoolExecutor(
                getRuntime().availableProcessors(), 1024 * 1024, 128 * 1024 * 1024);

        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new ExecutionHandler(executor), new ZMTP20Codec(new ZMTPSession(
                    Addressed, 1024, null, ROUTER), false), new ServerHandler());
        }
    });

My Client pipeline looks below:

    clientBootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
            newCachedThreadPool(), newCachedThreadPool()));
    clientBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new ZMTP20Codec(new ZMTPSession(Addressed, 1024), false),
                    new OneToOneDecoder() {
                        @Override
                        protected Object decode(ChannelHandlerContext ctx, Channel channel, Object msg)
                                throws Exception {
                            System.out.println(msg);
                            return null;
                        }
                    });
        }
    });

However if I try to send a message to server, I get the following exception

 Caused by: java.lang.IllegalArgumentException: unsupported message type: class com.spotify.netty.handler.codec.zmtp.ZMTPMessage
    at org.jboss.netty.channel.socket.nio.SocketSendBufferPool.acquire(SocketSendBufferPool.java:52)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.write0(AbstractNioWorker.java:193)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.writeFromTaskLoop(AbstractNioWorker.java:151)
    at org.jboss.netty.channel.socket.nio.AbstractNioChannel$WriteTask.run(AbstractNioChannel.java:335)
    at org.jboss.netty.channel.socket.nio.AbstractNioSelector.processTaskQueue(AbstractNioSelector.java:372)
    at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:296)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89)
    at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
    at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
    at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:744)

Can you please help?

Use more modern ZMTP protocol

Sorry to use the issue tracker for this, I didn't see any other way to contact the netty-zmtp team. It is probably a Bad Idea to implement the ZMTP 1.0 protocol in new code. This has no versioning and makes it very hard to upgrade later on.

There is at least a versioned spec here: http://rfc.zeromq.org/spec:15 (ZMTP 2.0).

There is also a new draft 3.0 protocol, http://rfc.zeromq.org/spec:23, which has various levels of security.

If you at least implement ZMTP 2.0 (and it's a trivial change to the signature), other ZMTP implementations will interoperate with netty-zmtp safely.

I'm more than happy to help in any way I can.