逍遥阁

nsq笔记

关键词:

### 用docker搭建服务

docker-compose.yaml内容

```
version: "2"
services:
    nsqlookupd:
        image: nsqio/nsq
        container_name: nsqlookupd
        command: /nsqlookupd
        ports:
            - "4160:4160"
            - "4161:4161"
    nsqd:
        image: nsqio/nsq
        container_name: nsqd
        command: /nsqd --lookupd-tcp-address=nsqlookupd:4160
        depends_on:
            - nsqlookupd
        ports:
            - "4150:4150"
            - "4151:4151"
    nsqadmin:
        image: nsqio/nsq
        container_name: nsqadmin
        command: /nsqadmin --lookupd-http-address=nsqlookupd:4161
        depends_on:
            - nsqlookupd
        ports:
            - "4171:4171"
```

### go发布端

```
package main

import (
    "github.com/nsqio/go-nsq"
)

var producer *nsq.Producer

func main() {
    nsqd := "192.168.21.16:4150"
    producer, err := nsq.NewProducer(nsqd, nsq.NewConfig())
    producer.Publish("test", []byte("nihao"))
    if err != nil {
        panic(err)
    }
}
/*
    curl -d 't1' 'http://192.168.21.16:4151/pub?topic=test'
*/
```

### go消费端

```
package main

import (
    "fmt"
    "github.com/nsqio/go-nsq"
    "sync"
)

type NSQHandler struct {
}

func (this *NSQHandler) HandleMessage(msg *nsq.Message) error {
    fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
    return nil
}

func testNSQ() {
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        defer wg.Done()
        config := nsq.NewConfig()
        config.MaxInFlight = 9
        for i := 0; i < 10; i++ {
            consumer, err := nsq.NewConsumer("test", "struggle", config)
            if nil != err {
                fmt.Println("err", err)
                return
            }
            consumer.AddHandler(&NSQHandler{})
            err = consumer.ConnectToNSQD("192.168.21.16:4150")
            if nil != err {
                fmt.Println("err", err)
                return
            }
        }
        select {}
    }()
    wg.Wait()
}

func main() {
    testNSQ();
}
```

### php服务封装

```
<?php
namespace Service;
use nsqphp\nsqphp;
use nsqphp\Message\Message;
use nsqphp\Lookup\FixedHosts;

class Nsq extends Base
{

public function publish($list, $topic, $message)
    {
        $nsq = new nsqphp();
        $nsq->publishTo($list[0])->publish($topic, new Message($message));
    }

public function subscribe($list, $topic, $channel, $cb)
    {
        $lookup = new FixedHosts($list);
        $nsq = new nsqphp($lookup);
        $nsq->subscribe($topic, $channel, $cb)->run();
    }

}
```

### php测试代码

```
    /*
        /usr/bin/php -d display_error /app/cli.php local Demo.publish
    */
    public function publish()
    {
        $list = ['192.168.21.16:4150'];
        $serviceNsq = $this->get(SERVICE_NSQ);
        $serviceNsq->publish($list, 'test', 'from php');
    }

/*
        /usr/bin/php -d display_error /app/cli.php local Demo.subscribe
    */
    public function subscribe()
    {
        $cb = function($msg)
        {
            print_r($msg);
            echo PHP_EOL;
        };

$list = ['192.168.21.16:4150'];
        $serviceNsq = $this->get(SERVICE_NSQ);
        $serviceNsq->subscribe($list, 'test', 'channel', $cb);
    }
```

admin
2020-04-07 18:11:34
phantoscope

文章分类

文章标签

文章归档

友情链接

nsq笔记