序 本文主要研究一下rocketmq-client-go的remoteBrokerOffsetStore remoteBrokerOffsetStore rocketmq-client-go-v2.0.0/consumer/offset_store.go {代码...} remoteBrokerOffsetStore定义了group、OffsetTable、client、namesrv、mutex属性 NewRemoteOffsetStore rocketmq-client-go-v2.0.0/consumer/offset_store.go {...

序 本文主要研究一下rocketmq-client-go的localFileOffsetStore OffsetStore rocketmq-client-go-v2.0.0/consumer/offset_store.go {代码...} OffsetStore定义了persist、remove、read、update方法 localFileOffsetStore rocketmq-client-go-v2.0.0/consumer/offset_store.go {代码...} localFileOffsetStore定义了group...

pushConsumer定义了queueFlowControlTimes、queueMaxSpanFlowControlTimes、consumeFunc、submitToConsume、subscribedTopic、interceptor、queueLock、done、closeOnce属性

PullConsumer定义了Start、Shutdown、Pull、UpdateOffset、PersistOffset、CurrentOffset方法

序 本文主要研究一下rocketmq-client-go的ACLInterceptor ACLInterceptor rocketmq-client-go-v2.0.0/internal/remote/interceptor.go {代码...} ACLInterceptor方法会调用calculateSignature计算签名，然后执行next calculateSignature rocketmq-client-go-v2.0.0/internal/remote/interceptor.go {代码...} calculateS...

序 本文主要研究一下rocketmq-client-go的TraceInterceptor TraceInterceptor rocketmq-client-go-v2.0.0/producer/interceptor.go {代码...} WithTrace方法在options.Interceptors后追加TraceInterceptor newTraceInterceptor rocketmq-client-go-v2.0.0/producer/interceptor.go {代码...} newTraceInterceptor方法首...

序 本文主要研究一下rocketmq-client-go的transactionProducer transactionProducer rocketmq-client-go-v2.0.0/producer/producer.go {代码...} transactionProducer定义了producer及listener属性 NewTransactionProducer rocketmq-client-go-v2.0.0/producer/producer.go {代码...} NewTransactionProducer方法实例化t...

defaultProducer定义了group、client、state、options、publishInfo、callbackCh、interceptor

Producer定义了Start、Shutdown、SendSync、SendAsync、SendOneWay方法；NewProducer方法通过producer.NewDefaultProducer创建Producer

序 本文主要研究一下rocketmq-client-go的QueueSelector QueueSelector rocketmq-client-go-v2.0.0/producer/selector.go {代码...} QueueSelector接口，定义了Select方法 manualQueueSelector rocketmq-client-go-v2.0.0/producer/selector.go {代码...} manualQueueSelector的select方法直接返回message.Queue NewRand...

序 本文主要研究一下canal-go的position Position canal-go-v1.0.7/protocol/Position/Position.go {代码...} Position.go定义了一个Position接口 MetaqPosition canal-go-v1.0.7/protocol/Position/metaq_position.go {代码...} MetaqPosition定义了Topic、MsgNewId、Offset属性 TimePosition canal-go-v1.0.7/protocol...

SimpleCanalConnector定义了Address、Port、UserName、PassWord、SoTime、IdleTimeOut、ClientIdentity、Connected、Running、Filter、RollbackOnConnect、LazyParseEntry属性

序 本文主要研究一下nacos-sdk-go的ConfigProxy ConfigProxy nacos-sdk-go-v0.3.2/clients/config_client/config_proxy.go {代码...} ConfigProxy定义了nacosServer、clientConfig属性 NewConfigProxy nacos-sdk-go-v0.3.2/clients/config_client/config_proxy.go {代码...} NewConfigProxy方法创建ConfigProxy，并设置...

序 本文主要研究一下nacos-sdk-go的PushReceiver PushReceiver nacos-sdk-go-v0.3.2/clients/naming_client/push_receiver.go {代码...} PushReceiver定义了port、host、hostReactor属性 NewPushRecevier nacos-sdk-go-v0.3.2/clients/naming_client/push_receiver.go {代码...} NewPushRecevier方法创建PushReceiver，...

NacosServer定义了securityLogin、serverList、httpAgent、timeoutMs、endpoint、lastSrvRefTime、vipSrvRefInterMills属性

序 本文主要研究一下nacos-sdk-go的NamingProxy NamingProxy nacos-sdk-go-v0.3.2/clients/naming_client/naming_proxy.go {代码...} NamingProxy定义了clientConfig、nacosServer属性 NewNamingProxy nacos-sdk-go-v0.3.2/clients/naming_client/naming_proxy.go {代码...} NewNamingProxy通过nacos_server.NewNacosSer...

BeatReactor定义了beatMap、serviceProxy、clientBeatInterval、beatThreadCount、beatThreadSemaphore、beatRecordMap属性

HostReactor定义了serviceInfoMap、cacheDir、updateThreadNum、serviceProxy、pushReceiver、subCallback、updateTimeMap、updateCacheWhenEmpty属性

NamingClient定义了hostReactor、serviceProxy、subCallback、beatReactor、indexMap属性

序 本文主要研究一下kingbus的resp.go writeOK kingbus/mysql/resp.go {代码...} writeOK方法写入gomysql.OK_HEADER writeError kingbus/mysql/resp.go {代码...} writeError方法写入gomysql.ERR_HEADER writeEOF kingbus/mysql/resp.go {代码...} writeEOF方法写入gomysql.EOF_HEADER writeResultset kingbus/mysql/res...

BinlogProgress定义了currentGtid、lastSaveGtid、lastBinlogFile、lastFilePosition、executedGtidSetStr、trxBoundaryParser、persistentTime、persistentAppliedIndex、executedGtidSet、store属性

DumpBinlogAt方法通过s.store.NewEntryReaderAt(startRaftIndex)获取reader，然后获取nextRaftIndex，之后通过utils.DecodeBinlogEvent(raftEntry)获取event，然后通过s.skipEvent(event, slaveGtids, inExcludeGroup)来判断是否该skip，之后将event写入到eventC

序 本文主要研究一下kingbus的command.go Close kingbus/mysql/command.go {代码...} Close方法执行c.closed.Store(true)、c.Conn.Close()、c.cancel() handleQuery kingbus/mysql/command.go {代码...} handleQuery方法通过sqlParser来解析stmt，然后根据stmt的type来执行不同的方法；对于ShowStmt执行c.handleShow(v)...

序 本文主要研究一下kingbus的startMasterServer startMasterServer kingbus/server/server.go {代码...} startMasterServer方法先执行NewBinlogServer创建master，然后执行master的Start方法 NewBinlogServer kingbus/server/binlog_server.go {代码...} NewBinlogServer方法通过new方法创建BinlogServer，之后设置其li...

StartBinlogServer方法主要是通过h.svr.StartServer(config.BinlogServerType, &args)来启动binlog server

StartBinlogSyncer方法先执行echoCtx.Bind(&args)，然后针对h.svr.IsLeader()为false的通过h.sendToLeader("PUT", "/binlog/syncer/start", req)将请求转发给leader；为true的则执行h.svr.StartServer(config.SyncerServerType, &args)启动syncer server，然后执行h.ProposeSyncerArgs(&args)启动propose sy...

序 本文主要研究一下kingbus的membership_handler.go GetMembers kingbus/api/membership_handler.go {代码...} GetMembers方法主要是通过h.cluster.Members()获取members，然后json化 AddMember kingbus/api/membership_handler.go {代码...} AddMember方法先通过h.svr.IsLeader()判断isLeader，为false的话通过h.sendT...

序 本文主要研究一下kingbus的startAdminServer startAdminServer kingbus/server/server.go {代码...} startAdminServer方法主要是执行api.NewAdminServer NewAdminServer kingbus/api/api_server.go {代码...} NewAdminServer创建adminServer，并设置了MembershipHandler、BinlogSyncerHandler、BinlogServerHandler R...

startRaftPeer方法根据peerURLs通过s.raftNode.NewPeerListener开启listener，之后执行s.raftNode.SetPeerHandler()、s.raftNode.PeerHandlerServe()

starRaft方法先通过rafthttp.NewRoundTripper创建http.RoundTripper，之后通过storage.NewDiskStorage创建DiskStorage，之后根据logExist及cfg.NewCluster做不同处理；若二者都为false则更新membership.RaftCluster的id为存在的cluster的id，然后执行startEtcdRaftNode；若cfg.NewCluster为true则使用cl.MemberIDs()来执...