golang zmqによるメッセージング

この記事は株式会社アイスタイルアドベントカレンダーの23日目の記事です。

今回は去年PHPアドベントカレンダーでも取り上げた
PHPでメッセージ送受信(zeromq)
を使って簡単にgolangで利用します。

zmq利用準備

golangでzmqを利用する場合は、zeromq/goczmq が利用できます。

Macを利用の場合は、Homebrewなどを利用するといいでしょう。

$ brew install zeromq
$ brew install czmq

ここではメッセージ送信元をPHPとしますので、
PHPの場合は、libzmq、php-zmqエクステンションが必要となります。

libzmq

$ git clone git://github.com/zeromq/libzmq.git
$ cd libzmq/
$ ./autogen.sh
$ ./configure && make
$ make install

php-zmq

$ git clone git://github.com/mkoppanen/php-zmq.git
$ cd php-zmq/
$ phpize
$ ./configure && make
$ make install

remiリポジトリなどを利用しても構いません。
DEALER/ROUTERを使って動かしていきます。
Using DEALER and ROUTER Sockets

PHP メッセージ送信

php-zmqエクステンションを利用してメッセージを送信します。
ここでは5555に接続して、ソケットのタイプを\ZMQ::SOCKET_DEALERとします。

<?php

$socket = new \ZMQSocket(new \ZMQContext(), \ZMQ::SOCKET_DEALER);
$socket->connect("tcp://localhost:5555");
$receive = $socket->send("this is PHP")->recv();

echo "from golang: $receive\n";

クライアントのPHPはメッセージ送信後、golangからメッセージが返却されるまで待ち続けます。

golang メッセージ受信

パッケージで紹介されているサンプルコードを利用して、
クライアントからのメッセージを受信後に返却するようにします。

package main

import (
    "log"
    "github.com/zeromq/goczmq"
)

func main() {

    router, err := goczmq.NewRouter("tcp://*:5555")
    if err != nil {
        log.Fatal(err)
    }
    defer router.Destroy()

    request, err := router.RecvMessage()
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("router received '%s' from '%v'", request[1], request[0])

    err = router.SendFrame(request[0], goczmq.FlagMore)
    if err != nil {
        log.Fatal(err)
    }

    err = router.SendFrame([]byte("this is golang"), goczmq.FlagNone)
    if err != nil {
        log.Fatal(err)
    }
}

これを実際に動かす場合は次のようになります。

# クライアントからメッセージを送信
$ php client.php
# 受信待機
$ ./main

これを用いて、例えばBleveのような簡単な検索パッケージを利用して、
インデックス作成を指示することもできます。

Bleve例

Bleve
クライアント側は受信を行わずに、メッセージングでインデックス追加のみを指示します。

import (
    "log"
    "github.com/zeromq/goczmq"
    "github.com/blevesearch/bleve"
)


type Document struct {
    Content  string
}

func main() {

    router, err := goczmq.NewRouter("tcp://*:5555")
    if err != nil {
        log.Fatal(err)
    }
    defer router.Destroy()

    request, err := router.RecvMessage()
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("router received '%s' from '%v'", request[1], request[0])

    mapping := bleve.NewIndexMapping()
    index, err := bleve.New("search.bleve", mapping)
    if err != nil {
        panic(err)
        return
    }
    // index登録
    index.Index("1", Document{string(request[1])})
}

zmqなどを利用することとでメッセージングによる簡単なアプリケーション開発が可能になりますので、導入してみましょう！