Fabric 1.4 orderer 服务启动流程
1.提要
orderer提供broadcast和deliver两个服务接口。orderer节点与各个peer节点通过grpc连接，orderer将所有peer节点通过broadcast发来的交易（Envelope格式，比如peer部署后的数据）按照配置的大小依次封装到一个个block中并写入orderer自己的账本中，然后供各个peer节点的gossip服务通过deliver来消费这个账本中的数据进行自身结点账本的同步。
2.初始化过程
先看看main函数。
// Main is the entry point of orderer process
func Main() {
fullCmd := kingpin.MustParse(app.Parse(os.Args[1:]))
// "version" command
if fullCmd == version.FullCommand() {
fmt.Println(metadata.GetVersionInfo())
return
}
conf, err := localconfig.Load()
if err != nil {
logger.Error("failed to parse config: ", err)
os.Exit(1)
}
initializeLogging()
initializeLocalMsp(conf)
prettyPrintStruct(conf)
Start(fullCmd, conf)
}
从中可知 orderer服务命令行是通过kingpin来实现的，基本上只是简单使用了下，也只实现了3个命令:
start*
Start the orderer node
version
Show version information
benchmark
Run orderer in benchmark mode
并且从上述main函数可知，仅version有对应操作，而orderer 默认为orderer start。
启动流程为：
- 加载配置（orderer.yaml/Defaults/环境变量）
- 初始化log（log级别和log格式）
- 初始化本地msp
- 启动服务Start()
接下来主要关注第4步。前面基本上是配置初始化第过程。
查看一下start函数：
- 从配置文件启动块路径获取配置块及验证是否可作为配置块（系统通道第一个块）
- 集群相关初始化配置
- 判断是否是raft共识及使用的是最新配置块，如果是，则进行下列流程：
- 获取所有应用链及其创世区块块（discoverChannels）
- 根据orderer是否在应用链配置块的raft节点中分类（channelsToPull topull/nottopull）
- 创建所有的应用通道账本
- 获取topull应用通道的账本(从orderer处获取)
- 获取系统通道账本
if clusterType && conf.General.GenesisMethod == "file" {
r.replicateIfNeeded(bootstrapBlock)
}
func (r *Replicator) ReplicateChains() []string {
var replicatedChains []string
channels := r.discoverChannels()
pullHints := r.channelsToPull(channels)
totalChannelCount := len(pullHints.channelsToPull) + len(pullHints.channelsNotToPull)
for _, channels := range [][]ChannelGenesisBlock{pullHints.channelsToPull, pullHints.channelsNotToPull} {
for _, channel := range channels {
...
r.appendBlock(gb, ledger, channel.ChannelName)
}
}
for _, channel := range pullHints.channelsToPull {
err := r.PullChannel(channel.ChannelName)
...
}
// Last, pull the system chain.
if err := r.PullChannel(r.SystemChannel); err != nil && err != ErrSkipped {
r.Logger.Panicf("Failed pulling system channel: %v", err)
}
return replicatedChains
}
-
启动及初始化必要模块
- 创建系统链
// Are we bootstrapping? if len(lf.ChainIDs()) == 0 { initializeBootstrapChannel(genesisBlock, lf) } else { logger.Info("Not bootstrapping because of existing chains") }-
多通道初始化(initializeMultichannelRegistrar)
- 初始化registrar实例
registrar := multichannel.NewRegistrar(lf, signer, metricsProvider, callbacks...)// Registrar serves as a point of access and control for the individual channel resources. type Registrar struct { lock sync.RWMutex //当前所有通道的chain对象 chains map[string]*ChainSupport //不同共识类型的consenter consenters map[string]consensus.Consenter //Factory通过chainID检索或创建新的分类帐 ledgerFactory blockledger.Factory //签名相关 signer crypto.LocalSigner blockcutterMetrics *blockcutter.Metrics //系统链id systemChannelID string //系统链chainSupport systemChannel *ChainSupport //通道配置模版 templator msgprocessor.ChannelConfigTemplator callbacks []channelconfig.BundleActor }- 初始化共识机制
consenters["solo"] = solo.New() var kafkaMetrics *kafka.Metrics consenters["kafka"], kafkaMetrics = kafka.New(conf, metricsProvider, healthChecker, registrar) go kafkaMetrics.PollGoMetricsUntilStop(time.Minute, nil) if isClusterType(bootstrapBlock) { initializeEtcdraftConsenter(consenters, conf, lf, clusterDialer, bootstrapBlock, ri, srvConf, srv, registrar, metricsProvider) }- 启动orderer现存的链（系统链/应用链，通过读取链的目录查看现存链），为每条链实例化了ChainSupport对象，然后启动
chain := newChainSupport( r, ledgerResources, r.consenters, r.signer, r.blockcutterMetrics, )for _, chainID := range existingChains { ... chain.start() ... } -
启动GRPC服务
server.go中的服务端对象实例server在main.go的main()中由server := NewServer(manager, signer)生成，使用ab.RegisterAtomicBroadcastServer(grpcServer.Server(), server)进行了注册，随后grpcServer.Start()启动起来。
其主要的两个接口为：
type AtomicBroadcastServer interface { Broadcast(AtomicBroadcast_BroadcastServer) error Deliver(AtomicBroadcast_DeliverServer) error }其接口的实现在：orderer/common/server/server.go
3.相关模块介绍
3.1 ChainSupport
提供支持chain相关操作的资源，既包含账本本身，也包含了账本用到的各种工具对象，如分割工具cutter，签名工具signer，最新配置在chain中的位置信息（lastConfig的值代表当前链中最新配置所在的block的编号，lastConfigSeq的值则代表当前链中最新配置消息自身的编号）等
type ChainSupport struct {
// 账本相关资源 包括账本的读写及配置的获取
*ledgerResources
// 提供从客户端获取消息分类处理接口
msgprocessor.Processor
// 将区块写入磁盘
*BlockWriter
// 链 提供对messages对处理方法
//This design allows for two primary flows
// 1. Messages are ordered into a stream, the stream is cut into blocks, the blocks are committed (solo, kafka)
// 2. Messages are cut into blocks, the blocks are ordered, then the blocks are committed (sbft)
consensus.Chain
// 广播消息接收器 提供切块方法
cutter blockcutter.Receiver
//签名
crypto.LocalSigner
// chains need to know if they are system or standard channel.
systemChannel bool
}
3.2 blockcutter模块
- 块分割工具，用于分割block，具体为orderer/common/blockcutter/blockcutter.go中定义的receiver。一条一条消息数据在流水线上被传送到cutter处，cutter按照configtx.yaml中的配置，把一条条消息打包成一批消息，同时返回整理好的这批消息对应的committer集合，至此cutter的任务完成。
MaxMessageCount指定了block中最多存储的消息数量
AbsoluteMaxBytes指定了block最大的字节数
PreferredMaxBytes指定了一条消息的最优的最大字节数（blockcutter处理消息的过程中会努力使每一批消息尽量保持在这个值上）。
- 若一个Envelope的数据大小（Payload+签名）大于PreferredMaxBytes时，无论当前缓存如何，立即Cut；
- 若一个Envelope被要求单纯存储在一个block（即该消息对应的committer的Isolated()返回为true），要立即Cut
- 若一个Envelope的大小加上blockcutter已有的消息的大小之和大于PreferredMaxBytes时，要立即Cut；
- 若blockcutter当前缓存的消息数量大于MaxMessageCount了，要立即Cut。
- 由configtx.yaml中BatchTimeout配置项（默认2s）控制，当时间超过此值，chain启动的处理消息的流程中主动触发的Cut。
3.3 chain start（）模块
主要是对消息进行处理，将交易消息传输给block cutter切成块及写入账本。不同的共识机制操作不同。（后续结合consenter模块一起详细介绍）
chain := newChainSupport(
r,
ledgerResources,
r.consenters,
r.signer,
r.blockcutterMetrics,
)
r.chains[chainID] = chain
chain.start()
3.4 consenter模块:
solo/kafka/etcdraft三种共识类型，用于序列化生产（即各个peer点传送来的Envelope）出来的消息。
参考：
https://blog.csdn.net/idsuf698987/article/details/78639203
