上一节分析reciept产生过程的时候提到：reciept会为日志数据生成一个Bloom过滤器，那Bloom过滤器是用来干嘛的呢？有什么用呢？
一，Bloom过滤器的数据结构和reciept创建Bloom的过程 type Bloom [BloomByteLength]byte BloomByteLength = 256 Bloom 就是一个256个字节数组。一共2048位。
我们看看怎么把庞大的收据日志数据放到bloom过滤器里面的。
func CreateBloom(receipts Receipts) Bloom { bloomBin := new(big.Int) for _, receipt := range receipts { bloomBin.Or(bloomBin, LogsBloom(receipt.Logs)) } return BytesToBloom(bloomBin.Bytes()) } func LogsBloom(logs []*Log) *big.Int { bin := new(big.Int) for _, log := range logs { bin.Or(bin, bloom9(log.Address.Bytes())) for _, b := range log.Topics { bin.Or(bin, bloom9(b[:])) } } return bin } func bloom9(b []byte) *big.Int { b = crypto.Keccak256(b[:]) r := new(big.Int) for i := 0; i < 6; i += 2 { t := big.NewInt(1) b := (uint(b[i+1]) + (uint(b[i]) << 8)) & 2047 r.Or(r, t.Lsh(t, b)) } return r }1，先看看bloom9(b []byte)算法函数。 1.1， 首先将传入的数据，进行hash256的运算，得到一个32字节的hash 1.2，然后取第0和第1字节的值合成一个2字节无符号的int，和2047做按位与运算，得到一个小于2048的值b，这个值就表示bloom里面第b位的值为1。同理取第2,3 和第4,5字节合成另外两个无符号int，增加在bloom里面的命中率。 1.3，也就是说对于任何一个输入，如果它对应的三个下标的值不都为1，那么它肯定不在这个区块中。 当如如果对应的三位都为1，也不能说明一定在这个区块中。 这就是布隆过滤器的特性。 1.4，这三个数取或，得到一个bigInt，代表这个传参数据的bloom9值。
2，LogsBloom(logs []*Log)方法把日志数据转成对应的bloom9值，包括日志的合约地址以及每个日志Topic
3，CreateBloom(receipts Receipts)方法创建收据的bloom 3.1，创建一个空的bigInt bloomBin，遍历receipts，取得receipt里的日志，调用LogsBloom(receipt.Logs)将取得所有日志的bloom值按位或和到bloomBin。这意味着bloomBin包括了所有日志的bloom9数据。 3.2，调用BytesToBloom(bloomBin.Bytes())方法，把bloomBin加入区块的bloom过滤器中，这时Bloom过滤器就有了本次交易的所有收据。 3.3，需要说明的是Bloom过滤器只是提供一个查找数据是否存在的工具，它本身不包含任何数据。
4, BloomLookup()方法查找对应的数据是否在bloom过滤器里面。
func BloomLookup(bin Bloom, topic bytesBacked) bool { bloom := bin.Big() cmp := bloom9(topic.Bytes()[:]) return bloom.And(bloom, cmp).Cmp(cmp) == 0 }先将传入的数据转成bloom9值，传入的bloomBin 转成bigInt。根据按位与操作，判断传入的值是否在Bloom过滤器里面。
二，Bloom过滤器的实际应用 bloom过滤器是用来快速的查找log的，那以太坊是如何用bloom过滤器来查找的呢？ 想要要找某一条log，如果从区块链的头区块开始，根据区块头的hash依次开始查找的话是效率比较低的，每个区块写在本地数据库是散列存储的， 会增加很多io请求，io请求的速度很慢的。如何能快速的找到目的区块，这时候就要用到Chain_Indexer。以太坊的BloomIndexer具体实现了Chain_Indexer，可以认为是Chain_Indexer的派生类。 Chain_Indexer的初始化：
func NewChainIndexer(chainDb, indexDb ethdb.Database, backend ChainIndexerBackend, section, confirm uint64, throttling time.Duration, kind string) *ChainIndexer { c := &ChainIndexer{ chainDb: chainDb, indexDb: indexDb, backend: backend, update: make(chan struct{}, 1), quit: make(chan chan error), sectionSize: section, confirmsReq: confirm, throttling: throttling, log: log.New("type", kind), } // Initialize database dependent fields and start the updater c.loadValidSections() go c.updateLoop() return c }chainDb是整个区块链的Db indexDb是这个BloomIndexer的Db sectionSize等于4096，把每4096个区块划到一个section中 loadValidSections，取得indexDb里面存放的section的数量 c.updateLoop是chainIndexer 更新的主循环，有新的区块，或者有新的没有在indexDb里面存放的section产生都会send到c.updateLoop的goroutine里面去。
func (c *ChainIndexer) updateLoop() { var ( updating bool updated time.Time ) for { select { case errc := <-c.quit: // Chain indexer terminating, report no failure and abort errc <- nil return case <-c.update: // Section headers completed (or rolled back), update the index c.lock.Lock() if c.knownSections > c.storedSections { // Periodically print an upgrade log message to the user if time.Since(updated) > 8*time.Second { if c.knownSections > c.storedSections+1 { updating = true c.log.Info("Upgrading chain index", "percentage", c.storedSections*100/c.knownSections) } updated = time.Now() } // Cache the current section count and head to allow unlocking the mutex section := c.storedSections var oldHead common.Hash if section > 0 { oldHead = c.SectionHead(section - 1) } // Process the newly defined section in the background c.lock.Unlock() newHead, err := c.processSection(section, oldHead) if err != nil { c.log.Error("Section processing failed", "error", err) } c.lock.Lock() // If processing succeeded and no reorgs occcurred, mark the section completed if err == nil && oldHead == c.SectionHead(section-1) { c.setSectionHead(section, newHead) c.setValidSections(section + 1) if c.storedSections == c.knownSections && updating { updating = false c.log.Info("Finished upgrading chain index") } c.cascadedHead = c.storedSections*c.sectionSize - 1 for _, child := range c.children { c.log.Trace("Cascading chain index update", "head", c.cascadedHead) child.newHead(c.cascadedHead, false) } } else { // If processing failed, don't retry until further notification c.log.Debug("Chain index processing failed", "section", section, "err", err) c.knownSections = c.storedSections } } // If there are still further sections to process, reschedule if c.knownSections > c.storedSections { time.AfterFunc(c.throttling, func() { select { case c.update <- struct{}{}: default: } }) } c.lock.Unlock() } } }1，c.updateLoop收到update的通知后，看是否有已知的未写入indexDb的section。 2，调用c.processSection(section, oldHead)生成新的section
func (c *ChainIndexer) processSection(section uint64, lastHead common.Hash) (common.Hash, error) { c.log.Trace("Processing new chain section", "section", section) // Reset and partial processing if err := c.backend.Reset(section, lastHead); err != nil { c.setValidSections(0) return common.Hash{}, err } for number := section * c.sectionSize; number < (section+1)*c.sectionSize; number++ { hash := GetCanonicalHash(c.chainDb, number) if hash == (common.Hash{}) { return common.Hash{}, fmt.Errorf("canonical block #%d unknown", number) } header := GetHeader(c.chainDb, hash, number) if header == nil { return common.Hash{}, fmt.Errorf("block #%d [%x…] not found", number, hash[:4]) } else if header.ParentHash != lastHead { return common.Hash{}, fmt.Errorf("chain reorged during section processing") } c.backend.Process(header) lastHead = header.Hash() } if err := c.backend.Commit(); err != nil { c.log.Error("Section commit failed", "error", err) return common.Hash{}, err } return lastHead, nil }2.1，调用c.backend.Reset(section, lastHead)产生一个待组装的section，每个section中存在一个bloom过滤器。 2.2，把number等于section * c.sectionSize到(section+1)*c.sectionSize的block依次加入到待组装的section中。并把这些block的header.bloom加入到section的bloom过滤器中。 2.3，调用c.backend.Commit()，把新的section写入db。返回最近的那block的header。
3，更新sectionHead和ValidSctions，如果还有新的没有在db里面的section的话，在throttling时间后在循环更新一次。
三，外部调用接口查找log的流程 PublicFilterAPI提供了给外部rpc调用的过滤查找接口。比如GetLogs（）方法
func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([]*types.Log, error) { // Convert the RPC block numbers into internal representations if crit.FromBlock == nil { crit.FromBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) } if crit.ToBlock == nil { crit.ToBlock = big.NewInt(rpc.LatestBlockNumber.Int64()) } // Create and run the filter to get all the logs filter := New(api.backend, crit.FromBlock.Int64(), crit.ToBlock.Int64(), crit.Addresses, crit.Topics) logs, err := filter.Logs(ctx) if err != nil { return nil, err } return returnLogs(logs), err }1，FilterCriteria是外部请求的过滤条件，可以根据起始区块，日志的合约地址，日志topics的hash值来设置过滤条件。 2，以太坊内部根据FilterCriteria，创建一个过滤器，把合约地址和topics的hash作为bloombit的匹配器的匹配条件。 3，调用filter.Logs(ctx)来获取日志
func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) { // Figure out the limits of the filter range header, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber) if header == nil { return nil, nil } head := header.Number.Uint64() if f.begin == -1 { f.begin = int64(head) } end := uint64(f.end) if f.end == -1 { end = head } // Gather all indexed logs, and finish with non indexed ones var ( logs []*types.Log err error ) size, sections := f.backend.BloomStatus() if indexed := sections * size; indexed > uint64(f.begin) { if indexed > end { logs, err = f.indexedLogs(ctx, end) } else { logs, err = f.indexedLogs(ctx, indexed-1) } if err != nil { return logs, err } } rest, err := f.unindexedLogs(ctx, end) logs = append(logs, rest...) return logs, err }3.1，如果没有设定起始位置，就认为从最新区块的header开始。找到开始位置区块对应的的section，如果开始位置在section里面就走f.indexedLogs()在chainIndexer里面找log，如果不是就调用f.unindexedLogs()不再chainIndexer里面找。 3.2，f.unindexedLogs()相对简单。
func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) { var logs []*types.Log for ; f.begin <= int64(end); f.begin++ { header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin)) if header == nil || err != nil { return logs, err } if bloomFilter(header.Bloom, f.addresses, f.topics) { found, err := f.checkMatches(ctx, header) if err != nil { return logs, err } logs = append(logs, found...) } } return logs, nil }3.2.1，因为没有并入section的区块都是比较新的区块，数量也不多。直接从最新的区块开始遍历查找就可以了。 3.2.2，bloomFilter(header.Bloom, f.addresses, f.topics)方法，根据合约地址和topics的bloom9值在header的bloom过滤器中按位与操作，看是否在这个区块中。 3.2.3，如果找到这个block，调用checkMatches方法在block里面查找对应的log
func (f *Filter) checkMatches(ctx context.Context, header *types.Header) (logs []*types.Log, err error) { // Get the logs of the block logsList, err := f.backend.GetLogs(ctx, header.Hash()) if err != nil { return nil, err } var unfiltered []*types.Log for _, logs := range logsList { unfiltered = append(unfiltered, logs...) } logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics) if len(logs) > 0 { // We have matching logs, check if we need to resolve full logs via the light client if logs[0].TxHash == (common.Hash{}) { receipts, err := f.backend.GetReceipts(ctx, header.Hash()) if err != nil { return nil, err } unfiltered = unfiltered[:0] for _, receipt := range receipts { unfiltered = append(unfiltered, receipt.Logs...) } logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics) } return logs, nil } return nil, nil }3.2.3.1 调用ethApi的f.backend.GetLogs(ctx, header.Hash())方法，找到这个区块的所有收据下的所有日志。 3.2.3.2 调用filterLogs(unfiltered, nil, nil, f.addresses, f.topics)，根据f.addresses, f.topics过滤出想要的logs。如果第一个log的hash是空的，需要通过light client重现获取一遍所有的日志，再走一下过滤。
3.3，f.indexedLogs() 在chainIndexer里面查找日志
func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) { // Create a matcher session and request servicing from the backend matches := make(chan uint64, 64) session, err := f.matcher.Start(ctx, uint64(f.begin), end, matches) if err != nil { return nil, err } defer session.Close() f.backend.ServiceFilter(ctx, session) // Iterate over the matches until exhausted or context closed var logs []*types.Log for { select { case number, ok := <-matches: // Abort if all matches have been fulfilled if !ok { err := session.Error() if err == nil { f.begin = int64(end) + 1 } return logs, err } f.begin = int64(number) + 1 // Retrieve the suggested block and pull any truly matching logs header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number)) if header == nil || err != nil { return logs, err } found, err := f.checkMatches(ctx, header) if err != nil { return logs, err } logs = append(logs, found...) case <-ctx.Done(): return logs, ctx.Err() } } }indexedLogs启动一个匹配器来查找Filter条件下对应的区块，这一节暂不分析f.matcher的工作原理。 找到对应区块，接下来的事情就和unindexedLogs的处理一样了。
总结： 以太坊的bloom过滤器大大的提高了查询的效率。以太坊先创建topics的bloom，再创建logs的bloom，再创建收据的bloom，在创建header的bloom，最后创建block的bloom，一步一步构建上去。于此对应的，在查找日志的过程正好相反，先在block的bloom里面找，再在header的bloom里面找，再在收据的bloom里面找，直到找到最终的日志。
