ZeroMQ Is The Answer

e r 3 e r
s w /2
2
5 arb
a n i ew ian
b

h e al k /v @
t r n/ t
e .i m il.c o m
is rb nd .co a
a oi ir m
B /j p g
n p:/ /ph er@
Ia tt :/ rb
h ttp a
h n.b
ia

“0MQ is
unbelievably cool
– if you haven’t
got a project that
needs it, make
one up”
jon gifford - loggly

queue esb

pipeline async

pub/sub gateway

request/response

$ctx = new ZMQContext(); rep.php
$server =
new ZMQSocket($ctx, ZMQ::SOCKET_REP);
$server->bind("tcp://*:5454");

while(true) {
$message = $server->recv();
$server->send($message . " World");
}

request/response

$ctx = new ZMQContext(); req.php
$req =
new ZMQSocket($ctx, ZMQ::SOCKET_REQ);
$req->connect("tcp://localhost:5454");

$req->send("Hello");
echo $req->recv();

import zmq rep.py
context = zmq.Context()
server = context.socket(zmq.REP)
server.connect("tcp://localhost:5455")

while True:
message = server.recv()
print "Sending", message, "Worldn"
server.send(message + " World")

Image: http://ﬂickr.com/photos/sebastian_bergmann/3318754086

wget http://download.zeromq.org/
zeromq-2.1.1.tar.gz
tar xvzf zeromq-2.1.1.tar.gz
cd zeromq-2.1.1/
./configure
make
sudo make install

pear channel-discover pear.zero.mq
pecl install zero.mq/zmq-beta
echo "extension=zmq.so" >
/etc/php.d/zmq.ini

http://github.com/zeromq/zeromq2

atomic string multipart

messaging

Post Office Image: http://www.flickr.com/photos/10804218@N00/4315282973
Post Box Image: http://www.flickr.com/photos/kenjonbro/3027166169

queue
$ctx = new ZMQContext();
$front = $ctx->getSocket(ZMQ::SOCKET_XREP);
$back = $ctx->getSocket(ZMQ::SOCKET_XREQ);
$front->bind('tcp://*:5454');
$back->bind ('tcp://*:5455');

$poll = new ZMQPoll();
$poll->add($front, ZMQ::POLL_IN);
$poll->add($back, ZMQ::POLL_IN);
$read = $write = array();
$snd = ZMQ::MODE_SNDMORE;
$rcv = ZMQ::SOCKOPT_RCVMORE;
queu e.php

while(true) {
$events = $poll->poll($read, $write);
foreach($read as $socket) {
if($socket === $front) {
do {
$msg = $front->recv();
$more = $front->getSockOpt($rcv);
$back->send($msg, $more ? $snd:0);
} while($more);
} else if($socket === $back) {
do {
$msg = $back->recv();
$more = $back->getSockOpt($rcv);
$front->send($msg, $more ? $snd:0);
} while($more);
}}}

0MQ1 0MQ2

Socket STDIN

POLL

events

$sock = $ctx->getSocket(ZMQ::SOCKET_PULL);
$sock->bind("tcp://*:5555");
$fh = fopen("php://stdin", 'r');

$poll->add($sock, ZMQ::POLL_IN);
$poll->add($fh, ZMQ::POLL_IN);

while(true) {
$events = $poll->poll($read, $write);
if($read[0] === $sock) {
echo "ZMQ: ", $read[0]->recv();
} else {
echo "STDIN: ", fgets($read[0]);
}}
poll.php

stable / unstable

Image: http://www.ﬂickr.com/photos/pelican/235461339/

pipeline
define("NUM_WORKERS", 10);
for($i = 0; $i < NUM_WORKERS; $i++) {
if(pcntl_fork() == 0) {
`php work.php`; exit;
}
} con troller.php
$work = $ctx->getSocket(ZMQ::SOCKET_PUSH);
$ctrl = $ctx->getSocket(ZMQ::SOCKET_PUSH);
$work->setSockOpt(ZMQ::SOCKOPT_HWM, 10);
$ctrl->setSockOpt(ZMQ::SOCKOPT_HWM, 1);

$work->bind("ipc:///tmp/work");
$ctrl->bind("ipc:///tmp/control");
sleep(1);

$fh = fopen('data.txt', 'r');

while($data = fgets($fh)) {
$work->send($data);
}

for($i = 0; $i < NUM_WORKERS+1; $i++) {
$ctrl->send("END");
}

work.php
$work = $ctx->getSocket(ZMQ::SOCKET_PULL);
$work->setSockOpt(ZMQ::SOCKOPT_HWM, 1);
$work->connect("ipc:///tmp/work");
$sink = $ctx->getSocket(ZMQ::SOCKET_PUSH);
$sink->connect("ipc:///tmp/results");
$ctrl = $ctx->getSocket(ZMQ::SOCKET_PULL);
$ctrl->connect("ipc:///tmp/control");

$poll->add($work, ZMQ::POLL_IN);

while(true) {
$ev = $poll->poll($read, $write, 5000);
if($ev) {
$message = $work->recv();
$sink->send(strlen($message));
} else {
try {
if($ctrl->recv(ZMQ::MODE_NOBLOCK)) {
exit();
}
} catch(ZMQException $e) {
// noop
}
}
}

sink.php

$res = $ctx->getSocket(ZMQ::SOCKET_PULL);
$res->bind("ipc:///tmp/results");

$ctrl = $ctx->getSocket(ZMQ::SOCKET_PULL);
$ctrl->connect("ipc:///tmp/control");

$poll->add($res, ZMQ::POLL_IN);

$total = 0;

while(true) {
$ev = $poll->poll($read, $write, 10000);
if($ev) {
$total += $res->recv();
} else {
try {
if($ctrl->recv(ZMQ::MODE_NOBLOCK)) {
echo $total, PHP_EOL;
exit();
}
} catch (ZMQException $e) {
//noop
}
}
}

$ php controller.php $ php sink.php
Starting Worker 0 39694
Starting Worker 1
Starting Worker 2
Starting Worker 3
Starting Worker 4
Starting Worker 5
Starting Worker 6
Starting Worker 7
Starting Worker 8
Starting Worker 9

im age up loaded
for proc essing

pub/sub

Image: http://www.ﬂickr.com/photos/nikonvscanon/4519133003/

pub send
“h

“hi”
i”
i”
“h

sub sub sub

i”
“h
ted ann ned

pub/sub

$pub = $ctx->getSocket(ZMQ::SOCKET_PUB);
$pub->bind('tcp://*:5566');
$pull = $ctx->getSocket(ZMQ::SOCKET_PULL);
$pull->bind('tcp://*:5567');

while(true) {
$message = $pull->recv();
$pub->send($message);
} server.php

$name = htmlspecialchars($_POST['name']);
$msg =htmlspecialchars($_POST['message']);

$send = $ctx->getSocket(ZMQ::SOCKET_PUSH);
$send->connect('tcp://localhost:5567');

if($msg == 'm:joined') {
$send->send(
"<em>" . $name . " has joined</em>");
} else {
$send->send($name . ': ' . $msg);
}

send.php

$sub = $ctx->getSocket(ZMQ::SOCKET_SUB);
$sub->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE,'');
$sub->connect('tcp://localhost:5566');
$poll->add($sub, ZMQ::POLL_IN);
$read = $wri = array();
while(true) {
$ev = $poll->poll($read, $wri, 5000000);
if($ev > 0) {
echo "<script type='text/javascript'>
parent.updateChat('";
echo $sub->recv() ."');</script>";
}
ob_flush();
flush();
} ch at.php

sub sub
user data

pub pub pub

web web web

$socket = $ctx->getSocket(ZMQ::SOCKET_PUB);
$socket->connect("ipc:///tmp/usercache");
$socket->connect("ipc:///tmp/datacache");
$type = array('users', 'data');

while(true) {
$socket->send($type[array_rand($type)],
ZMQ::MODE_SNDMORE);
$socket->send(rand(0, 12));
sleep(rand(0,3));
} cach e.php

$socket = $ctx->getSocket(ZMQ::SOCKET_SUB);
$socket->setSockOpt(
ZMQ::SOCKOPT_SUBSCRIBE, "users");
$socket->bind("ipc:///tmp/usercache");

while(true) {
$cache = $socket->recv();
$request = $socket->recv();
echo "Clearing $cache $requestn";
}

userlistener.php

types of transport

inproc ipc

tcp pgm

distro sub web
client
sub web
sub

event
sub sub web
pub

distro sub web
client
sub db

$out = $ctx->getSocket(ZMQ::SOCKET_PUB);
$out->setSockOpt(ZMQ::SOCKOPT_RATE, 10000);
$out->connect("epgm://;239.192.0.1:7601");

$in = $ctx->getSocket(ZMQ::SOCKET_PULL);
$in->bind("tcp://*:6767");

$device = new ZMQDevice(
ZMQ::DEVICE_FORWARDER, $in, $out);

eventhub.php

$in = $ctx->getSocket(ZMQ::SOCKET_SUB);
$in->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, '');
$in->setSockOpt(ZMQ::SOCKOPT_RATE, 10000);
$in->connect("epgm://;239.192.0.1:7601");
$out->bind("ipc:///tmp/events");

$device = new ZMQDevice(
ZMQ::DEVICE_FORWARDER, $in, $out);

distro.php

for($i = 0; $i<100; $i++) {
$in->setSockOpt(
ZMQ::SOCKOPT_SUBSCRIBE,
rand(100000, 999999));
}
$in->connect("ipc:///tmp/events");
$i = 0;
while($i++ < 1000) {
$who = $in->recv();
$msg = $in->recv();
printf("%s %s %s", $who, $msg, PHP_EOL);
}
cli ent.php

push handler
mongrel
pub

client client

mongrel 2
http://mongrel2.org/

$in = $ctx->getSocket(ZMQ::SOCKET_PULL);
$in->connect('tcp://localhost:9997');
$out->connect('tcp://localhost:9996');
$http = "HTTP/1.1 200 OKrnContent-Length:
%srnrn%s";

while(true) {
$msg = $in->recv();
list($uuid, $id, $path, $rest) =
explode(" ", $msg, 4);
$res = $uuid." ".strlen($id).':'.$id.", ";
$res .= sprintf($http, 6, "Hello!");
$out->send($res);
}
handler.php

simple_handler = Handler(
send_spec='tcp://*:9997',
send_ident='ab206881-6f49-4276-9db1-1676bfae18b0',
recv_spec='tcp://*:9996', recv_ident=''
)
main = Server(
uuid="9e71cabf-6afb-4ee1-b550-7972245f7e0a",
access_log="/logs/access.log",
error_log="/logs/error.log",
chroot="./",
default_host="general.local",
name="example",
pid_file="/run/mongre2.pid",
port=6767,
hosts = [
Host(name="general.local",
routes={'/test':simple_handler})
]
)
settings = {"zeromq.threads": 1}
servers = [main]

namespace m2php;
$id ="82209006-86FF-4982-B5EA-D1E29E55D481";
$con = new m2phpConnection($id,
"tcp://127.0.0.1:9997",
"tcp://127.0.0.1:9996");
$in->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE,'');
$sub->connect('tcp://localhost:5566');
$poll->add($sub, ZMQ::POLL_IN);
$poll->add($con->reqs, ZMQ::POLL_IN);
$read = $write = $ids = array(); $snd = '';
$h = "HTTP/1.1 200 OKrnContent-Type: text/
htmlrnTransfer-Encoding: chunkedrnrn";

https://github.com/winks/m2php

while (true) {
$ev = $poll->poll($read, $write);
foreach($read as $r) {
if($r === $in) {
$m = "<script type='text/javascript'>
parent.updateChat('".$in->recv()."');
</script>rn";
$con->send($snd, implode(' ',$ids),
sprintf("%xrn%s",strlen($m),$m));
} else {
$req = $con->recv();
$snd = $req->sender;
if($req->is_disconnect()) {
unset($ids[$req->conn_id]);
} else {
$ids[$req->conn_id] = $req->conn_id;
$con->send($snd, $req->conn_id,$h);
} } } }

Helpful Links
http://zero.mq
http://zguide.zero.mq
Ian Barber nbarber/ZeroMQ-Talk
http://github.com/ia
http://phpir.com
ian.barber@gmail.com @ianbarber
http://joind.in/talk/view/2523

thanks!

ZeroMQ Is The Answer

Recommended

Recommended

More Related Content

What's hot

What's hot (20)

Similar to ZeroMQ Is The Answer

Similar to ZeroMQ Is The Answer (20)

More from Ian Barber

More from Ian Barber (10)

Recently uploaded

Recently uploaded (20)

ZeroMQ Is The Answer