集群模式
nats的集群模式對(duì)客戶端來說并不是透明的。
所以集群對(duì)發(fā)布者和訂閱者都是有影響的。
發(fā)布者和訂閱者都知道連接的是一個(gè)集群的服務(wù)，而不是一個(gè)單點(diǎn)服務(wù),換句話說發(fā)布者訂閱者都必須指明集群中各個(gè)節(jié)點(diǎn)的地址僻他。
當(dāng)然笆包，發(fā)布者和訂閱者可以只針對(duì)集群中的某節(jié)點(diǎn)發(fā)布消息和訂閱消息环揽，不過這并不是集群模式的目的。
目的
提高可用性和可伸縮性庵佣。
實(shí)現(xiàn)原理
可用性歉胶，多個(gè)節(jié)點(diǎn)，掛掉任意一個(gè)巴粪，不影響整個(gè)集群對(duì)外提供服務(wù)通今。
伸縮性，服務(wù)端支持隨意增加節(jié)點(diǎn)肛根。訂閱者可以感知服務(wù)端節(jié)點(diǎn)的變動(dòng)辫塌，但是發(fā)布者并不能自動(dòng)感知。
3個(gè)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 端口：服務(wù)端口派哲，發(fā)布者臼氨，訂閱者需要使用此端口。
-m 端口: 監(jiān)控端口芭届。
-cluster 地址:作為集群節(jié)點(diǎn)對(duì)其他節(jié)點(diǎn)提供服務(wù)的地址储矩，其他節(jié)點(diǎn)需要連接的地址。(其他節(jié)點(diǎn)的-routes 可以填寫此地址)
-routes 地址：此節(jié)點(diǎn)喉脖，路由到其他地址的列表(也就是其他節(jié)點(diǎn)的-cluster)
-DV Debug and trace
gnatsd -p 服務(wù)提供端口 -m 服務(wù)監(jiān)控端口 -cluster 集群內(nèi)node地址 -routes 集群內(nèi)其他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
}
注意
- 發(fā)布者和訂閱者都需要指明3個(gè)節(jié)點(diǎn)的ur地址
nats://localhost:4222,nats://localhost:5222,nats://localhost:6222- 如果3個(gè)node都不可用椰苟，發(fā)布者會(huì)發(fā)送消息失敗。
- 如果3個(gè)node至少有一個(gè)可用树叽，訂閱者就會(huì)收到消息舆蝴。
- 如果3個(gè)node全都不可用，訂閱者會(huì)自動(dòng)斷開連接。
- 增加一個(gè)node nats://localhost:7222,訂閱者可以自動(dòng)連接洁仗。
- 增加node后层皱，3個(gè)node全都不可用，訂閱者不會(huì)斷開連接赠潦，可以接受從新node發(fā)布的消息叫胖。
- 3個(gè)node恢復(fù)后，訂閱者可以接受3個(gè)node的消息她奥。
后續(xù)
發(fā)布者和訂閱者
- 原始集群中node都不可用
- 主動(dòng)查詢可用node
- 接受可用node通知
- 向可用node發(fā)送消息瓮增，訂閱可用node的消息
- 以上內(nèi)容需要配合服務(wù)發(fā)現(xiàn)中間件或者自己實(shí)現(xiàn)
配置文件啟動(dòng)
$ 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"
