allen holub [email protected] @allenholubsocks5:// security: curvezmq.org •authentication...

Allen Holub http://holub.com [email protected]

@allenholub

1

2-1

Zero brokerZero latencyZero adminZero costZero waste

2-2

Basics

3-1

Basicszeromq.orgzguide.zeromq.orggithub.com/zeromq

3-2


High-volume financial apps

3-3

Basicszeromq.orgzguide.zeromq.orggithub.com/zeromqHigh-volume financial appsLinux/Windows/Mac/mobile…

3-4


High-volume financial appsLinux/Windows/Mac/mobile…C, C++, C# (.Net and Mono), Clojure, Delphi,Erlang, F#, Felix, Go, Haskell, Haxe, Java, Lua,Node.js, Objective-C, Perl, PHP, Python, Q,Racket, Ruby, Scala, Tcl, Ada, Basic, ooc, …

3-5

Lightweightzmq zmq_bind zmq_close zmq_connect zmq_cpp zmq_device zmq_pgm zmq_errno zmq_getsockopt zmq_init zmq_inproc zmq_ipc

zmq_msg_close zmq_msg_copy zmq_msg_data zmq_msg_init_data zmq_msg_init_size zmq_msg_init zmq_msg_move zmq_msg_size zmq_pgm zmq_poll zmq_recv zmq_send

zmq_setsockopt zmq_socket zmq_strerror zmq_tcp zmq_term zmq_version

4-1

Lightweightzmq zmq_bind zmq_close zmq_connect zmq_cpp zmq_device zmq_pgm zmq_errno zmq_getsockopt zmq_init zmq_inproc zmq_ipc

zmq_msg_close zmq_msg_copy zmq_msg_data zmq_msg_init_data zmq_msg_init_size zmq_msg_init zmq_msg_move zmq_msg_size zmq_pgm zmq_poll zmq_recv zmq_send

zmq_setsockopt zmq_socket zmq_strerror zmq_tcp zmq_term zmq_version

4-2

Transport

The code doesn't careCan move from inproc to a

separate VM with no changes

No server/setup: just launch your application and connect

5-1

Transportinproc://tcp://tipc://sctp://pgm://norm://socks5://




5-2

Transportinproc://tcp://tipc://sctp://pgm://norm://socks5://

Security: CurveZMQ.org• authentication• elliptic curve crypto




5-3

6-1

Not you grandfather’s socket

Really a “port” into messaging system.

6-2


Really a “port” into messaging system.Asynchronous front end to queue,

6-3


Really a “port” into messaging system.Asynchronous front end to queue,Multiple endpoints,

6-4


Really a “port” into messaging system.Asynchronous front end to queue,Multiple endpoints,Multiple transport protocols,

6-5


Really a “port” into messaging system.Asynchronous front end to queue,Multiple endpoints,Multiple transport protocols,Handles all threading for you,

6-6


Really a “port” into messaging system.Asynchronous front end to queue,Multiple endpoints,Multiple transport protocols,Handles all threading for you,Automatic connection/reconnection,

6-7


Really a “port” into messaging system.Asynchronous front end to queue,Multiple endpoints,Multiple transport protocols,Handles all threading for you,Automatic connection/reconnection,Hides work of implementing a messaging pattern.

6-8

7-1

7-2

© 2014 Allen I. Holub www.holub.com 8

REQ-REP (Client)

8-1

© 2014 Allen I. Holub www.holub.com

import org.zeromq.ZMQ;

public class HwClient {public static void main(…) {

ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket requester = context.socket(ZMQ.REQ);requester.connect("tcp://localhost:5555");

for (int req = 0; req != 10; req++) {String request = “Hello-“ + req;

requester.send(request.getBytes(), 0);

byte[] reply = requester.recv(0);}requester.close();context.term();

}}

8

REQ-REP (Client)

8-2


REQ-REP (Server)

9-1



public class HwServer {public static void main(…) throws Exception {

ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket responder = context.socket(ZMQ.REP);responder.bind("tcp://*:5555");

while (!Thread.currentThread().isInterrupted()) {byte[] request = responder.recv(0);System.out.println("Received Hello");

Thread.sleep(1000); // Simulate “work”

String reply = "World";responder.send(reply.getBytes(), 0);

}responder.close();context.term();

}}

9

REQ-REP (Server)

9-2

Pub/Sub Broker

Publisher Subscriber(s)

10-1

Pub/Sub Broker

PublisherSubscriber(s)

Topic

10-2

Pub/Sub Broker

Publisher Subscriber(s)

Topic

10-3

11-1

11-2


PUB-SUB (Publisher)

12-1



public class Publisher {

public static void main (…) throws Exception {ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket publisher = context.socket(ZMQ.PUB);publisher.bind("tcp://*:5556");publisher.bind("ipc://publisherID");

while(!Thread.currentThread().isInterrupted()) {Thread.currentThread.sleep(1000) // “work”String update = “messageID messageData”publisher.send(update, 0);

}

publisher.close ();context.term ();

}}

12

PUB-SUB (Publisher)

12-2


PUB-SUB (Subscriber)

13-1



public class Subscriber {

public static void main (…) {ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket subscriber = context.socket(ZMQ.SUB);subscriber.connect("tcp://localhost:5556");

subscriber.subscribe("messageID".getBytes());

for (int i = 100; --i>0; ) {String string = subscriber.recvStr(0).trim();…

}

subscriber.close();context.term();

}}

13

PUB-SUB (Subscriber)

13-2

while( !Thread.currentThread ().isInterrupted ()) { publisher.sendMore ("A"); publisher.send ("We don't want to see this"); publisher.sendMore ("B"); publisher.send("We would like to see this");}

subscriber.subscribe(“B".getBytes());while (!Thread.currentThread ().isInterrupted ()) {

String contents = subscriber.recvStr (); ...}

14

15-1

15-2

© 2014 Allen I. Holub www.holub.com 16PUSH-PULL (Ventilator)16-1


public class Task {public static void main (...) throws Exception {

ZMQ.Context context = ZMQ.context(1);ZMQ.Socket worker = context.socket(ZMQ.PUSH);worker.bind("tcp://*:" + Ports.WORKER);ZMQ.Socket sink = context.socket(ZMQ.PUSH);sink.connect("tcp://localhost:" + Ports.SINK);ZMQ.Socket done = context.socket(ZMQ.SUB);done.connect( "tcp://localhost:" + Ports.DONE );done.subscribe("DONE ".getBytes());

// Handle "slow-joiner" problemSystem.out.println("Hit Enter when workers ready");System.in.read();

sink.send("0", 0); // start batchfor ( int i = 100; --i >= 0; )

worker.send( "someWork" + i , 0);String result = done.recvStr(0).trim();

sink.close(); worker.close(); done.close();context.term();

}} 16PUSH-PULL (Ventilator)

16-2


PUSH-PULL (Worker)

17-1


import org.zeromq.ZMQ;public class Worker {

public static void main (...) throws Exception {ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket worker = context.socket(ZMQ.PULL);task.connect(“tcp://localhost:" + Ports.WORKER);

ZMQ.Socket sink = context.socket(ZMQ.PUSH);sink.connect(“tcp://localhost:" + Ports.SINK);

while(!Thread.currentThread().isInterrupted()) {String work = new String(worker.recv(0)).trim();String result = doTheWork( work );sink.send(result.getBytes(), 0);

}

sink.close(); worker.close(); context.term();}...

}17

PUSH-PULL (Worker)

17-2


PUSH-PULL (Sink)

18-1


import org.zeromq.ZMQ;public class TaskSink {

public static void main (…) throws Exception {ZMQ.Context context = ZMQ.context(1);

ZMQ.Socket sink = context.socket(ZMQ.PULL);receiver.bind("tcp://*:" + Ports.SINK );

ZMQ.Socket done = context.socket(ZMQ.PUB);publisher.bind("tcp://*:" + Ports.DONE )

String result = new String( receiver.recv(0) );for( int i = 100; --i >= 0 ) {

result = update(result, sink.recv(0) )}done.send("DONE result", 0);

sink.close(); done.close(); context.term();}...

}18

PUSH-PULL (Sink)

18-2

REQ/REPProxy (Broker)

19-1

REQ/REPProxy (Broker)

19-2

import org.zeromq.ZMQ; import org.zeromq.ZMQ.Context; import org.zeromq.ZMQ.Poller; import org.zeromq.ZMQ.Socket;

public class rrbroker{

public static void main (String[] args) { Context context = ZMQ.context(1);

Socket frontend = context.socket(ZMQ.ROUTER); Socket backend = context.socket(ZMQ.DEALER); frontend.bind("tcp://*:5559"); backend.bind("tcp://*:5560");

Poller items = new Poller (2); items.register(frontend, Poller.POLLIN); items.register(backend, Poller.POLLIN);

boolean more = false; byte[] message;

Simple Broker

20-1

boolean more = false; byte[] message;

while(!Thread.currentThread().isInterrupted()) { items.poll();

if (items.pollin(0)) { while (true) { // receive message message = frontend.recv(0); more = frontend.hasReceiveMore();

// Broker it backend.send(message, more ? ZMQ.SNDMORE : 0); if(!more){ break; } } } if (items.pollin(1)) { while (true) {20-2 } } } if (items.pollin(1)) { while (true) { // receive message message = backend.recv(0); more = backend.hasReceiveMore(); // Broker it frontend.send(message, more ? ZMQ.SNDMORE : 0); if(!more){ break; } } } } // We never get here but clean up anyhow frontend.close(); backend.close(); context.term(); }}

20-3

public class msgqueue{

public static void main (String[] args) { // Prepare our context and sockets Context context = ZMQ.context(1);

// Socket facing clients Socket frontend = context.socket(ZMQ.ROUTER); frontend.bind("tcp://*:5559");

// Socket facing services Socket backend = context.socket(ZMQ.DEALER); backend.bind("tcp://*:5560");

// Start the proxy ZMQ.proxy (frontend, backend, null);

// We never get here but clean up anyhow frontend.close(); backend.close(); context.term(); }}

The proxy() method

21-1public class msgqueue{

public static void main (String[] args) { // Prepare our context and sockets Context context = ZMQ.context(1);

// Socket facing clients Socket frontend = context.socket(ZMQ.ROUTER); frontend.bind("tcp://*:5559");

// Socket facing services Socket backend = context.socket(ZMQ.DEALER); backend.bind("tcp://*:5560");

// Start the proxy ZMQ.proxy (frontend, backend, null);

// We never get here but clean up anyhow frontend.close(); backend.close(); context.term(); }}

21-2

Pub/SubProxy (Broker)

22-1

Pub/SubProxy (Broker)

22-2

Threading“Workers” (i.e. tasks) run on one thread. They wait on a socket at the top of an “event” loop.

Send them a message to start them up.

Wait for a really (pub/sub or polling) to get result.

Send them data via json or equivalent

23

Inter-thread signaling

Pair sockets: talk to only one socket are very efficient

24


import org.zeromq.ZMQ;import org.zeromq.ZMQ.Context;import org.zeromq.ZMQ.Socket;

public class mtrelay{ private static class Step1 extends Thread { private Context context; private Step1 (Context context) { this.context = context; }

@Override public void run(){ // Signal downstream to step 2 Socket xmitter = context.socket(ZMQ.PAIR); xmitter.connect("inproc://step2"); System.out.println("S1 ready, start S2"); xmitter.send("READY", 0); xmitter.close (); }

} private static class Step2 extends Thread

25

25-1


} private static class Step2 extends Thread { private Context context; private Step2 (Context context){ this.context = context; } @Override public void run(){ Socket receiver = context.socket(ZMQ.PAIR); receiver.bind("inproc://step2"); Thread step1 = new Step1 (context); step1.start();

receiver.recv(0); receiver.close ();

Socket xmitter = context.socket(ZMQ.PAIR); xmitter.connect("inproc://step3"); xmitter.send("READY", 0);

xmitter.close (); } } 25

25-2


xmitter.connect("inproc://step3"); xmitter.send("READY", 0);

xmitter.close (); } }

public static void main (String[] args) {

Context context = ZMQ.context(1); Socket receiver = context.socket(ZMQ.PAIR); receiver.bind("inproc://step3"); Thread step2 = new Step2 (context); step2.start(); receiver.recv(0); receiver.close (); System.out.println ("Test successful!"); context.term (); }} 25

25-3

Req

Router

Under-the-coversframes

26

LoadBalancing

Asks for some worker to do something

Asks for something to do

27

28-1

28-2

Added by REQ socket

28-3

Added by REQ socket

Added by ROUTER socket

28-4

Added by REQ socket


28-5

Added by REQ socket


28-6


import java.util.LinkedList;import java.util.Queue;

import org.zeromq.ZMQ;import org.zeromq.ZMQ.Context;import org.zeromq.ZMQ.Poller;import org.zeromq.ZMQ.Socket;

public class lbbroker {

private static final int NBR_CLIENTS = 10; private static final int NBR_WORKERS = 3;

private static class ClientTask extends Thread { public void run() { Context context = ZMQ.context(1);

Socket client = context.socket(ZMQ.REQ); ZHelper.setId(client); // Make ID printable

client.connect("ipc://frontend.ipc");29

29


private static final int NBR_CLIENTS = 10; private static final int NBR_WORKERS = 3;

private static class ClientTask extends Thread { public void run() { Context context = ZMQ.context(1);

Socket client = context.socket(ZMQ.REQ); ZHelper.setId(client); // Make ID printable

client.connect("ipc://frontend.ipc");

client.send("HELLO"); String reply = client.recvStr (); System.out.println("Client: " + reply);

client.close(); context.term(); } }

private static class WorkerTask extends Thread { public void run() {

30

30


private static class WorkerTask extends Thread { public void run() { Context context = ZMQ.context(1); Socket worker = context.socket(ZMQ.REQ); ZHelper.setId (worker); // use Printable ID

worker.connect("ipc://backend.ipc"); worker.send("READY"); while(!Thread.currentThread ().isInterrupted ()) { String address = worker.recvStr (); String empty = worker.recvStr (); assert (empty.length() == 0);

String request = worker.recvStr (); System.out.println("Worker: " + request);

worker.sendMore (address); worker.sendMore (""); worker.send("OK"); } worker.close (); context.term (); } }

31

31


worker.sendMore (""); worker.send("OK"); } worker.close (); context.term (); } }

public static void main (String[] args) { Context context = ZMQ.context(1); Socket frontend = context.socket(ZMQ.ROUTER); Socket backend = context.socket(ZMQ.ROUTER); frontend.bind("ipc://frontend.ipc"); backend. bind("ipc://backend.ipc");

int clientNbr = 0; for(; clientNbr < NBR_CLIENTS; clientNbr++) new ClientTask().start();

for (int workerNbr = 0; workerNbr < NBR_WORKERS; workerNbr++) new WorkerTask().start();

Queue<String> available = new LinkedList<String>(); 32

32


Socket frontend = context.socket(ZMQ.ROUTER); Socket backend = context.socket(ZMQ.ROUTER); frontend.bind("ipc://frontend.ipc"); backend. bind("ipc://backend.ipc");

int clientNbr = 0; for(; clientNbr < NBR_CLIENTS; clientNbr++) new ClientTask().start();

for (int workerNbr = 0; workerNbr < NBR_WORKERS; workerNbr++) new WorkerTask().start();

Queue<String> available = new LinkedList<String>();

while (!Thread.currentThread().isInterrupted()) { Poller items = new Poller (2);

items.register(backend, Poller.POLLIN); if(available.size() > 0) // workers are available items.register(frontend, Poller.POLLIN);

if (items.poll() < 0) break;

if (items.pollin(0)) { // Handle backend

33

33


if (items.pollin(0)) { // Handle backend available.add (backend.recvStr ());

// Second frame better be empty String empty = backend.recvStr (); assert (empty.length() == 0);

// Third frame is READY or a client address String clientAddr = backend.recvStr ();

// If client reply, send rest back to frontend if (!clientAddr.equals("READY")) { empty = backend.recvStr (); assert (empty.length() == 0);

String reply = backend.recvStr (); frontend.sendMore(clientAddr); frontend.sendMore(""); frontend.send(reply);

if (--clientNbr == 0) break; } }

34

34


if (items.pollin(1)) { // Get next client request, route to LRU worker // Client request is [address][empty][request] String clientAddr = frontend.recvStr ();

String empty = frontend.recvStr (); assert (empty.length() == 0);

String request = frontend.recvStr ();

String workerAddr = available.poll();

backend.sendMore (workerAddr); backend.sendMore (""); backend.sendMore (clientAddr ); backend.sendMore (""); backend.send (request); } } frontend.close(); backend.close(); context.term(); }}

35

35

Mongrel236-1

Mongrel236-2

Mongrel236-3

Issues

37-1

No built-in durability or guaranteed deliveryIssues

37-2

No built-in durability or guaranteed deliveryNo brokers

but you can easily build your own;learn more from http://zguide.zeromq.org

Issues

37-3



Issues

37-4



Not thread safedo not pass sockets between threads

Issues

37-5

allen holub [email protected] @allenholubsocks5:// security: curvezmq.org •authentication...

Documents