专栏首页码匠的流水账聊聊kingbus的startMasterServer

聊聊kingbus的startMasterServer

本文主要研究一下kingbus的startMasterServer

startMasterServer

kingbus/server/server.go

func (s *KingbusServer) startMasterServer(args *config.BinlogServerConfig) error {
    master, err := NewBinlogServer(args, s, s.store, s.applyBroadcast)
    if err != nil {
        log.Log.Errorf("NewBinlogServer error,err:%s,args:%v", err, *args)
        return err
    }
    s.master = master
    s.master.Start()
    log.Log.Infof("startMasterServer success,args:%v", *args)
    return nil
}
  • startMasterServer方法先执行NewBinlogServer创建master,然后执行master的Start方法

NewBinlogServer

kingbus/server/binlog_server.go

//NewBinlogServer create a binlog server
func NewBinlogServer(cfg *config.BinlogServerConfig, ki KingbusInfo, store storage.Storage, broadcast *utils.Broadcast) (*BinlogServer, error) {
    var err error
    s := new(BinlogServer)

    s.started = atomic.NewBool(false)
    s.cfg = cfg
    s.listener, err = net.Listen("tcp", s.cfg.Addr)
    if err != nil {
        log.Log.Errorf("Listen error,err:%s,addr:%s", err, s.cfg.Addr)
        return nil, err
    }
    s.store = store
    s.broadcast = broadcast
    s.kingbusInfo = ki
    s.slaves = make(map[string]*mysql.Slave)
    s.errch = make(chan error, 1)

    return s, nil
}
  • NewBinlogServer方法通过new方法创建BinlogServer,之后设置其listener、store、broadcast等属性

BinlogServer

kingbus/server/binlog_server.go

//BinlogServer is a binlog server,send binlog event to slave.
//The generic process:
//1.authentication
//SHOW GLOBAL VARIABLES LIKE 'BINLOG_CHECKSUM'
//SET @master_binlog_checksum='NONE'
//SET @master_heartbeat_period=%d
//2.COM_REGISTER_SLAVE
//3.semi-sync:
//SHOW VARIABLES LIKE 'rpl_semi_sync_master_enabled';
//SET @rpl_semi_sync_slave = 1
//4.COM_BINLOG_DUMP_GTID
type BinlogServer struct {
    started *atomic.Bool
    cfg     *config.BinlogServerConfig

    listener net.Listener
    errch    chan error

    l      sync.RWMutex
    slaves map[string]*mysql.Slave //key is uuid

    broadcast   *utils.Broadcast
    kingbusInfo KingbusInfo
    store       storage.Storage
}
  • BinlogServer定义了Bool、BinlogServerConfig、Listener、Slave、Broadcast、KingbusInfo、Storage属性

Start

kingbus/server/binlog_server.go

//Start implements binlog server start
func (s *BinlogServer) Start() {
    s.started.Store(true)
    go func() {
        for s.started.Load() {
            select {
            case err := <-s.errch:
                log.Log.Errorf("binlog server Run error,err:%s", err)
                s.Stop()
                return
            default:
                conn, err := s.listener.Accept()
                if err != nil {
                    log.Log.Infof("BinlogServer.Start:Accept error,err:%s", err)
                    continue
                }
                go s.onConn(conn)
            }
        }
    }()
}
  • Start方法先执行s.started.Store(true),然后通过select机制监听s.listener.Accept(),之后执行s.onConn(conn)

onConn

kingbus/server/binlog_server.go

func (s *BinlogServer) onConn(c net.Conn) {
    mysqlConn, err := mysql.NewConn(c, s, s.cfg.User, s.cfg.Password)
    if err != nil {
        log.Log.Errorf("onConn error,err:%s", err)
        return
    }
    mysqlConn.Run()
}
  • onConn方法通过mysql.NewConn创建mysqlConn,然后执行mysqlConn.Run方法

NewConn

kingbus/mysql/conn.go

//NewConn create a Conn
func NewConn(conn net.Conn, s BinlogServer, user string, password string) (*Conn, error) {
    c := new(Conn)

    c.user = user
    c.BaseConn = NewBaseConn(conn)
    c.connectionID = baseConnID.Add(1)
    c.salt, _ = gomysql.RandomBuf(20)
    c.closed = atomic.NewBool(false)
    masterInfo, err := s.GetMasterInfo()
    if err != nil {
        c.BaseConn.Close()
        log.Log.Errorf("NewConn:GetMasterInfo error,err:%s", err)
        return nil, err
    }

    err = c.handshake(masterInfo.Version, password)
    if err != nil {
        c.BaseConn.Close()
        log.Log.Errorf("NewConn:handshake error,err:%s", err)
        return nil, err
    }

    c.ctx, c.cancel = context.WithCancel(context.Background())
    c.userVariables = make(map[string]interface{})
    c.binlogServer = s

    return c, nil
}
  • NewConn方法通过s.GetMasterInfo()获取master信息,然后执行c.handshake(masterInfo.Version, password)

handshake

kingbus/mysql/conn.go

//handshake implements the handshake protocol in mysql
func (c *Conn) handshake(serverVersion, password string) error {
    if err := c.writeInitialHandshake(serverVersion); err != nil {
        return err
    }

    if err := c.readHandshakeResponse(password); err != nil {
        c.writeError(err)
        return err
    }

    if err := c.writeOK(nil); err != nil {
        return err
    }

    c.ResetSequence()

    return nil
}
  • handshake方法实现的是mysql的handshake协议

Run

kingbus/mysql/conn.go

//Run implements handle client request in Conn
func (c *Conn) Run() {
    defer func() {
        r := recover()
        if err, ok := r.(error); ok {
            const size = 4096
            buf := make([]byte, size)
            buf = buf[:runtime.Stack(buf, false)]

            log.Log.Errorf("Conn Run error,err:%s,stack:%s", err, string(buf))
        }
        c.Close()
        log.Log.Debugf("close client connection")
    }()

    for {
        select {
        case <-c.ctx.Done():
            log.Log.Debugf("BinlogServer closed, close connection")
            return
        default:
            data, err := c.ReadPacket()
            if err != nil {
                log.Log.Errorf("ReadPacket error,err:%s", err)
                return
            }

            if err := c.dispatch(c.ctx, data); err != nil {
                log.Log.Errorf("dispatch error,err:%s,data:%v", err.Error(), data)
                //if the error is canceled, means the connection was killed by cmd
                //don't need send error message
                if err != context.Canceled && c.closed.Load() == false {
                    c.writeError(err)
                }
                return
            }

            //if the connection is closed, return from loop
            if c.closed.Load() == true {
                log.Log.Infof("connection status is closed,need return")
                return
            }
            c.Sequence = 0
        }
    }
}

-Run方法接收请求,然后通过c.dispatch(c.ctx, data)进行分发

dispatch

kingbus/mysql/command.go

func (c *Conn) dispatch(ctx context.Context, data []byte) error {
    cmd := data[0]
    data = data[1:]

    switch cmd {
    case gomysql.COM_QUIT:
        c.Close()
        log.Log.Debugf("close client connection")
        return nil
    case gomysql.COM_QUERY:
        return c.handleQuery(utils.BytesToString(data))
    case gomysql.COM_PING:
        return c.writeOK(nil)
    case gomysql.COM_BINLOG_DUMP_GTID:
        return c.handleBinlogDumpGtid(ctx, data)
    case gomysql.COM_REGISTER_SLAVE:
        return c.handleRegisterSlave(data)
    default:
        log.Log.Errorf("master not support this cmd:%v", data)
        return c.writeError(ErrSQLNotSupport)
    }

    return nil
}
  • dispatch根据不同的cmd来执行conn的不同方法

小结

startMasterServer方法先执行NewBinlogServer创建master,然后执行master的Start方法;master的Start方法先执行s.started.Store(true),然后通过select机制监听s.listener.Accept(),之后执行s.onConn(conn);onConn方法通过mysql.NewConn创建mysqlConn,然后执行mysqlConn.Run方法

doc

  • server

本文分享自微信公众号 - 码匠的流水账(geek_luandun),作者:码匠乱炖

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-06-19

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊kingbus的startMasterServer

    -Run方法接收请求,然后通过c.dispatch(c.ctx, data)进行分发

    codecraft
  • 聊聊kingbus的command.go

    kingbus的command.go提供了Close、handleQuery、writeOK、handleBinlogDumpGtid、handleRegist...

    codecraft
  • 聊聊kingbus的command.go

    kingbus的command.go提供了Close、handleQuery、writeOK、handleBinlogDumpGtid、handleRegist...

    codecraft
  • 聊聊kingbus的startMasterServer

    -Run方法接收请求,然后通过c.dispatch(c.ctx, data)进行分发

    codecraft
  • 49. 访问PostgreSQL数据库增删改查 | 厚土Go学习笔记

    PostgreSQL是常见的免费的大型关系型数据库,具有丰富的数据类型,也是软件项目常用的数据库之一。 因其可靠的稳定性,通常我们可以拿它来做Oracle的替代...

    李海彬
  • Go 语言操作 MySQL 之 CURD 操作

    友情提示:此篇文章大约需要阅读 10分钟 52秒,不足之处请多指教,感谢你的阅读。

    Debug客栈
  • 03 . Go开发一个日志平台之Elasticsearch使用及kafka消费消息发送到Elasticsearch

    https://www.cnblogs.com/you-men/p/13391265.html

    youmen
  • golang简单tls协议用法完整示例

    本文实例讲述了golang简单tls协议用法。分享给大家供大家参考,具体如下: 生成私钥: openssl genrsa -out key.pem 2048 ...

    李海彬
  • golang简单tls协议用法完整示例

    本文实例讲述了golang简单tls协议用法。分享给大家供大家参考,具体如下: 生成私钥: openssl genrsa -out key.pem 2048 生...

    李海彬
  • 《Network Programming with Go》阅读重点备忘(一)

    jeremyxu

扫码关注云+社区

领取腾讯云代金券