Docker部属Nsq集群-白红宇

　　用一了段时间NSQ还是很稳定的。除了稳定，还有一个特别值的说的就是部署非常简单。总想写点什么推荐给大家使用nsq来做一些东西。但是就是因为他太简单易用，文档也比较简单易懂。一直不知道要写啥！！！！！

　　nsq官网：

　　为了容灾需要对nsqd多机器部属，有了Docker后，快速扩还是很方便的。

　　部署完后我会用go和c#写一些代码方便大家学习。

　　准备工作：

　　》两台服务器：192.168.0.49; 192.168.0.105.

　　》需要在两台机器上安装好Docker

　　》两台机器上镜像的拉取

docker pull nsqio/nsq

　　我们在105上启动lookup, nsqd和客户端都需要连接这个lookup。　　

docker run --name lookupd -p 4160:4160 -p 4161:4161 nsqio/nsq /nsqlookupd

　　在105和49上启动nsqd， lookup的地址要写105

docker run --name nsqd -p 4150:4150 -p 4151:4151     nsqio/nsq /nsqd     --broadcast-address=192.168.0.105     --lookupd-tcp-address=192.168.0.105:4160

docker run --name nsqd -p 4150:4150 -p 4151:4151     nsqio/nsq /nsqd     --broadcast-address=192.168.0.49     --lookupd-tcp-address=192.168.0.105:4160

　　到了这一步就可以写代码发送和接收信息了。但是还有一个管理系统需要启动一下。nsqadmin

docker run --name nsqadmin -p 4171:4171 nsqio/nsq /nsqadmin --lookupd-http-address=192.168.0.105:4161

　　用浏览器看一下管理端：http://192.168.0.105:4171/nodes。找开Nodes标签里面有两个节点。192.168.0.105 和 192.168.0.49。其他的你可以点开看看。

　　我用go语言简单写一个发送信息的例子：

　　go使用的库是 go-nsq 地址：

func main() {    config := nsq.NewConfig()    // 随便给哪个ip发都可以    //w1, _ := nsq.NewProducer("192.168.0.105:4150", config)    w1, _ := nsq.NewProducer("192.168.0.49:4150", config)    err1 := w1.Ping()    if err1 != nil {        log.Fatal("should not be able to ping after Stop()")        return    }    defer w1.Stop()    topicName := "publishtest"    msgCount := 2    for i := 1; i < msgCount; i++ {        err1 := w1.Publish(topicName, []byte("测试测试publis test case"))        if err1 != nil {            log.Fatal("error")        }    }}

　　可以尝试给49和105都发送一次试试。再看一下我们的管理页面：

　　publishtest被ip105和49都发送过。但是还没有channel:

客户端golang代码

package mainimport (    "fmt"    "github.com/nsqio/go-nsq"    "log"    "os"    "os/signal"    "strconv"    "time"    "sync")func main() {    topicName := "publishtest"    msgCount := 2    for i := 0; i < msgCount; i++ {        //time.Sleep(time.Millisecond * 20)        go readMessage(topicName, i)    }    //cleanup := make(chan os.Signal, 1)    cleanup := make(chan os.Signal)    signal.Notify(cleanup, os.Interrupt)    fmt.Println("server is running....")    quit := make(chan bool)    go func() {                select {            case <- cleanup:                fmt.Println("Received an interrupt , stoping service ...")                for _, ele := range consumers {                    ele.StopChan <- 1                    ele.Stop()                }                quit <- true        }    }()    <-quit    fmt.Println("Shutdown server....")}type ConsumerHandle struct {    q       *nsq.Consumer    msgGood int}var consumers []*nsq.Consumer = make([]*nsq.Consumer, 0)var mux *sync.Mutex = &sync.Mutex{}func (h *ConsumerHandle) HandleMessage(message *nsq.Message) error {    msg := string(message.Body) + "  " + strconv.Itoa(h.msgGood)    fmt.Println(msg)    return nil}func readMessage(topicName string, msgCount int) {    defer func() {        if err := recover(); err != nil {            fmt.Println("error: ", err)        }    }()    config := nsq.NewConfig()    config.MaxInFlight = 1000    config.MaxBackoffDuration = 500 * time.Second    //q, _ := nsq.NewConsumer(topicName, "ch" + strconv.Itoa(msgCount), config)    //q, _ := nsq.NewConsumer(topicName, "ch" + strconv.Itoa(msgCount) + "#ephemeral", config)    q, _ := nsq.NewConsumer(topicName, "ch"+strconv.Itoa(msgCount), config)    h := &ConsumerHandle{q: q, msgGood: msgCount}    q.AddHandler(h)    err := q.ConnectToNSQLookupd("192.168.0.105:4161")    //err := q.ConnectToNSQDs([]string{"192.168.0.105:4161"})    //err := q.ConnectToNSQD("192.168.0.49:4150")    //err := q.ConnectToNSQD("192.168.0.105:4415")    if err != nil {        fmt.Println("conect nsqd error")        log.Println(err)    }    mux.Lock()    consumers = append(consumers, q)    mux.Unlock()    <-q.StopChan    fmt.Println("end....")}

　　运行一下，会启动两个终端：

　　用我们的发送代码发送信息，再看我们的客户端

　　c# 使用的库为NsqSharp.Core 地址为：

　　简单客户端代码为：

class Program    {        static void Main()        {            // Create a new Consumer for each topic/channel            var consumerCount = 2;            var listC = new  List
     
      ();            for (var i = 0; i < consumerCount; i++)            {                var consumer = new Consumer("publishtest", $"channel{i}" );                consumer.ChangeMaxInFlight(2500);                consumer.AddHandler(new MessageHandler());                consumer.ConnectToNsqLookupd("192.168.0.105:4161");                listC.Add(consumer);            }            var exitEvent = new ManualResetEvent(false);            Console.CancelKeyPress += (sender, eventArgs) => {                eventArgs.Cancel = true;                listC.ForEach(x => x.Stop());                exitEvent.Set();            };            exitEvent.WaitOne();                               }    }    public class MessageHandler : IHandler    {        /// 
      Handles a message.
        public void HandleMessage(IMessage message)        {            string msg = Encoding.UTF8.GetString(message.Body);            Console.WriteLine(msg);        }        /// 
              /// Called when a message has exceeded the specified 
       .        /// 
        /// 
      The failed message.        public void LogFailedMessage(IMessage message)        {            // Log failed messages        }    }