fabric创建通道过程

本文主要是介绍fabric创建通道过程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

创建通道

1. Client 处理过程
- 1.1. 发送创建通道的交易
- 1.2. 获取通道的创世区块
2. Orderer 处理过程
- 2.1. 处理消息
3. 参考资料

创建通道的过程

1. Client 处理过程

internal/peer/channel/create.go

func executeCreate(cf *ChannelCmdFactory) error {// 发送创建通道的交易err := sendCreateChainTransaction(cf)if err != nil {return err}// 获取通道的创世区块block, err := getGenesisBlock(cf)if err != nil {return err}b, err := proto.Marshal(block)if err != nil {return err}file := channelID + ".block"if outputBlock != common.UndefinedParamValue {file = outputBlock}// 保存通道创世区块信息err = ioutil.WriteFile(file, b, 0o644)if err != nil {return err}return nil
}

1.1. 发送创建通道的交易

internal/peer/channel/create.go

func sendCreateChainTransaction(cf *ChannelCmdFactory) error {var err errorvar chCrtEnv *cb.Envelope// 如果传入创建通道的交易信息，则读取并解析if channelTxFile != "" {if chCrtEnv, err = createChannelFromConfigTx(channelTxFile); err != nil {return err}} else { // 如果没有传入，则根据默认的配置yaml进行创建if chCrtEnv, err = createChannelFromDefaults(cf); err != nil {return err}}// 对交易信息进行必要检查，通过后对消息进行签名if chCrtEnv, err = sanityCheckAndSignConfigTx(chCrtEnv, cf.Signer); err != nil {return err}var broadcastClient common.BroadcastClientbroadcastClient, err = cf.BroadcastFactory()if err != nil {return errors.WithMessage(err, "error getting broadcast client")}defer broadcastClient.Close()// 发送封装好的消息err = broadcastClient.Send(chCrtEnv)return err
}

1.2. 获取通道的创世区块

每隔 200 毫秒尝试从排序节点获取一次通道的创世区块，即区块号为 0 的区块。

internal/peer/channel/create.go

func getGenesisBlock(cf *ChannelCmdFactory) (*cb.Block, error) {timer := time.NewTimer(timeout)defer timer.Stop()for {select {case <-timer.C:cf.DeliverClient.Close()return nil, errors.New("timeout waiting for channel creation")default:// 获取创世区块if block, err := cf.DeliverClient.GetSpecifiedBlock(0); err != nil {cf.DeliverClient.Close()cf, err = InitCmdFactory(EndorserNotRequired, PeerDeliverNotRequired, OrdererRequired)if err != nil {return nil, errors.WithMessage(err, "failed connecting")}// 休眠 200 毫秒time.Sleep(200 * time.Millisecond)} else {cf.DeliverClient.Close()return block, nil}}}
}

2. Orderer 处理过程

orderer/common/broadcast/broadcast.go

// Handle reads requests from a Broadcast stream, processes them, and returns the responses to the stream
func (bh *Handler) Handle(srv ab.AtomicBroadcast_BroadcastServer) error {addr := util.ExtractRemoteAddress(srv.Context())logger.Debugf("Starting new broadcast loop for %s", addr)for {// 接收消息msg, err := srv.Recv()if err == io.EOF {logger.Debugf("Received EOF from %s, hangup", addr)return nil}if err != nil {logger.Warningf("Error reading from %s: %s", addr, err)return err}// 处理消息resp := bh.ProcessMessage(msg, addr)// 发送处理结果err = srv.Send(resp)if resp.Status != cb.Status_SUCCESS {return err}if err != nil {logger.Warningf("Error sending to %s: %s", addr, err)return err}}
}

2.1. 处理消息

orderer/common/broadcast/broadcast.go

// ProcessMessage validates and enqueues a single message
func (bh *Handler) ProcessMessage(msg *cb.Envelope, addr string) (resp *ab.BroadcastResponse) {tracker := &MetricsTracker{ChannelID: "unknown",TxType:    "unknown",Metrics:   bh.Metrics,}defer func() {// This looks a little unnecessary, but if done directly as// a defer, resp gets the (always nil) current state of resp// and not the return valuetracker.Record(resp)}()tracker.BeginValidate()// 获取通道消息处理器，如果是创建通道，则调用默认的系统通道管理器chdr, isConfig, processor, err := bh.SupportRegistrar.BroadcastChannelSupport(msg)if chdr != nil {tracker.ChannelID = chdr.ChannelIdtracker.TxType = cb.HeaderType(chdr.Type).String()}if err != nil {logger.Warningf("[channel: %s] Could not get message processor for serving %s: %s", tracker.ChannelID, addr, err)return &ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST, Info: err.Error()}}if !isConfig {logger.Debugf("[channel: %s] Broadcast is processing normal message from %s with txid '%s' of type %s", chdr.ChannelId, addr, chdr.TxId, cb.HeaderType_name[chdr.Type])configSeq, err := processor.ProcessNormalMsg(msg)if err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of normal message from %s because of error: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: ClassifyError(err), Info: err.Error()}}tracker.EndValidate()tracker.BeginEnqueue()if err = processor.WaitReady(); err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of message from %s with SERVICE_UNAVAILABLE: rejected by Consenter: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE, Info: err.Error()}}err = processor.Order(msg, configSeq)if err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of normal message from %s with SERVICE_UNAVAILABLE: rejected by Order: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE, Info: err.Error()}}} else { // 对于创建通道，isConfig 值为 true，执行以下逻辑logger.Debugf("[channel: %s] Broadcast is processing config update message from %s", chdr.ChannelId, addr)// 对消息进行重新封装，头部消息类型为ORDERER_TRANSACTION，内层消息类型为CONFIGconfig, configSeq, err := processor.ProcessConfigUpdateMsg(msg)if err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of config message from %s because of error: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: ClassifyError(err), Info: err.Error()}}tracker.EndValidate()tracker.BeginEnqueue()if err = processor.WaitReady(); err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of message from %s with SERVICE_UNAVAILABLE: rejected by Consenter: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE, Info: err.Error()}}// 将配置更新消息进行广播err = processor.Configure(config, configSeq)if err != nil {logger.Warningf("[channel: %s] Rejecting broadcast of config message from %s with SERVICE_UNAVAILABLE: rejected by Configure: %s", chdr.ChannelId, addr, err)return &ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE, Info: err.Error()}}}logger.Debugf("[channel: %s] Broadcast has successfully enqueued message of type %s from %s", chdr.ChannelId, cb.HeaderType_name[chdr.Type], addr)return &ab.BroadcastResponse{Status: cb.Status_SUCCESS}
}