go-micro中的發佈訂閱Broker分析
最近手上有點時間，打算繼續瞭解下go-micro的發佈訂閱（消息），看了micro的examples後，有個疑問，go-micro在提供發佈訂閱的插件Broker（以及幾種實現）的同時，go-micro自己還實現了Publish（Client）以及Subscribe（Server）功能,因而翻了下源碼，作個記錄。node
Broker
Broker是go-micro定義的一個異步消息的接口，同時使用插件的形式，可隨意在不一樣的實現（http，nats,rabbitmq）之間無縫切換。git
// Broker is an interface used for asynchronous messaging.
type Broker interface {
Init(...Option) error
Options() Options
Address() string
Connect() error
Disconnect() error
Publish(topic string, m *Message, opts ...PublishOption) error
Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
String() string
}
從上面的接口能夠看出，使用Broker來完成發佈訂閱只須要如下幾步：github
- 初始化一個Broker(
Init) - 鏈接Broker(
Connect) - 使用準備好的Broker發佈/訂閱(
Publish/Subscribe) - 關閉Broker(
Disconnect)
go-micro中默認的broker實現
go-micro默認有基於http的Broker實現，能夠直接使用。micro有給出具體的example，具體看下source code中的實現。golang
下面是go-micro中broer.go中對DefaultBroker的相關code:json
var (
DefaultBroker Broker = NewBroker()
)
func Init(opts ...Option) error {
return DefaultBroker.Init(opts...)
}
func Connect() error {
return DefaultBroker.Connect()
}
func Disconnect() error {
return DefaultBroker.Disconnect()
}
func Publish(topic string, msg *Message, opts ...PublishOption) error {
return DefaultBroker.Publish(topic, msg, opts...)
}
func Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
return DefaultBroker.Subscribe(topic, handler, opts...)
}
func String() string {
return DefaultBroker.String()
}
能夠看到都是基於NewBroker()返回的broker實例來作的公用方法封裝，咱們進一步看看。緩存
// NewBroker returns a new http broker
func NewBroker(opts ...Option) Broker {
return newHttpBroker(opts...)
}
這裏是直接返回了一個http實現的broker(和上面提到的默認是基於http實現的匹配)，繼續跟newHttpBroker。併發
這裏這列出部分code，詳細的可直接參考go-micro下的http.goapp
h := &httpBroker{
id: uuid.New().String(),
address: addr,
opts: options,
r: options.Registry,
c: &http.Client{Transport: newTransport(options.TLSConfig)},
subscribers: make(map[string][]*httpSubscriber),
exit: make(chan chan error),
mux: http.NewServeMux(),
inbox: make(map[string][][]byte),
}
這裏的核心是new了一個httpBroker，作爲Broker接口的實現，在具體的實現就不在這裏說了，下來咱們看看上面提到接口的實現。異步
Init
func (h *httpBroker) Init(opts ...Option) error {
h.RLock()
if h.running {
h.RUnlock()
return errors.New("cannot init while connected")
}
h.RUnlock()
h.Lock()
defer h.Unlock()
for _, o := range opts {
o(&h.opts)
}
if len(h.opts.Addrs) > 0 && len(h.opts.Addrs[0]) > 0 {
h.address = h.opts.Addrs[0]
}
if len(h.id) == 0 {
h.id = "go.micro.http.broker-" + uuid.New().String()
}
// get registry
reg := h.opts.Registry
if reg == nil {
reg = registry.DefaultRegistry
}
// get cache
if rc, ok := h.r.(cache.Cache); ok {
rc.Stop()
}
// set registry
h.r = cache.New(reg)
// reconfigure tls config
if c := h.opts.TLSConfig; c != nil {
h.c = &http.Client{
Transport: newTransport(c),
}
}
return nil
}
從上面的code中能夠看到，Init的做用就是初始化各類配置，若是Option參數有提供，就是用參數提供的，若是沒有就在這裏設置一個，這裏有2個點咱們須要額外關注下：async
-
Registry
Registry是註冊中心，若是option中沒有提供registry，就會使用go-micro默認實現的（msdn）
-
TLSConfig
TLSConfig是針對https的配置，默認是http
Connect
func (h *httpBroker) Connect() error {
h.RLock()
if h.running {
h.RUnlock()
return nil
}
h.RUnlock()
h.Lock()
defer h.Unlock()
var l net.Listener
var err error
if h.opts.Secure || h.opts.TLSConfig != nil {
config := h.opts.TLSConfig
fn := func(addr string) (net.Listener, error) {
if config == nil {
hosts := []string{addr}
// check if its a valid host:port
if host, _, err := net.SplitHostPort(addr); err == nil {
if len(host) == 0 {
hosts = maddr.IPs()
} else {
hosts = []string{host}
}
}
// generate a certificate
cert, err := mls.Certificate(hosts...)
if err != nil {
return nil, err
}
config = &tls.Config{Certificates: []tls.Certificate{cert}}
}
return tls.Listen("tcp", addr, config)
}
l, err = mnet.Listen(h.address, fn)
} else {
fn := func(addr string) (net.Listener, error) {
return net.Listen("tcp", addr)
}
l, err = mnet.Listen(h.address, fn)
}
if err != nil {
return err
}
addr := h.address
h.address = l.Addr().String()
go http.Serve(l, h.mux)
go func() {
h.run(l)
h.Lock()
h.opts.Addrs = []string{addr}
h.address = addr
h.Unlock()
}()
// get registry
reg := h.opts.Registry
if reg == nil {
reg = registry.DefaultRegistry
}
// set cache
h.r = cache.New(reg)
// set running
h.running = true
return nil
}
Connect方法的主要做用是建立一個Htto Server用來接收Publish時發送的消息
Subscribe
func (h *httpBroker) Subscribe(topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
var err error
var host, port string
options := NewSubscribeOptions(opts...)
// parse address for host, port
host, port, err = net.SplitHostPort(h.Address())
if err != nil {
return nil, err
}
addr, err := maddr.Extract(host)
if err != nil {
return nil, err
}
var secure bool
if h.opts.Secure || h.opts.TLSConfig != nil {
secure = true
}
// register service
node := ®istry.Node{
Id: topic + "-" + h.id,
Address: mnet.HostPort(addr, port),
Metadata: map[string]string{
"secure": fmt.Sprintf("%t", secure),
"broker": "http",
"topic": topic,
},
}
// check for queue group or broadcast queue
version := options.Queue
if len(version) == 0 {
version = broadcastVersion
}
service := ®istry.Service{
Name: serviceName,
Version: version,
Nodes: []*registry.Node{node},
}
// generate subscriber
subscriber := &httpSubscriber{
opts: options,
hb: h,
id: node.Id,
topic: topic,
fn: handler,
svc: service,
}
// subscribe now
if err := h.subscribe(subscriber); err != nil {
return nil, err
}
// return the subscriber
return subscriber, nil
}
這部分代碼的核心功能就是建立用於訂閱的server，一個topic建立一個server並收集（註冊）到httpSubscriber的svc列表中（發佈消息時使用topic在subscriber的svc列表中查詢到對應的server給他發送消息）。
Publish
func (h *httpBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
// create the message first
m := &Message{
Header: make(map[string]string),
Body: msg.Body,
}
for k, v := range msg.Header {
m.Header[k] = v
}
m.Header["Micro-Topic"] = topic
// encode the message
b, err := h.opts.Codec.Marshal(m)
if err != nil {
return err
}
// save the message
h.saveMessage(topic, b)
// now attempt to get the service
h.RLock()
s, err := h.r.GetService(serviceName)
if err != nil {
h.RUnlock()
return err
}
h.RUnlock()
pub := func(node *registry.Node, t string, b []byte) error {
scheme := "http"
// check if secure is added in metadata
if node.Metadata["secure"] == "true" {
scheme = "https"
}
vals := url.Values{}
vals.Add("id", node.Id)
uri := fmt.Sprintf("%s://%s%s?%s", scheme, node.Address, DefaultPath, vals.Encode ())
r, err := h.c.Post(uri, "application/json", bytes.NewReader(b))
if err != nil {
return err
}
// discard response body
io.Copy(ioutil.Discard, r.Body)
r.Body.Close()
return nil
}
srv := func(s []*registry.Service, b []byte) {
for _, service := range s {
var nodes []*registry.Node
for _, node := range service.Nodes {
// only use nodes tagged with broker http
if node.Metadata["broker"] != "http" {
continue
}
// look for nodes for the topic
if node.Metadata["topic"] != topic {
continue
}
nodes = append(nodes, node)
}
// only process if we have nodes
if len(nodes) == 0 {
continue
}
switch service.Version {
// broadcast version means broadcast to all nodes
case broadcastVersion:
var success bool
// publish to all nodes
for _, node := range nodes {
// publish async
if err := pub(node, topic, b); err == nil {
success = true
}
}
// save if it failed to publish at least once
if !success {
h.saveMessage(topic, b)
}
default:
// select node to publish to
node := nodes[rand.Int()%len(nodes)]
// publish async to one node
if err := pub(node, topic, b); err != nil {
// if failed save it
h.saveMessage(topic, b)
}
}
}
}
// do the rest async
go func() {
// get a third of the backlog
messages := h.getMessage(topic, 8)
delay := (len(messages) > 1)
// publish all the messages
for _, msg := range messages {
// serialize here
srv(s, msg)
// sending a backlog of messages
if delay {
time.Sleep(time.Millisecond * 100)
}
}
}()
return nil
看過了上面的Subscribe實現，這裏的Publish就比較簡單
- 建立消息體並存儲在inbox
- 根據topic以及broker的標籤(這裏是固定http)來查找訂閱的server（在上面訂閱模塊建立的）
上面有可能會查找出多個node(訂閱server)，因此裏面還有一個版本的機制，若是指定了版本就會給全部的匹配節點發送（默認是隨機發送一個）
- 使用http post的方式（異步）把消息發送出去
Disconnect
func (h *httpBroker) Disconnect() error {
h.RLock()
if !h.running {
h.RUnlock()
return nil
}
h.RUnlock()
h.Lock()
defer h.Unlock()
// stop cache
rc, ok := h.r.(cache.Cache)
if ok {
rc.Stop()
}
// exit and return err
ch := make(chan error)
h.exit <- ch
err := <-ch
// set not running
h.running = false
return err
}
這部分功能很簡單，清空緩存併發送退出的消息，同時中止服務
以上就是go-micro中默認基於http的broker實現。
go-micro中對於broker的包裝
在看完broker的http默認實現後，咱們對於broker有了一個大致瞭解，接下來咱們在看下go-micro對於broker作的包裝部分，應該是爲了簡化使用（確實只須要一步就能夠）。
訂閱RegisterSubscriber：
func main() {
// create a service
service := micro.NewService(
micro.Name("go.micro.srv.pubsub"),
)
// parse command line
service.Init()
// register subscriber
micro.RegisterSubscriber("example.topic.pubsub.1", service.Server(), new(Sub))
// register subscriber with queue, each message is delivered to a unique subscriber
micro.RegisterSubscriber("example.topic.pubsub.2", service.Server(), subEv, server.SubscriberQueue("queue.pubsub"))
if err := service.Run(); err != nil {
log.Fatal(err)
}
}
發佈NewPublisher, Publish：
func main() {
// create a service
service := micro.NewService(
micro.Name("go.micro.cli.pubsub"),
)
// parse command line
service.Init()
// create publisher
pub1 := micro.NewPublisher("example.topic.pubsub.1", service.Client())
pub2 := micro.NewPublisher("example.topic.pubsub.2", service.Client())
// pub to topic 1
go sendEv("example.topic.pubsub.1", pub1)
// pub to topic 2
go sendEv("example.topic.pubsub.2", pub2)
// block forever
select {}
}
以上只是代碼節選，具體使用方法能夠參考example中的pubsub。
Subscriber
訂閱對比直接用Broker只須要一步RegisterSubscriber,咱們看看裏面實現
//go-micro/micro.go
// RegisterSubscriber is syntactic sugar for registering a subscriber
func RegisterSubscriber(topic string, s server.Server, h interface{}, opts ...server.SubscriberOption) error {
return s.Subscribe(s.NewSubscriber(topic, h, opts...))
}
//go-micro/server/rpc_server.go
func (s *rpcServer) NewSubscriber(topic string, sb interface{}, opts ...SubscriberOption) Subscriber {
return s.router.NewSubscriber(topic, sb, opts...)
}
func (s *rpcServer) Subscribe(sb Subscriber) error {
s.Lock()
defer s.Unlock()
if err := s.router.Subscribe(sb); err != nil {
return err
}
s.subscribers[sb] = nil
return nil
}
//go-micro/server/rpc_router.go
// router represents an RPC router.
type router struct {
.......
subscribers map[string][]*subscriber
}
//go-micro/server/subscriber.go
type subscriber struct {
topic string
rcvr reflect.Value
typ reflect.Type
subscriber interface{}
handlers []*handler
endpoints []*registry.Endpoint
opts SubscriberOptions
}
上面的節選code能夠看出，在默認server(rpcServer)中的router中定義了個map類型的變量subscribers用來存儲訂閱的topic和對應處理的subscriber,server在接收到消息後，只須要根據topic去map中找到subscriber，去處理便可。
subscriber中具體的處理，能夠從定義中看出來，裏面存儲對應路由和響應的handler（server自己的功能），有興趣能夠在go-micro/server/subscriber.go看看具體代碼實現。
Publisher
發佈是在go-micro的默認client實現（rpc_client）裏面定義了一個默認的broker(上面有分析過的http實現)
//go-micro/micro.go
// Deprecated: NewPublisher returns a new Publisher
func NewPublisher(topic string, c client.Client) Event {
return NewEvent(topic, c)
}
// NewEvent creates a new event publisher
func NewEvent(topic string, c client.Client) Event {
if c == nil {
c = client.NewClient()
}
return &event{c, topic}
}
//go-micro/event.go
type event struct {
c client.Client
topic string
}
func (e *event) Publish(ctx context.Context, msg interface{}, opts ...client.PublishOption) error {
return e.c.Publish(ctx, e.c.NewMessage(e.topic, msg), opts...)
}
這裏能夠看到其實是使用傳遞進來的client來初始化一個event，並用來發送消息，若是傳遞的是空，默認建立一個client（rpcClient）。
總結
通過以上過程的追蹤，最終總結下來就幾點：
- broker定義了接口，micro提供的插件的形式可無縫替換實現
- go-micro提供了一個默認的broker實現，是基於http
- go-micro的基於默認的server、client以及brkoer包裝了一套更簡單的pub和sub方法
相關文章
- 1. redis發佈訂閱解析
- 2. 發佈訂閱
- 3. SQLServer 中發佈與訂閱
- 4. Redis 發佈訂閱
- 5. Redis 訂閱發佈
- 6. redis發佈訂閱
- 7. Redis發佈訂閱
- 8. 發佈訂閱EventEmitter
- 9. Redis訂閱發佈
- 10. Redis_發佈訂閱
- 更多相關文章...
- • Redis發佈訂閱模式 - Redis教程
- • RSS 發佈您的 Feed - RSS 教程
- • 常用的分佈式事務解決方案
- • JDK13 GA發佈：5大特性解讀
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. redis發佈訂閱解析
- 2. 發佈訂閱
- 3. SQLServer 中發佈與訂閱
- 4. Redis 發佈訂閱
- 5. Redis 訂閱發佈
- 6. redis發佈訂閱
- 7. Redis發佈訂閱
- 8. 發佈訂閱EventEmitter
- 9. Redis訂閱發佈
- 10. Redis_發佈訂閱