NSQ异步消费模型

「爱情、让人受尽委屈。」 2021-11-04 10:10 324阅读 0赞

网上搜了很久，除了官网的类库，没找到合适的PHP作为消费者的容易理解的案例，那干脆自己写一个好了。

![大致结构图][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ndW9ydWliaWFvLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

搭建环境

# 拉取官方最新镜像
    docker pull nsqio/nsq:latest
    # 启动lookupd 大管家
    docker run -d --name lookupd -p 4160:4160 -p 4161:4161 nsqio/nsq /nsqlookupd
    
    # 启动nsqd 真正干活的，里面的broadcast-address是容器本身的ip
    docker run -d --name=nsqd -p 4150:4150 -p 4151:4151 nsqio/nsq /nsqd --broadcast-address=172.17.0.2 --lookupd-tcp-address=172.17.0.2:4160
    
    # 启动web端admin，可选
    docker run -d --name=nsqadmin -p 4171:4171 nsqio/nsq /nsqadmin --lookupd-http-address=172.17.0.2:4161

生产者模型

package main
    // nsq-productor.go
    import (
       "github.com/bitly/go-nsq"
       "log"
          "flag"
       "encoding/json"
       "bytes"
    )
    
    type Entity struct { 
       Classname string
       MethodName string
       Parameters []string
    }
    
    func main() { 
       var nsqAddr = flag.String("nsqd", "localhost:4150", "nsqd tcp address")
       flag.Parse()
       config := nsq.NewConfig()
    
       producer, err := nsq.NewProducer(*nsqAddr, config)
       if err != nil { 
          log.Fatal(err)
          return
       }
    
       // 生产者生产消息
       for i := 0; i < 2; i++ { 
          params := make([]string, 1)
          params[0] = "biao"
    
          entry := &Entity{ 
             Classname: "DemoService",
             MethodName: "say",
             Parameters: params,
          }
          writer := bytes.NewBuffer(nil)
          encoder := json.NewEncoder(writer)
          encoder.Encode(entry)
          producer.Publish("topic", writer.Bytes())
       }
    }

此异步模拟的目标是PHP端作为消费者，而nsq无法直接把消息转给PHP，所以需要用一个转发者角色来实现这个proxy的功能。

package main
    // nsq-dispatcher.go
    import (
       "github.com/bitly/go-nsq"
       "log"
       "os"
       "fmt"
       "net/http"
       "net/url"
       "io/ioutil"
    )
    
    func encode(request string) url.Values { 
       return url.Values{ "data": { request}}
    }
    
    func msgDispatcher(message string) (string, error){ 
       url := "http://mydomain/atest/201908/fpm.php"
       resp, err := http.PostForm(url, encode(message))
       if err != nil { 
          log.Fatal(err)
          return "", err
       }
       defer resp.Body.Close()
       content, err := ioutil.ReadAll(resp.Body)
       if err != nil { 
          log.Fatal(err)
          return "", err
       }
       return string(content), nil
    }
    
    
    // 从 nsq中拿到数据，然后转发给PHP
    func main() { 
       // 如果有多个处理需求，直接使用多播发给不同的channel即可，交由底层的consumer对channel进行处理
       consumer, err := nsq.NewConsumer("topic", "channel", nsq.NewConfig())
       if err != nil { 
          log.Fatal(err)
          os.Exit(0)
       }
       consumer.AddHandler(nsq.HandlerFunc(func(msg *nsq.Message) error { 
          //fmt.Println("message: ", string((msg.Body)))
          response, err := msgDispatcher(string(msg.Body))
          fmt.Println(response, err)
          return nil
       }))
       err = consumer.ConnectToNSQD("localhost:4150")
       if err != nil { 
          log.Fatal(err)
          os.Exit(0)
       }
       <-make(chan bool)
    }

真正的消费者是PHP端，对上端golang传过来的nsq数据，进行承载，这里使用fpm服务来承接。

<?php
    // fpm.php
    require "./DemoService.php";
    $rawdata = file_get_contents('php://input');
    $rows = explode("=", $rawdata);
    foreach($rows as $key=>$data) { 
        if($key == "data") { 
            $data = json_decode(urldecode($rows[1]), true);
            var_dump($data);
            $classname = $data["Classname"];
            $object = new $classname();
            $ret = call_user_func_array(array($object, $data["MethodName"]), $data["Parameters"]);
            file_put_contents("./nsq-worker.log", json_encode($ret));
        }
    }

里面用到了DemoService.php这个自己写的文件，如下：

<?php
    // DemoService.php
    class DemoService { 
        public function say($message) { 
            return "hello { $message}, this is from nsqworker.";
        }
    }

这样，让proxy一直跑着，用到了阻塞式的channel

<-make(chan bool)

然后，nsq-productor.go一旦Publish了任务，后端的PHP-FPM（可选fastcgi、php-fpm两种模式）就可以接受到对应的内容，并进行对应内容的解析和消费，具体结果可以在日志文件nsq-worker.log 中看到具体的内容。

# nsq-worker.log 
    "hello biao, this is from nsqworker."

至此，简易的demo算是完成了，proxy功能得好好做，要想真正高效率地支撑起服务，不是一件容易的事。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ndW9ydWliaWFvLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20211104/2f67ac8727ec404681d70505050eb3c3.png