集群模式
nats的集群模式對客戶端來說並不是透明的。
所以集群對發布者和訂閱者都是有影響的。
發布者和訂閱者都知道連接的是一個集群的服務，而不是一個單點服務,換句話說發布者訂閱者都必須指明集群中各個節點的地址。
當然，發布者和訂閱者可以只針對集群中的某節點發布消息和訂閱消息，不過這並不是集群模式的目的。
目的
提高可用性和可伸縮性。
實現原理
可用性，多個節點，掛掉任意一個，不影響整個集群對外提供服務。
伸縮性，服務端支持隨意增加節點。訂閱者可以感知服務端節點的變動，但是發布者並不能自動感知。
3個node的集群
$ gnatsd -p 4222 -m 4333 -cluster nats://localhost:4248 -routes nats://localhost:5248,nats://localhost:6248 -DV $ gnatsd -p 5222 -m 5333 -cluster nats://localhost:5248 -routes nats://localhost:4248,nats://localhost:6248 -DV $ gnatsd -p 6222 -m 6333 -cluster nats://localhost:6248 -routes nats://localhost:4248,nats://localhost:5248 -DV -p 端口：服務端口，發布者，訂閱者需要使用此端口。
-m 端口: 監控端口。
-cluster 地址:作為集群節點對其他節點提供服務的地址，其他節點需要連接的地址。(其他節點的-routes 可以填寫此地址)
-routes 地址：此節點，路由到其他地址的列表(也就是其他節點的-cluster)
-DV Debug and trace
gnatsd -p 服務提供端口 -m 服務監控端口 -cluster 集群內node地址 -routes 集群內其他node地址列表 -DV
Server
package main
import ( "github.com/nats-io/go-nats" "log" "flag" "fmt" "time" ) const ( //url = "nats://192.168.3.125:4222" //url = nats.DefaultURL url = "nats://localhost:4222,nats://localhost:6222" //url = "nats://localhost:4222,nats://localhost:5222,nats://localhost:6222" ) var ( nc *nats.Conn err error ) func init() { if nc, err = nats.Connect(url, nats.DontRandomize(), nats.MaxReconnects(5), nats.ReconnectWait(2*time.Second), nats.DisconnectHandler(func(nc *nats.Conn) { fmt.Printf("Got disconnected!\n") }), nats.ReconnectHandler(func(_ *nats.Conn) { fmt.Printf("Got reconnected to %v!\n", nc.ConnectedUrl()) }), nats.ClosedHandler(func(nc *nats.Conn) { fmt.Printf("Connection closed. Reason: %q\n", nc.LastError()) }), nats.DiscoveredServersHandler(func(conn *nats.Conn) { fmt.Printf("Got Discover Server %v!\n", nc.ConnectedUrl()) }), nats.ErrorHandler(func(conn *nats.Conn, subscription *nats.Subscription, e error) { fmt.Printf("Got Error Server %v!\n",e) })); checkErr(err) { // } } func main() { var ( servername = flag.String("servername", "y", "name for server") queueGroup = flag.String("group", "", "group name for Subscribe") subj = flag.String("subj", "abc", "subject name") ) flag.Parse() log.Println(*servername, *queueGroup, *subj) startService(*subj, *servername+" worker1", *queueGroup) //startService(*subj, *servername+" worker2", *queueGroup) //startService(*subj, *servername+" worker3", *queueGroup) select {} } //receive message func startService(subj, name, queue string) { go async(nc, subj, name, queue) } func async(nc *nats.Conn, subj, name, queue string) { _, e := nc.QueueSubscribe(subj, queue, func(msg *nats.Msg) { log.Println(name, "Received a message From Async : ", string(msg.Data)) }) checkErr(e) } func checkErr(err error) bool { if err != nil { log.Println("error:", err) return false } return true } Client
package main import ( "github.com/nats-io/go-nats" "log" "strconv" "github.com/pborman/uuid" "flag" "time" "fmt" ) const ( //url = "nats://192.168.3.125:4222" //url = "nats://localhost:4222" //url = "nats://localhost:4222,nats://localhost:6222" url = "nats://localhost:4222,nats://localhost:5222,nats://localhost:6222" //url = "nats://localhost:5222" ) var ( nc *nats.Conn err error ) func init() { if nc, err = nats.Connect(url, nats.DontRandomize(), nats.MaxReconnects(10), nats.ReconnectWait(2*time.Second), nats.DisconnectHandler(func(nc *nats.Conn) { fmt.Printf("Got disconnected!\n") }), nats.ReconnectHandler(func(_ *nats.Conn) { fmt.Printf("Got reconnected to %v!\n", nc.ConnectedUrl()) }), nats.ClosedHandler(func(nc *nats.Conn) { fmt.Printf("Connection closed. Reason: %q\n", nc.LastError()) })); checkErr(err) { // } nc.SetDiscoveredServersHandler(func(conn *nats.Conn) { }) } func main() { var ( subj = flag.String("subj", "abc", "subject name") ) flag.Parse() log.Println(*subj) startClient(*subj) time.Sleep(time.Second) } //send message to server func startClient(subj string) { for i := 0; i < 1; i++ { id := uuid.New() log.Println(id) nc.Publish(subj, []byte(id+" Golang "+strconv.Itoa(i))) //nc.Publish(subj, []byte(id+" Rain "+strconv.Itoa(i))) //nc.Publish(subj, []byte(id+" Fog "+strconv.Itoa(i))) //nc.Publish(subj, []byte(id+" Cloudy "+strconv.Itoa(i))) } } func checkErr(err error) bool { if err != nil { log.Println(err) return false } return true } 注意
- 發布者和訂閱者都需要指明3個節點的ur地址
nats://localhost:4222,nats://localhost:5222,nats://localhost:6222- 如果3個node都不可用，發布者會發送消息失敗。
- 如果3個node至少有一個可用，訂閱者就會收到消息。
- 如果3個node全都不可用，訂閱者會自動斷開連接。
- 增加一個node nats://localhost:7222,訂閱者可以自動連接。
- 增加node后，3個node全都不可用，訂閱者不會斷開連接，可以接受從新node發布的消息。
- 3個node恢復后，訂閱者可以接受3個node的消息。
后續
發布者和訂閱者
- 原始集群中node都不可用
- 主動查詢可用node
- 接受可用node通知
- 向可用node發送消息，訂閱可用node的消息
- 以上內容需要配合服務發現中間件或者自己實現
配置文件啟動
$ gnatsd -c nodea.cfg $ gnatsd -c nodeb.cfg $ gnatsd -c nodec.cfg nodea.cfg
listen: localhost:4222 # host/port to listen for client connections http: localhost:4333 # HTTP monitoring port # Authorization for client connections #authorization { #user: yasenagat # ./util/mkpasswd -p T0pS3cr3t #password: $2a$11$W2zko751KUvVy59mUTWmpOdWjpEm5qhcCZRd05GjI/sSOT.xtiHyG #ytc #token: $2a$11$ZuYXelbdaRQnOcADEx40yOtinCvEi9c3X64K2Kyx7wLJq7ECPUnA2 #timeout: 1 #} # Cluster definition cluster { listen: localhost:4248 # host/port for inbound route connections # Authorization for route connections #authorization { #user: user2 # ./util/mkpasswd -p T0pS3cr3tT00! #password: $2a$11$xH8dkGrty1cBNtZjhPeWJewu/YPbSU.rXJWmS6SFilOBXzmZoMk9m #yctc #token: $2a$11$d/RrRseSiPOd/fxurspFquSirrjseRFRFGHdRbte7D8wj2laCLcVS #timeout: 0.5 #} # Routes are actively solicited and connected to from this server. # Other servers can connect to us if they supply the correct credentials # in their routes definitions from above. routes = [ nats-route://127.0.0.1:5248 nats-route://127.0.0.1:6248 ] } # logging options debug: false trace: true logtime: false log_file: "nodea.log" # pid file pid_file: "nodea.pid" # Some system overides # max_connections max_connections: 100 # max_subscriptions (per connection) max_subscriptions: 1000 # maximum protocol control line max_control_line: 512 # maximum payload max_payload: 65536 # Duration the server can block on a socket write to a client. Exceeding the # deadline will designate a client as a slow consumer. write_deadline: "2s" nodeb.cfg
listen: localhost:5222 # host/port to listen for client connections http: localhost:5333 # HTTP monitoring port # Authorization for client connections authorization { #user: yasenagat # ./util/mkpasswd -p T0pS3cr3t #password: $2a$11$W2zko751KUvVy59mUTWmpOdWjpEm5qhcCZRd05GjI/sSOT.xtiHyG #ytb token: $2a$11$ToARKoxzTSTXxKCljOFe4eDmiPQ/EcaB0M7V8mGE1tfgOv97.iECe timeout: 1 } # Cluster definition cluster { listen: localhost:5248 # host/port for inbound route connections # Authorization for route connections authorization { #user: user1 # ./util/mkpasswd -p T0pS3cr3tT00! #password: pass1 #yctb token: $2a$11$EriHSUV8WO7PWUXTxOCY5uP7MhAswLE2tqQQPuz6kaoF89KhO8CcW timeout: 0.5 } # Routes are actively solicited and connected to from this server. # Other servers can connect to us if they supply the correct credentials # in their routes definitions from above. routes = [ nats-route://127.0.0.1:4248 nats-route://127.0.0.1:6248 ] } # logging options debug: false trace: true logtime: false log_file: "nodeb.log" # pid file pid_file: "nodeb.pid" # Some system overides # max_connections max_connections: 100 # max_subscriptions (per connection) max_subscriptions: 1000 # maximum protocol control line max_control_line: 512 # maximum payload max_payload: 65536 # Duration the server can block on a socket write to a client. Exceeding the # deadline will designate a client as a slow consumer. write_deadline: "2s" nodec.cfg
listen: localhost:6222 # host/port to listen for client connections http: localhost:6333 # HTTP monitoring port # Authorization for client connections #authorization { #user: yasenagat # ./util/mkpasswd -p T0pS3cr3t #password: $2a$11$W2zko751KUvVy59mUTWmpOdWjpEm5qhcCZRd05GjI/sSOT.xtiHyG #ytc #token: $2a$11$HZy0M3lcxxzJRsFhtAoiX.jCuqKLyztcYYZPWRtlR.APhs/4mFYGC #timeout: 1 #} # Cluster definition cluster { listen: localhost:6248 # host/port for inbound route connections # Authorization for route connections #authorization { #user: user2 # ./util/mkpasswd -p T0pS3cr3tT00! #password: $2a$11$xH8dkGrty1cBNtZjhPeWJewu/YPbSU.rXJWmS6SFilOBXzmZoMk9m #yctc #token: $2a$11$srwaIbFHGwIt37t3GrPynOHSpZ2LHTtw1QXWuznXGOaknEwulP4o6 #timeout: 0.5 #} # Routes are actively solicited and connected to from this server. # Other servers can connect to us if they supply the correct credentials # in their routes definitions from above. routes = [ nats-route://127.0.0.1:5248 nats-route://127.0.0.1:4248 ] } # logging options debug: false trace: true logtime: false log_file: "nodec.log" # pid file pid_file: "nodec.pid" # Some system overides # max_connections max_connections: 100 # max_subscriptions (per connection) max_subscriptions: 1000 # maximum protocol control line max_control_line: 512 # maximum payload max_payload: 65536 # Duration the server can block on a socket write to a client. Exceeding the # deadline will designate a client as a slow consumer. write_deadline: "2s"作者：luckyase
鏈接：https://www.jianshu.com/p/0a54ffd7430f
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯系作者獲得授權並注明出處。