Docker部屬Nsq叢集
阿新 • • 發佈:2018-12-30
用一了段時間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語言 簡單寫一個傳送資訊的例子:
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 main import ( "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<Consumer>(); 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 { /// <summary>Handles a message.</summary> public void HandleMessage(IMessage message) { string msg = Encoding.UTF8.GetString(message.Body); Console.WriteLine(msg); } /// <summary> /// Called when a message has exceeded the specified <see cref="Config.MaxAttempts"/>. /// </summary> /// <param name="message">The failed message.</param> public void LogFailedMessage(IMessage message) { // Log failed messages } }