Fabric源码流程分析之Orderer篇

导言：
　　本文使用fabric1.1版本,此时有小朋友会问了，fabric都出1.4.2了你怎么还在看1.1呢！首先fabric自1.0以后大的架构基本没有变化，小版本升级只是功能性上更加丰满了，当然最重要的是有一本对源码解读超详细的书《Hyperledger Fabric技术内幕：架构设计与实现原理》使用的就是1.1版本，不懂的地方看看这本书就会茅塞顿开，强烈推荐嗷！,如果阅读源码有障碍，可阅读本人使用的版本，其中有大量注释，不过因为是随手笔记有些许潦草 https://github.com/mikesen1994/FabricSourceAnalyze
　　在直接阅读源代码前，如果您只运行过官方docker-compose的一键部署，那么强烈建议手动搭建一套Fabric网络，这样会更加了解Fabric的运行细节，对阅读源码很有帮助，推荐博客 https://www.lijiaocn.com/%E9%A1%B9%E7%9B%AE/2018/04/26/hyperledger-fabric-deploy.html
　　本文不会细讲源码中每个函数每个方法，只是讲解orderer服务运行的大体流程以及官方使用了哪些开源工具

参考书籍：《Hyperledger Fabric技术内幕：架构设计与实现原理》
　　　　　《深度探索区块链 Hyperledger技术与应用》

orderer模块入口：hyperledger\fabric\orderer\common\server\main.go

一.首先通过kingpin技术（命令行解析工具）接收命令行命令

比如在orderer文件存在的目录命令行直接敲：
　　　orderer start 　　　　　//启动orderer模块
　　　orderer version 　　　　//打印当前版本信息
　　　orderer benchmark　　 //测试运行orderer
　　　orderer help 　　　　　//显示帮助信息

    fullCmd := kingpin.MustParse(app.Parse(os.Args[1:]))if fullCmd == version.FullCommand() {   //如果接收到version命令直接打印当前orderer版本fmt.Println(metadata.GetVersionInfo())return}

二.然后通过viper技术(管理配置信息的工具)加载配置信息，并将当前目录下的的orderer.yaml文件转化为具体配置信息的结构体

    conf, err := config.Load()   ——>  config := viper.New())   ——>  cf.InitViper(config, configName)

下图为本人实操时orderer的配置信息，orderer启动时实际转化的就是此文件

下图为转化后的结构体，可以看到结构体与orderer.yaml的配置信息很匹配

目录：hyperledger\fabric\orderer\common\localconfig\config.go

    type TopLevel struct {General    GeneralFileLedger FileLedgerRAMLedger  RAMLedgerKafka      KafkaDebug      Debug}

三.通过配置信息初始化日志等级，fabric的日志管理主要使用了第三方包go-logging（主要特点是可以根据设置的日志等级来显示不同的颜色，以及可以直接定位到产生日志的代码位置），部分使用了go语言标准库中的log。在此基础上fabric自己封装出来了flogging，默认的日志等级是debug

    initializeLoggingLevel(conf)

四.通过配置信息加载并初始化Msp文件

    initializeLocalMsp(conf) ——> mspmgmt.LoadLocalMsp(conf.General.LocalMSPDir, conf.General.BCCSP, conf.General.LocalMSPID)

主要用到的就是orderer.yaml配置文件里的以下三个信息：

五.将命令与配置信息传递给Start函数

      Start(fullCmd, conf)

六.创建本地MSP签名者实例，实际就是一个空的mspSigner结构体指针

      signer := localmsp.NewSigner()

七.通过TLS相关的配置信息，初始化TLS认证需要的的安全服务器配置项

      serverConfig := initializeServerConfig(conf)

获得的serverConfig由两种结构体组成SecureOptions与KeepaliveOptions
SecureOptions保存了用于tls验证的公钥私钥，服务器CA证书，客户端CA证书等。
KeepaliveOptions则是用于设置grpc双方通讯的配置的信息包括客户端等待响应时间，服务端等待响应时间，客户端通讯响应间隔等信息

八.初始化grpc服务，传入配置文件与TLS配置信息，通过TLS配置信息来决定是否启用tls验证

     grpcServer := initializeGrpcServer(conf, serverConfig)

九.使用serverConfig的客户端CA证书列表构造CA证书支持组件对象，在注册多通道管理器对象时需要用到

    caSupport := &comm.CASupport{   //构造CA证书支持组件对象AppRootCAsByChain:     make(map[string][][]byte), //application根CA证书字典OrdererRootCAsByChain: make(map[string][][]byte), //Orderer根CA证书字典ClientRootCAs:         serverConfig.SecOpts.ClientRootCAs,  //设置TLS认证的客户端CA证书列表}

十.设置TLS连接认证的回调函数，用于更新每个通道的TLS客户端CA证书

    tlsCallback := func(bundle *channelconfig.Bundle) {// only need to do this if mutual TLS is requiredif grpcServer.MutualTLSRequired() {   //检测是否需要TLS证书logger.Debug("Executing callback to update root CAs")updateTrustedRoots(grpcServer, caSupport, bundle)  //执行回调函数更新根CA证书}}

十一.初始化多通道管理器对象（核心步骤）

    manager := initializeMultichannelRegistrar(conf, signer, tlsCallback)

1.首先根据配置信息创建orderer数据存储目录（存储目录包含了索引数据库，以及每个通道的区块数据）

根据下图的配置信息生成路径

    ld = conf.FileLedger.Location //orderer数据的存储位置信息if ld == "" {ld = createTempDir(conf.FileLedger.Prefix)  //如果没有设置存储位置则在当前目录创建一个临时目录}

2.创建一个新的账本工厂对象

    lf = fileledger.New(ld)

实际创建的是实现了区块账本工厂接口的结构体blockledger.Factory---->fileLedgerFactory

    type fileLedgerFactory struct {blkstorageProvider blkstorage.BlockStoreProvider   //通道账本仓库提供器ledgers            map[string]blockledger.ReadWriter    //存储了所有通道账本的读写句柄mutex              sync.Mutex}

blkstorage.BlockStoreProvider —> FsBlockstoreProvider，FsBlockstoreProvider实现了BlockStoreProvider并创建了索引数据库（调用leveldb创建一个以DBPath(orderer_data/index)为目录的db操作句柄）

    blkstorageProvider: fsblkstorage.NewProvider(fsblkstorage.NewConf(directory, -1),&blkstorage.IndexConfig{AttrsToIndex: []blkstorage.IndexableAttr{blkstorage.IndexableAttrBlockNum}},)func NewProvider(conf *Conf, indexConfig *blkstorage.IndexConfig) blkstorage.BlockStoreProvider {p := leveldbhelper.NewProvider(&leveldbhelper.Conf{DBPath: conf.getIndexDir()})   //DBPath:orderer_data目录下的index目录return &FsBlockstoreProvider{conf, indexConfig, p}}

ledgers则定义了所有通道账本的读写句柄 key为通道名称chainid value为blockledger.ReadWriter接口（blockledger.ReadWriter接口定义了区块账本的读写方法）

3.在orderer数据目录下创建区块链目录例：/orderer_data/chains

    createSubDir(ld, fsblkstorage.ChainsDir)

4.将创世区块文件解码为定义的区块结构体（创世区块文件，需要手动用官方提供的工具configtxgen生成，创建语句为configtxgen -profile TwoOrgsOrdererGenesis -channelID byfn-sys-channel -outputBlock ./channel-artifacts/genesis.block,创世区块的主要作用是定义了系统通道的共识算法，准入的组织名称，策略等信息)

genesisBlock = file.New(conf.General.GenesisFile).GenesisBlock()

下图为本人实操过程中，用工具生成的创世区块文件

目录：hyperledger\fabric\protos\common\common.pb.go

    type Block struct {//区块头 包含区块高度，上一个区块的哈希值，本区块的哈希值Header   *BlockHeader   `protobuf:"bytes,1,opt,name=header" json:"header,omitempty"`//交易数据集合，封装了打包的交易集合Data     *BlockData     `protobuf:"bytes,2,opt,name=data" json:"data,omitempty"`/*区块元数据，封装了如下4个元数据索引项·BlockMetadataIndex_SIGNATURES：区块签名；·BlockMetadataIndex_LAST_CONFIG：最新配置区块的区块号；·BlockMetadataIndex_TRANSACTIONS_FILTER：最新交易过滤器，封装了交易数据集合Data中所有交易对应的交易验证码，标识其交易的有效性。·BlockMetadataIndex_ORDERER：Orderer配置信息，如Kafka共识组件的初始化参数。*/Metadata *BlockMetadata `protobuf:"bytes,3,opt,name=metadata" json:"metadata,omitempty"`}

5.通过Block的结构体实例获得创世通道名称（创世通道名称默认应该为genesis）

    chainID, err := utils.GetChainIDFromBlock(genesisBlock)

6.通过创世通道名称获取一个新的区块账本fsBlockStore对象。以及创建indexStoreHandle对象，indexStoreHandle对象为DBHandle类型（DBHandle的作用是根据通道名称获得索引数据库的处理句柄）

    gl, err := lf.GetOrCreate(chainID) ——>blkstorageProvider.OpenBlockStore(chainID)——>newFsBlockStore(chainID, p.conf, p.indexConfig, indexStoreHandle)type fsBlockStore struct {id      string //通道名称conf    *Conf   //包含orderer数据的存储位置信息，以及每个区块文件最大大小，默认是64MBfileMgr *blockfileMgr}

下图为索引数据库的操作句柄结构体：

    type DBHandle struct {dbName string    //通道名称db     *DB       //封装好的leveldb底层}

7.创建fsBlockStore对象的同时创建了一个blockfileMgr对象（创建此对象时进行了一系列重要操作，包括生成区块文件，添加索引数据，生成检查点信息等等）

    type blockfileMgr struct {rootDir           string    //orderer_data/chains下的chainId目录conf              *Conf     //此结构体两个成员 1.orderer_data的目录地址  2.最大区块文件大小（默认64MB）db                *leveldbhelper.DBHandle   //通过leveldbProvider的gethandle方法传入chainID 来获得一个levelDB的处理句柄，该句柄绑定了通道名称与一个已经打开的数据库操作对象index             index             //实际是一个blockIndex结构体 包含k,v形式的map blockNum->true 和一个db的处理句柄 该结构体实现了Index接口cpInfo            *checkpointInfo  //区块检查点信息cpInfoCond        *sync.Cond       //基于互斥锁的 加强版双开关锁currentFileWriter *blockfileWriter //其中包含区块文件的实际路径与区块文件的可操作性句柄bcInfo            atomic.Value      //区块链简要信息 包含当前区块高度  当前区块hash 上一个区块hash}

以下为创建blockfileMgr对象的重要步骤：

（1）扫描区块文件，获得区块的检查点信息checkpointInfo

    cpInfo, err = constructCheckpointInfoFromBlockFiles(rootDir)

检查点结构体如下：

    type checkpointInfo struct {latestFileChunkSuffixNum int     //最新区块的文件名后缀编号latestFileChunksize      int     //最新区块文件的字节数isChainEmpty             bool    //是否为空链lastBlockNumber          uint64  //最新区块文件的最新区块号}

（2）将检查点信息cpInfo，利用 proto.NewBuffer编码成[]byte类型，并存入索引数据库，其中key值为"blkMgrInfo"字符串的[]byte类型

    err = mgr.saveCurrentInfo(cpInfo, true)

（3）获得currentFileWriter对象其中包含区块文件的实际路径与区块文件的可操作性句柄，并存入blockfileMgr对象中

currentFileWriter, err := newBlockfileWriter(deriveBlockfilePath(rootDir, cpInfo.latestFileChunkSuffixNum))

（4）循环读取每个区块的数据，并将每个区块的区块信息中的区块序号与偏移量存入索引数据库，存入的键值对为 key:dbName+索引类型+区块文件号+区块序号 value:区块位置的偏移量（某个区块在区块文件中的位置）
封装后的leveldb数据库操作句柄，每次操作时，在存入数据库的key值前面加上dbName

    type DBHandle struct {dbName string //通道名称db     *DB   //封装好的leveldb底层}

索引类型：

    const (blockNumIdxKeyPrefix           = 'n'blockHashIdxKeyPrefix          = 'h'txIDIdxKeyPrefix               = 't'blockNumTranNumIdxKeyPrefix    = 'a'blockTxIDIdxKeyPrefix          = 'b'txValidationResultIdxKeyPrefix = 'v'indexCheckpointKeyStr          = "indexCheckpointKey")

区块文件号（默认64MB,超过则自动新增文件例：blockfile_000000 blockfile_000001 blockfile_000002）：

区块信息：

    type blockIdxInfo struct {blockNum  uint64 //区块序号blockHash []byte //区块头哈希值flp       *fileLocPointer //区块位置偏移量txOffsets []*txindexInfo  //交易索引信息列表（含交易ID与位置指针）metadata  *common.BlockMetadata  //区块元数据}

8.将fsBlockStore对象存入账本工厂对象fileLedgerFactory的ledger map类型中

    flf.ledgers[key] = ledger

9.通过创世通道名称获取到该通道账本句柄，将创世区块加入到账本中

    gl.Append(genesisBlock) ——> fsBlockStore.AddBlock(block *common.Block)——>blockfileMgr.addBlock(block *common.Block)

主要步骤如下：
1.首先利用proto技术，将创世区块结构体序列化为字节数组
2.根据文件大小确定是否需要新建新的区块文件(默认64MB)
3.将此区块的长度大小加入到区块文件中
3.将区块的字节数组加入到区块文件中
4.索引数据库更新新的索引信息
5.更新检查点信息以及更新区块链简要信息

10.创建并设置共识组件字典,包含solo排序共识组件以及kafka排序共识组件

    consenters := make(map[string]consensus.Consenter)  //创建并设置共识组件字典consenters["solo"] = solo.New()  //solo排序共识组件consenters["kafka"] = kafka.New(conf.Kafka) //kafka排序共识组件

11.传入账本工厂对象，共识组件字典等信息，创建多通道注册管理器对象

    multichannel.NewRegistrar(lf, consenters, signer, callbacks...)

首先传入账本工厂对象fileLedgerFactory，通过账本工厂对象循环获取每个通道的名称，然后通过通道名称获得每个区块的账本对象，并构建出链支持对象chainsupport，将通道名称、链支持对象chainsupport以键值对形式进行存储（chainsupport结构体非常重要，包含了对此通道账本的读写操作，签名配置，消息切割组件，消息过滤组件和共识排序）。最后依次通过共识组件接口启动通道

    type ChainSupport struct {*ledgerResources  /*账本资源对象封装了通道配置资源对象（configResources类 型）与区块账本对象（FileLedger类型）*/msgprocessor.Processor /*负责过滤处理应用通道上的消息，以筛选出符合 通道要求的消息，默认初始化4个标准通道消息过滤器，即Empty-RejectRule拒绝空消息过滤器、 expirationRejectRule拒绝过期的签名者身份证书的过滤器、MaxBytesRule验证消息最大字节数（默认 98MB）的过滤器和sigFilter验证消息签名是否满足ChannelWriters（/Channel/Writers）通道写权限策略 要求的过滤器。*/*BlockWriter /*负责构造新区块并向账本提交区块文件，同时创建新的应 用通道与更新通道配置。该对象在初始化时设置最新的区块号lastBlock、通道配置序号lastConfigSeq、 最新的配置区块号lastConfigBlockNum、多通道注册管理器Registrar对象（用于创建新的应用通道）以 及关联通道的链支持对象（用于更新通道配置）。*/consensus.Chain /*采用共识排序后端对交易排序，再添加到缓存交易消 息列表，同时利用链支持对象上的消息切割组件、通道消息处理器、区块账本写组件等模块执行打包 出块、通道管理等操作。*/cutter blockcutter.Receiver  /*消息切割组件获取指定通道上 的Orderer配置，包含共识组件类型、交易出块周期时间、区块最大字节数、通道限制参数（如通道数 量）等。接着，基于该配置创建消息切割组件（receiver类型），将本地的缓存交易消息列表按照交易 出块规则切割成批量交易集合（[]*cb.Envelope类型），再交由区块账本写组件构造新区块，并提交到 账本区块文件*/crypto.LocalSigner // 本地签名者}chain := newChainSupport(  // 构造应用通道的链支持对象r,ledgerResources,consenters,signer)r.chains[chainID] = chain // 将链支持对象注册到多通道管理器chain.start()  // 启动链支持对象

十二.设置TLS双向认证标志位（需要客户端与服务端均开启TLS认证）

    mutualTLS := serverConfig.SecOpts.UseTLS && serverConfig.SecOpts.RequireClientCert

十三.通过传入多通道管理器，是否进行TLS双向认证信息等信息创建Orderer排序服务器（其中包含两个重要服务，广播与分发服务）

    server := NewServer(manager, signer, &conf.Debug, conf.General.Authentication.TimeWindow, mutualTLS)

十四.将orderer排序服务器注册到grpc服务器上(利用了protoc技术自动生成的pb.go文件)

    ab.RegisterAtomicBroadcastServer(grpcServer.Server(), server)

十五.启动grpc监听peer节点请求，Orderer服务的启动到此算是结束了

     grpcServer.Start()

下面来讲讲orderer节点作为服务端提供的广播与分发服务

一.广播服务Broadcast

服务端通过解析客户端发来的消息，区分为正常交易消息或配置消息，其中配置消息又分为新增通道消息与更新已有通道配置信息。消息写入区块成功后向客户端返回操作成功的响应信息200。

正常交易信息
解析消息获取通道名称，根据通道名称查找本地账本文件，利用共识组件的order排序方法，利用先进先出的原则写入区块文件。
新增通道消息
调用newChainSupport()方法创建新应用通道的链支持对象，并注册到通道字典中，然后启动start方法提供正常服务。
更新已有通道配置消息
调用bw.support.CreateBundle()方法，构造新的通道配置实体对象bundle（Bundle类型）。然后，调用bw.support.Update(bundle)方法，将其更新为当前链支持对象管理的底层通道配置实体，而不需要直接修改多通道注册管理器上的链支持对象字典chains。最后调用bw.WriteBlock(block，encodedMetadataValue)方法，更新配置区块的元数据，包括BlockMetadataIndex_ORDERER、BlockMetadataIndex_SIGNATURES、BlockMetadataIndex_LAST_CONFIG等索引项，再将该配置区块写入系统通道（创建新应用通道的情况）或应用通道（更新通道配置的情况）的账本中。

二. 区块分发服务deliver

通过解析客户端发来的请求信息，获取账本上的区块数据，回复给请求客户端。如果还未生成请求的区块，则阻塞等待直到该区块创建提交完成

接收客户端发送的请求消息
解析客户端发来的消息并进行各种验证，包括消息的完整性，是否有通道的读权限等检查操作
通过解析请求消息后，获得的通道名称来获取通道的支持对象
通过请求信息的解析，获取到seekInfo对象（包含请求下发的区块的开始与结束位置）
读取区块数据处理循环，利用seekInfo对象依次分发区块数据
请求的区块全部分发完毕，返回操作成功的响应信息200。

建了个QQ交流群：722124200 有问题可以加群互相讨论：）
邮箱:mikesen1994@gmail.com vx:965952482