# Messaging технологии

---

## За мен

### Емил Иванов

* Програмист (по душа)
* Python, (Java|Coffee)script, Erlang, Scala, Has<strike>kell</strike> (ужасно труден)...
* [bolddream.com](http://bolddream.com)
* [bitbucket.org/vladev/messaging-openfest](http://bitbucket.org/vladev/messaging-openfest)
* Презентацията е генерирана с landslide: [github.com/adamzap/landslide](https://github.com/adamzap/landslide)

---

## Твърдение

### Threads + shared state

![hard](hard.jpg)

.fx: center-3

---

## Пример 1

![concurrent-java](concurrent-java.jpg)

430 стр.

---


## Пример 2

----

Да вземем една опашка

![tail][tail]


[tail]: cat-tail.jpg

----

Не точно:

![not-tail][not-tail]

По скоро:

![queue][queue]

[queue]: queue.jpg
[not-tail]: cat-tail-not.jpg

---

## Thread safe: прост случай

    !python
    class Queue(object):
        def __init__(self):
            self.lock = Lock()
            self.queue = []

        def pop(self):
            with self.lock:
                return self.queue.pop(0)

        def push(self, v):
            with self.lock:
                self.queue.append(v)
---

## Thread safe: бърз случай

![work][work]

[work]: work.png

---

## Threads

Прекалено ниско ниво на абстракция.

Подобно на GC.

Не е нужно всеки път да откриваме топлата вода.

----

## Алтернатива?

![escape][escape]

[escape]: escape.jpg

----

## Оставете на професионалистите

![pro](fire.jpg)

---

## Комуникация със съобщения

![actor][actor]

[actor]: actor.png

Actor model

---

## Erlang

---

## Actors in Erlang

    !erlang
    -module(counter).
    -export([run/0, counter/1]).

    run() ->
        S = spawn(counter, counter, [0]),
        send_msgs(S, 100000),
        S.

    counter(Sum) ->
        receive
            value -> io:fwrite("Value is ~w~n", [Sum]);
            {inc, Amount} -> counter(Sum+Amount)
        end.

    send_msgs(_, 0) -> true;
    send_msgs(S, Count) ->
        S ! {inc, 1},
        send_msgs(S, Count-1).

.fx: code-70

---

## <strike>Erlang</strike>

---

## Съобщения, но кога?

---

## Email

![email](email.png)

---

## При големи натоварвания

![load](failwhale.gif)

---

## В началото...

![load-simple](load-simple.png)

---

## ... по-късно...

![load-bal](load-bal.png)

---

## Решение със съобщения

![load-msg](load-msg.png)

Map-Reduce?

---

## Когато редът има значение

![in-order](in-order.jpg)

---

## Ред със съобщения

![pipeline](pipeline.png)

Pipeline

---

## RabbitMQ

---

## RabbitMQ

* AMQP протокол
* Монолитен сървър
* Черна кутия
* Написан на Erlang

---

![rabbit](rabbit-post.png)

---

## Как се работи с RabbitMQ?

---

## 1. Свързване


    !python
    from pika.adapters import BlockingConnection
    from pika import BasicProperties

    # Open a connection to RabbitMQ on localhost
    # using all default parameters
    connection = BlockingConnection()

    # Open the channel
    channel = connection.channel()

.fx: code-90

---

## 2. Exchange

    !python
    channel.exchange_declare(
        exchange='fest-exchange',
        type='direct')

.notes: Аналогия с Поща (Postal service)

---

## 3. Queue

    !python
    channel.queue_declare(
        queue="fest-queue",
        durable=True,
        auto_delete=False)

---

## 4. Bind

    !python
    channel.queue_bind(
        queue='fest-queue',
        exchange='fest-exchange')

---

## Exchanges types

---

## Direct

![mq-direct][mq-direct]

[mq-direct]: mq-direct.png

---

## Fanout

![mq-fanout][mq-fanout]

[mq-fanout]: mq-fanout.png

---

## Topic

![mq-topic][mq-topic]

[mq-topic]: mq-topic.png

---

## 5. Публикуване

    !java
    byte[] messageBodyBytes =
        "Hello, world!".getBytes();

    channel.basicPublish(
        "fest-exchange", // Exchange
        null, // Routing key
        null, // Properties
        messageBodyBytes);

.fx: code-80

---

## 6. Абониране

    !python
    def handle_delivery(channel,
        method_frame, header_frame, body):
        # Process the message
        log.debug('Got message with data: %s',
                  body)

    channel.basic_consume(
        handle_delivery, queue='fest-queue')

---

## 7. Потвърждаване

    !python
    def handle_delivery(channel,
        method_frame, header_frame, body):
        # Process the message and acknowledge it

        channel.basic_ack(
            delivery_tag=method_frame.delivery_tag)


    channel.basic_consume(
        handle_delivery, queue='fest-queue')

.fx: code-80

---

## RabbitMQ: Плюсове

* Инсталирате го и готово
* Изцяло се конфигурира през протокола
* Стабилен
* Бърз
* Сходно API независимо от езика

---

## RabbitMQ: Минуси

* Single point of failure - no redundancy (has clustering, though)

Не бива опашките или съобщенията в тях да стават много.

---

## ØMQ (ZeroMQ)

---

## ØMQ

* Библиотека
* Много бърз
* Общо API на всички езици (поне 30)
* API-то представлява супер-сокет

---

## Broadcast

![broadcast](broadcast.svg)

---

## Broadcast сървър

    !python
    import zmq, datetime, time

    ctx = zmq.Context()
    socket = ctx.socket(zmq.PUB)
    socket.bind("tcp://*:6000")

    while True:
        socket.send('Time is %s' %
            datetime.datetime.now())
        time.sleep(1)

---

## Broadcast клиент

    !python
    import zmq

    ctx = zmq.Context()
    socket = ctx.socket(zmq.SUB)
    socket.connect("tcp://127.0.0.1:6000")
    socket.setsockopt(zmq.SUBSCRIBE, '')

    for i in range(10):
        print socket.recv()

    # Time is 2011-10-27 17:30:27.811003
    # Time is 2011-10-27 17:30:28.812346
    # Time is 2011-10-27 17:30:29.813524


.notes: Изпуска съобщенията, ако няма кой да слуша (неможе с RabbitMQ)

---

## Pipeline сървър (Scala)

    !scala
    val ctx = ZMQ.context(2)
    val s = ctx.socket(ZMQ.PUSH)
    s.bind("tcp://*:6001")

    for (i <- 1 to 100) {
      var r1 = (Math.random() * 100).toInt
      var r2 = (Math.random() * 100).toInt

      s.send("%s+%s".format(r1, r2).getBytes(),
             0) // flags
    }

.fx: code-90

---

## Pipeline клиент (PHP)

    !php
    <?php
    $ctx = new ZMQContext();
    $socket = $ctx->getSocket(ZMQ::SOCKET_PULL);
    $socket->connect('tcp://127.0.0.1:6001');

    for ($i = 0; $i < 100; $i++) {
        $data = $socket->recv();
        list($a, $b) = explode('+', $data);
        $c = $a + $b;
        echo "$a + $b = $c\n";
    }

    // 1 + 31 = 32
    // 84 + 50 = 134
    // 88 + 62 = 150

.fx: code-90

---

## Socket types

# `PUSH → PULL`
# `PUB ⇉ SUB`
# `REQ ⇄ REP`

И още няколко

---

## <strike>Излагация</strike> Демо

---

## URL Fetching pipeline

![pipeline](demo.svg)

---

## Webserver

(Node.js + Coffeecript)

    !coffeescript
    zmq = require 'zmq'

    inPort = 5555
    outPort = 5556

    connectedClients = {}

    inSocket = zmq.createSocket 'push'
    inSocket.bind "tcp://127.0.0.1:#{inPort}", (e) ->

    outSocket = zmq.createSocket 'pull'
    outSocket.bind "tcp://127.0.0.1:#{outPort}", (e) ->
      outSocket.on 'message', (m) ->
        title = m.toString('utf-8')
        for id of connectedClients
          connectedClients[id].send(title)



socket.io


.fx: code-70

---

## Fetcher

    !python
    import time, zmq
    from lxml import html

    c = zmq.Context()
    in_socket = c.socket(zmq.PULL)
    in_socket.connect('tcp://127.0.0.1:5555')

    out_socket = c.socket(zmq.PUSH)
    out_socket.connect('tcp://127.0.0.1:5556')

    while True:
        try:
            url = in_socket.recv()
            time.sleep(1) # Simulate latency
            dom = html.parse(url)
            title = dom.getroot().find('head/title').text

            out_socket.send(title)
        except KeyboardInterrupt:
            break

.fx: code-70

---

## URL fetch

Демо

.fx: msg-demo

---

## ØMQ - Плюсове

* Лесен за работа
* Добро и еднакво API
* Мега-хипер-турбо-плюс бърз
* Библиотека (не framework)

---

## ØMQ - Минуси

* Малко по-ниско ниво от RabbitMQ
* Не бива да се използва за външна комуникация

---

## Обобщение

.notes: Програмирането с нишки, семафори и други ЖП елементи е трудно. Една от алтернативите е различните процеси комуникират със съобщения. RabbitMQ и ZeroMQ са едни от примерите. Нека трудните проблеми да бъдат решени веднъж за винаги.

---

## Благодаря

![thank-you](thank-you.png)

---

## Въпроси?

![questions](question-cat.jpg)

`emil.vladev@gmail.com`
