From 148dbbb23c6194e1112b5bf4e2376597b4e8f48f Mon Sep 17 00:00:00 2001 From: Marc Magnin Date: Mon, 5 Feb 2018 15:14:46 +0100 Subject: [PATCH] #13 introduced sync.Pool --- broker/auth.go | 12 ++-- broker/broker.go | 147 ++++++++++++++++++------------------------- broker/client.go | 57 ++++++++--------- broker/config.go | 10 +-- broker/dispatcher.go | 25 ++++++++ broker/info.go | 16 ++--- broker/retain.go | 8 +-- broker/sublist.go | 2 +- broker/worker.go | 28 +++++++++ 9 files changed, 164 insertions(+), 141 deletions(-) create mode 100644 broker/dispatcher.go create mode 100644 broker/worker.go diff --git a/broker/auth.go b/broker/auth.go index d3e7dfc..04466d2 100644 --- a/broker/auth.go +++ b/broker/auth.go @@ -44,10 +44,10 @@ func (b *Broker) handleFsEvent(event fsnotify.Event) error { case b.config.AclConf: if event.Op&fsnotify.Write == fsnotify.Write || event.Op&fsnotify.Create == fsnotify.Create { - brokerLog.Info("text:handling acl config change event:", zap.String("filename", event.Name)) + log.Info("text:handling acl config change event:", zap.String("filename", event.Name)) aclconfig, err := acl.AclConfigLoad(event.Name) if err != nil { - brokerLog.Error("aclconfig change failed, load acl conf error: ", zap.Error(err)) + log.Error("aclconfig change failed, load acl conf error: ", zap.Error(err)) return err } b.AclConfig = aclconfig @@ -60,24 +60,24 @@ func (b *Broker) StartAclWatcher() { go func() { wch, e := fsnotify.NewWatcher() if e != nil { - brokerLog.Error("start monitor acl config file error,", zap.Error(e)) + log.Error("start monitor acl config file error,", zap.Error(e)) return } defer wch.Close() for _, i := range watchList { if err := wch.Add(i); err != nil { - brokerLog.Error("start monitor acl config file error,", zap.Error(err)) + log.Error("start monitor acl config file error,", zap.Error(err)) return } } - brokerLog.Info("watching acl config file change...") + log.Info("watching acl config file change...") for { select { case evt := <-wch.Events: b.handleFsEvent(evt) case err := <-wch.Errors: - brokerLog.Error("error:", zap.Error(err)) + log.Error("error:", zap.Error(err)) } } }() diff --git a/broker/broker.go b/broker/broker.go index de8d230..3045c8c 100644 --- a/broker/broker.go +++ b/broker/broker.go @@ -12,7 +12,6 @@ import ( "time" "github.com/fhmq/hmq/lib/acl" - "github.com/fhmq/hmq/pool" "github.com/eclipse/paho.mqtt.golang/packets" "github.com/shirou/gopsutil/mem" @@ -22,7 +21,8 @@ import ( ) var ( - brokerLog *zap.Logger + log *zap.Logger + messagePoolQueueSize = 4096 ) type Message struct { @@ -31,40 +31,36 @@ type Message struct { } type Broker struct { - id string - cid uint64 - mu sync.Mutex - config *Config - tlsConfig *tls.Config - AclConfig *acl.ACLConfig - wpool *pool.WorkerPool - clients sync.Map - routes sync.Map - remotes sync.Map - nodes map[string]interface{} - clusterPool chan *Message - messagePool chan *Message - sl *Sublist - rl *RetainList - queues map[string]int + id string + cid uint64 + mu sync.Mutex + config *Config + tlsConfig *tls.Config + AclConfig *acl.ACLConfig + dispatcher *Dispatcher + clients sync.Map + routes sync.Map + remotes sync.Map + nodes map[string]interface{} + sl *Sublist + rl *RetainList + queues map[string]int } func NewBroker(config *Config) (*Broker, error) { b := &Broker{ - id: GenUniqueId(), - config: config, - wpool: pool.New(config.Worker), - sl: NewSublist(), - rl: NewRetainList(), - nodes: make(map[string]interface{}), - queues: make(map[string]int), - clusterPool: make(chan *Message), - messagePool: make(chan *Message), + id: GenUniqueId(), + config: config, + dispatcher: NewDispatcher(), + sl: NewSublist(), + rl: NewRetainList(), + nodes: make(map[string]interface{}), + queues: make(map[string]int), } if b.config.TlsPort != "" { tlsconfig, err := NewTLSConfig(b.config.TlsInfo) if err != nil { - brokerLog.Error("new tlsConfig error", zap.Error(err)) + log.Error("new tlsConfig error", zap.Error(err)) return nil, err } b.tlsConfig = tlsconfig @@ -72,7 +68,7 @@ func NewBroker(config *Config) (*Broker, error) { if b.config.Acl { aclconfig, err := acl.AclConfigLoad(b.config.AclConf) if err != nil { - brokerLog.Error("Load acl conf error", zap.Error(err)) + log.Error("Load acl conf error", zap.Error(err)) return nil, err } b.AclConfig = aclconfig @@ -81,26 +77,16 @@ func NewBroker(config *Config) (*Broker, error) { return b, nil } -func (b *Broker) StartDispatcher() { - for { - msg, ok := <-b.messagePool - if !ok { - brokerLog.Error("read message from client channel error") - return - } - b.wpool.Submit(func() { - ProcessMessage(msg) - }) - } +func (b *Broker) DispatchMessage(msg *Message) { + b.dispatcher.Dispatch(msg) } func (b *Broker) Start() { if b == nil { - brokerLog.Error("broker is null") + log.Error("broker is null") return } - go b.StartDispatcher() //listen clinet over tcp if b.config.Port != "" { @@ -124,7 +110,6 @@ func (b *Broker) Start() { //connect on other node in cluster if b.config.Router != "" { - go b.processClusterInfo() b.ConnectToDiscovery() } @@ -149,7 +134,7 @@ func StateMonitor() { func (b *Broker) StartWebsocketListening() { path := b.config.WsPath hp := ":" + b.config.WsPort - brokerLog.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path)) + log.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path)) http.Handle(path, websocket.Handler(b.wsHandler)) var err error if b.config.WsTLS { @@ -158,7 +143,7 @@ func (b *Broker) StartWebsocketListening() { err = http.ListenAndServe(hp, nil) } if err != nil { - brokerLog.Error("ListenAndServe:" + err.Error()) + log.Error("ListenAndServe:" + err.Error()) return } } @@ -177,14 +162,14 @@ func (b *Broker) StartClientListening(Tls bool) { if Tls { hp = b.config.TlsHost + ":" + b.config.TlsPort l, err = tls.Listen("tcp", hp, b.tlsConfig) - brokerLog.Info("Start TLS Listening client on ", zap.String("hp", hp)) + log.Info("Start TLS Listening client on ", zap.String("hp", hp)) } else { hp := b.config.Host + ":" + b.config.Port l, err = net.Listen("tcp", hp) - brokerLog.Info("Start Listening client on ", zap.String("hp", hp)) + log.Info("Start Listening client on ", zap.String("hp", hp)) } if err != nil { - brokerLog.Error("Error listening on ", zap.Error(err)) + log.Error("Error listening on ", zap.Error(err)) return } tmpDelay := 10 * ACCEPT_MIN_SLEEP @@ -192,7 +177,7 @@ func (b *Broker) StartClientListening(Tls bool) { conn, err := l.Accept() if err != nil { if ne, ok := err.(net.Error); ok && ne.Temporary() { - brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms", + log.Error("Temporary Client Accept Error(%v), sleeping %dms", zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond)) time.Sleep(tmpDelay) tmpDelay *= 2 @@ -200,7 +185,7 @@ func (b *Broker) StartClientListening(Tls bool) { tmpDelay = ACCEPT_MAX_SLEEP } } else { - brokerLog.Error("Accept error: %v", zap.Error(err)) + log.Error("Accept error: %v", zap.Error(err)) } continue } @@ -219,7 +204,7 @@ func (b *Broker) Handshake(conn net.Conn) bool { // Force handshake if err := nc.Handshake(); err != nil { - brokerLog.Error("TLS handshake error, ", zap.Error(err)) + log.Error("TLS handshake error, ", zap.Error(err)) return false } nc.SetReadDeadline(time.Time{}) @@ -235,18 +220,18 @@ func TlsTimeout(conn *tls.Conn) { } cs := nc.ConnectionState() if !cs.HandshakeComplete { - brokerLog.Error("TLS handshake timeout") + log.Error("TLS handshake timeout") nc.Close() } } func (b *Broker) StartClusterListening() { var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port - brokerLog.Info("Start Listening cluster on ", zap.String("hp", hp)) + log.Info("Start Listening cluster on ", zap.String("hp", hp)) l, e := net.Listen("tcp", hp) if e != nil { - brokerLog.Error("Error listening on ", zap.Error(e)) + log.Error("Error listening on ", zap.Error(e)) return } @@ -256,7 +241,7 @@ func (b *Broker) StartClusterListening() { conn, err := l.Accept() if err != nil { if ne, ok := err.(net.Error); ok && ne.Temporary() { - brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms", + log.Error("Temporary Client Accept Error(%v), sleeping %dms", zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond)) time.Sleep(tmpDelay) tmpDelay *= 2 @@ -264,7 +249,7 @@ func (b *Broker) StartClusterListening() { tmpDelay = ACCEPT_MAX_SLEEP } } else { - brokerLog.Error("Accept error: %v", zap.Error(err)) + log.Error("Accept error: %v", zap.Error(err)) } continue } @@ -278,16 +263,16 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) { //process connect packet packet, err := packets.ReadPacket(conn) if err != nil { - brokerLog.Error("read connect packet error: ", zap.Error(err)) + log.Error("read connect packet error: ", zap.Error(err)) return } if packet == nil { - brokerLog.Error("received nil packet") + log.Error("received nil packet") return } msg, ok := packet.(*packets.ConnectPacket) if !ok { - brokerLog.Error("received msg that was not Connect") + log.Error("received msg that was not Connect") return } connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket) @@ -295,7 +280,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) { connack.SessionPresent = msg.CleanSession err = connack.Write(conn) if err != nil { - brokerLog.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier)) + log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier)) return } @@ -335,7 +320,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) { case CLIENT: old, exist = b.clients.Load(cid) if exist { - brokerLog.Warn("client exist, close old...", zap.String("clientID", c.info.clientID)) + log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID)) ol, ok := old.(*client) if ok { ol.Close() @@ -345,7 +330,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) { case ROUTER: old, exist = b.routes.Load(cid) if exist { - brokerLog.Warn("router exist, close old...") + log.Warn("router exist, close old...") ol, ok := old.(*client) if ok { ol.Close() @@ -354,7 +339,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) { b.routes.Store(cid, c) } - c.readLoop(b.messagePool) + c.readLoop() } func (b *Broker) ConnectToDiscovery() { @@ -364,8 +349,8 @@ func (b *Broker) ConnectToDiscovery() { for { conn, err = net.Dial("tcp", b.config.Router) if err != nil { - brokerLog.Error("Error trying to connect to route: ", zap.Error(err)) - brokerLog.Debug("Connect to route timeout ,retry...") + log.Error("Error trying to connect to route: ", zap.Error(err)) + log.Debug("Connect to route timeout ,retry...") if 0 == tempDelay { tempDelay = 1 * time.Second @@ -381,7 +366,7 @@ func (b *Broker) ConnectToDiscovery() { } break } - brokerLog.Debug("connect to router success :", zap.String("Router", b.config.Router)) + log.Debug("connect to router success :", zap.String("Router", b.config.Router)) cid := b.id info := info{ @@ -401,22 +386,10 @@ func (b *Broker) ConnectToDiscovery() { c.SendConnect() c.SendInfo() - go c.readLoop(b.clusterPool) + go c.readLoop() go c.StartPing() } -func (b *Broker) processClusterInfo() { - for { - msg, ok := <-b.clusterPool - if !ok { - brokerLog.Error("read message from cluster channel error") - return - } - ProcessMessage(msg) - } - -} - func (b *Broker) connectRouter(id, addr string) { var conn net.Conn var err error @@ -431,13 +404,13 @@ func (b *Broker) connectRouter(id, addr string) { conn, err = net.Dial("tcp", addr) if err != nil { - brokerLog.Error("Error trying to connect to route: ", zap.Error(err)) + log.Error("Error trying to connect to route: ", zap.Error(err)) if retryTimes > 50 { return } - brokerLog.Debug("Connect to route timeout ,retry...") + log.Debug("Connect to route timeout ,retry...") if 0 == timeDelay { timeDelay = 1 * time.Second @@ -477,7 +450,7 @@ func (b *Broker) connectRouter(id, addr string) { c.SendConnect() - go c.readLoop(b.messagePool) + go c.readLoop() go c.StartPing() } @@ -536,7 +509,7 @@ func (b *Broker) SendLocalSubsToRouter(c *client) { if len(subInfo.Topics) > 0 { err := c.WriterPacket(subInfo) if err != nil { - brokerLog.Error("Send localsubs To Router error :", zap.Error(err)) + log.Error("Send localsubs To Router error :", zap.Error(err)) } } } @@ -553,7 +526,7 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke return true }) - // brokerLog.Info("BroadcastInfoMessage success ") + // log.Info("BroadcastInfoMessage success ") } func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) { @@ -565,7 +538,7 @@ func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) { } return true }) - // brokerLog.Info("BroadcastSubscribeMessage remotes: ", s.remotes) + // log.Info("BroadcastSubscribeMessage remotes: ", s.remotes) } func (b *Broker) removeClient(c *client) { @@ -579,7 +552,7 @@ func (b *Broker) removeClient(c *client) { case REMOTE: b.remotes.Delete(clientId) } - // brokerLog.Info("delete client ,", clientId) + // log.Info("delete client ,", clientId) } func (b *Broker) PublishMessage(packet *packets.PublishPacket) { @@ -593,7 +566,7 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) { if sub != nil { err := sub.client.WriterPacket(packet) if err != nil { - brokerLog.Error("process message for psub error, ", zap.Error(err)) + log.Error("process message for psub error, ", zap.Error(err)) } } } diff --git a/broker/client.go b/broker/client.go index 941c2b2..0914469 100644 --- a/broker/client.go +++ b/broker/client.go @@ -86,7 +86,7 @@ func (c *client) init() { c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0] } -func (c *client) keepAlive(ch chan int, mpool chan *Message) { +func (c *client) keepAlive(ch chan int) { defer close(ch) keepalive := time.Duration(c.info.keepalive*3/2) * time.Second timer := time.NewTimer(keepalive) @@ -100,9 +100,8 @@ func (c *client) keepAlive(ch chan int, mpool chan *Message) { timer.Reset(keepalive) continue } - brokerLog.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive)) - msg := &Message{client: c, packet: DisconnectdPacket} - mpool <- msg + log.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive)) + c.broker.DispatchMessage(&Message{client: c, packet: DisconnectdPacket}) timer.Stop() return case _, ok := <-c.closed: @@ -113,33 +112,31 @@ func (c *client) keepAlive(ch chan int, mpool chan *Message) { } } -func (c *client) readLoop(mpool chan *Message) { +func (c *client) readLoop() { nc := c.conn - if nc == nil || mpool == nil { + if nc == nil { return } ch := make(chan int, 1000) - go c.keepAlive(ch, mpool) + go c.keepAlive(ch) for { packet, err := packets.ReadPacket(nc) if err != nil { - brokerLog.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) break } // keepalive channel ch <- 1 - msg := &Message{ + c.broker.DispatchMessage(&Message{ client: c, packet: packet, - } - mpool <- msg + }) } - msg := &Message{client: c, packet: DisconnectdPacket} - mpool <- msg + c.broker.DispatchMessage(&Message{client: c, packet: DisconnectdPacket}) } func ProcessMessage(msg *Message) { @@ -149,7 +146,7 @@ func ProcessMessage(msg *Message) { return } - brokerLog.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID)) + log.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID)) switch ca.(type) { case *packets.ConnackPacket: case *packets.ConnectPacket: @@ -174,7 +171,7 @@ func ProcessMessage(msg *Message) { case *packets.DisconnectPacket: c.Close() default: - brokerLog.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID)) + log.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID)) } } @@ -190,7 +187,7 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) { } if !c.CheckTopicAuth(PUB, topic) { - brokerLog.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) + log.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) return } @@ -201,21 +198,21 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) { puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket) puback.MessageID = packet.MessageID if err := c.WriterPacket(puback); err != nil { - brokerLog.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) return } c.ProcessPublishMessage(packet) case QosExactlyOnce: return default: - brokerLog.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID)) + log.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID)) return } if packet.Retain { if b := c.broker; b != nil { err := b.rl.Insert(topic, packet) if err != nil { - brokerLog.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) } } } @@ -235,7 +232,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) { topic := packet.TopicName r := b.sl.Match(topic) - // brokerLog.Info("psubs num: ", len(r.psubs)) + // log.Info("psubs num: ", len(r.psubs)) if len(r.qsubs) == 0 && len(r.psubs) == 0 { return } @@ -249,7 +246,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) { if sub != nil { err := sub.client.WriterPacket(packet) if err != nil { - brokerLog.Error("process message for psub error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("process message for psub error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) } } } @@ -259,7 +256,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) { t := "$queue/" + topic cnt, exist := b.queues[t] if exist { - // brokerLog.Info("queue index : ", cnt) + // log.Info("queue index : ", cnt) for _, sub := range r.qsubs { if sub.client.typ == ROUTER { if typ != CLIENT { @@ -275,7 +272,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) { if sub != nil { err := sub.client.WriterPacket(packet) if err != nil { - brokerLog.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) } } @@ -329,7 +326,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) { t := topic //check topic auth for client if !c.CheckTopicAuth(SUB, topic) { - brokerLog.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) + log.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) retcodes = append(retcodes, QosFailure) continue } @@ -376,7 +373,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) { } err := b.sl.Insert(sub) if err != nil { - brokerLog.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) retcodes = append(retcodes, QosFailure) } else { retcodes = append(retcodes, qoss[i]) @@ -386,7 +383,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) { err := c.WriterPacket(suback) if err != nil { - brokerLog.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) return } //broadcast subscribe message @@ -398,7 +395,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) { for _, t := range topics { packets := b.rl.Match(t) for _, packet := range packets { - brokerLog.Info("process retain message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID)) + log.Info("process retain message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID)) if packet != nil { c.WriterPacket(packet) } @@ -445,7 +442,7 @@ func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) { err := c.WriterPacket(unsuback) if err != nil { - brokerLog.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) return } // //process ubsubscribe message @@ -474,7 +471,7 @@ func (c *client) ProcessPing() { resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket) err := c.WriterPacket(resp) if err != nil { - brokerLog.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) return } } @@ -505,7 +502,7 @@ func (c *client) Close() { for _, sub := range subs { err := b.sl.Remove(sub) if err != nil { - brokerLog.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) + log.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) } } if c.typ == CLIENT { diff --git a/broker/config.go b/broker/config.go index eea8b84..d61b15d 100644 --- a/broker/config.go +++ b/broker/config.go @@ -105,7 +105,7 @@ func ConfigureConfig(args []string) (*Config, error) { }) logger.InitLogger(config.Debug) - brokerLog = logger.Get().Named("Broker") + log = logger.Get().Named("Broker") if configFile != "" { tmpConfig, e := LoadConfig(configFile) @@ -128,15 +128,15 @@ func LoadConfig(filename string) (*Config, error) { content, err := ioutil.ReadFile(filename) if err != nil { - brokerLog.Error("Read config file error: ", zap.Error(err)) + log.Error("Read config file error: ", zap.Error(err)) return nil, err } - // brokerLog.Info(string(content)) + // log.Info(string(content)) var config Config err = json.Unmarshal(content, &config) if err != nil { - brokerLog.Error("Unmarshal config file error: ", zap.Error(err)) + log.Error("Unmarshal config file error: ", zap.Error(err)) return nil, err } @@ -168,7 +168,7 @@ func (config *Config) check() error { if config.TlsPort != "" { if config.TlsInfo.CertFile == "" || config.TlsInfo.KeyFile == "" { - brokerLog.Error("tls config error, no cert or key file.") + log.Error("tls config error, no cert or key file.") return errors.New("tls config error, no cert or key file.") } if config.TlsHost == "" { diff --git a/broker/dispatcher.go b/broker/dispatcher.go new file mode 100644 index 0000000..f2f68ef --- /dev/null +++ b/broker/dispatcher.go @@ -0,0 +1,25 @@ +package broker + +import ( + "sync" +) + +// Dispatcher will delegate ProcessMessage func to multiple goroutines +type Dispatcher struct { + workerPool *sync.Pool +} + +// NewDispatcher create a *Dispatcher instance +func NewDispatcher() *Dispatcher { + return &Dispatcher{workerPool: &sync.Pool{ + New: func() interface{} { + return NewWorker() + }, + }, + } +} + +// Dispatch a message to the workers +func (d *Dispatcher) Dispatch(message *Message) { + d.workerPool.Get().(Worker).WorkerChannel <- Work{WorkerPool: d.workerPool, Message: message} +} diff --git a/broker/info.go b/broker/info.go index e167767..45c77a7 100644 --- a/broker/info.go +++ b/broker/info.go @@ -21,7 +21,7 @@ func (c *client) SendInfo() { infoMsg := NewInfo(c.broker.id, url, false) err := c.WriterPacket(infoMsg) if err != nil { - brokerLog.Error("send info message error, ", zap.Error(err)) + log.Error("send info message error, ", zap.Error(err)) return } } @@ -34,7 +34,7 @@ func (c *client) StartPing() { case <-timeTicker.C: err := c.WriterPacket(ping) if err != nil { - brokerLog.Error("ping error: ", zap.Error(err)) + log.Error("ping error: ", zap.Error(err)) c.Close() } case _, ok := <-c.closed: @@ -57,10 +57,10 @@ func (c *client) SendConnect() { m.Keepalive = uint16(60) err := c.WriterPacket(m) if err != nil { - brokerLog.Error("send connect message error, ", zap.Error(err)) + log.Error("send connect message error, ", zap.Error(err)) return } - brokerLog.Info("send connect success") + log.Info("send connect success") } func NewInfo(sid, url string, isforword bool) *packets.PublishPacket { @@ -69,7 +69,7 @@ func NewInfo(sid, url string, isforword bool) *packets.PublishPacket { pub.TopicName = BrokerInfoTopic pub.Retain = false info := fmt.Sprintf(`{"brokerID":"%s","brokerUrl":"%s"}`, sid, url) - // brokerLog.Info("new info", string(info)) + // log.Info("new info", string(info)) pub.Payload = []byte(info) return pub } @@ -81,17 +81,17 @@ func (c *client) ProcessInfo(packet *packets.PublishPacket) { return } - brokerLog.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload))) + log.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload))) js, err := simplejson.NewJson(packet.Payload) if err != nil { - brokerLog.Warn("parse info message err", zap.Error(err)) + log.Warn("parse info message err", zap.Error(err)) return } routes, err := js.Get("data").Map() if routes == nil { - brokerLog.Error("receive info message error, ", zap.Error(err)) + log.Error("receive info message error, ", zap.Error(err)) return } diff --git a/broker/retain.go b/broker/retain.go index a13f8a8..8943198 100644 --- a/broker/retain.go +++ b/broker/retain.go @@ -39,7 +39,7 @@ func (r *RetainList) Insert(topic string, buf *packets.PublishPacket) error { if err != nil { return err } - // brokerLog.Info("insert tokens:", tokens) + // log.Info("insert tokens:", tokens) r.Lock() l := r.root @@ -72,7 +72,7 @@ func (r *RetainList) Match(topic string) []*packets.PublishPacket { l := r.root matchRLevel(l, tokens, results) r.Unlock() - // brokerLog.Info("results: ", results) + // log.Info("results: ", results) return results.msg } @@ -82,7 +82,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) { if l == nil { return } - // brokerLog.Info("l info :", l.nodes) + // log.Info("l info :", l.nodes) if t == "#" { for _, n := range l.nodes { n.GetAll(results) @@ -111,7 +111,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) { } func (r *rnode) GetAll(results *RetainResult) { - // brokerLog.Info("node 's message: ", string(r.msg)) + // log.Info("node 's message: ", string(r.msg)) if r.msg != nil { results.msg = append(results.msg, r.msg) } diff --git a/broker/sublist.go b/broker/sublist.go index 19f9cce..3ee0ff7 100644 --- a/broker/sublist.go +++ b/broker/sublist.go @@ -211,7 +211,7 @@ func (s *Sublist) Match(topic string) *SublistResult { tokens, err := PublishTopicCheckAndSpilt(topic) if err != nil { - brokerLog.Error("\tserver/sublist.go: ", zap.Error(err)) + log.Error("\tserver/sublist.go: ", zap.Error(err)) return nil } diff --git a/broker/worker.go b/broker/worker.go new file mode 100644 index 0000000..5e73ebf --- /dev/null +++ b/broker/worker.go @@ -0,0 +1,28 @@ +package broker + +import "sync" + +type Work struct { + WorkerPool *sync.Pool + Message *Message +} + +type Worker struct { + WorkerChannel chan Work +} + +func NewWorker() Worker { + w := Worker{WorkerChannel: make(chan Work)} + return w.Start() +} + +func (w Worker) Start() Worker { + go func() { + for work := range w.WorkerChannel { + ProcessMessage(work.Message) + // put the worker back + work.WorkerPool.Put(w) + } + }() + return w +}