以前，我能夠毫無(wú)問(wèn)題地為我的搜索引擎抓取網(wǎng)頁(yè)并為其建立索引，直到我的數(shù)據(jù)庫(kù)增長(zhǎng)超過(guò) RabbitMQ 消息隊(duì)列的容納能力。如果消息隊(duì)列中的消息超過(guò)其默認(rèn)大小，RabbitMQ 會(huì)拋出錯(cuò)誤并引發(fā)恐慌，我可以更改默認(rèn)大小，但如果我的數(shù)據(jù)庫(kù)增長(zhǎng)，則不會(huì)擴(kuò)展，因此為了讓用戶抓取網(wǎng)頁(yè)而不必?fù)?dān)心消息代理崩潰了。

創(chuàng)建段

我已經(jīng)實(shí)現(xiàn)了一個(gè)函數(shù)，用于創(chuàng)建具有最大段大小或 MSS 的段，其原理與創(chuàng)建段時(shí) TCP 的原理相同，該段包含一個(gè) 8 字節(jié)標(biāo)頭，其中 8 字節(jié)標(biāo)頭中的每個(gè) 4 字節(jié)是序列號(hào)，分段總數(shù)，主體的其余部分是分段數(shù)據(jù)庫(kù)的有效負(fù)載。

// MSS is number in bytes
function createSegments(
  webpages: Array<Webpage>, // webpages queried from database
  MSS: number,
): Array<ArrayBufferLike> {
  const text_encoder = new TextEncoder();
  const encoded_text = text_encoder.encode(JSON.stringify(webpages));
  const data_length = encoded_text.byteLength;
  let currentIndex = 0;
  let segmentCount = Math.trunc(data_length / MSS) + 1; // + 1 to store the remainder
  let segments: Array<ArrayBufferLike> = [];
  let pointerPosition = MSS;

  for (let i = 0; i < segmentCount; i++) {
    let currentDataLength = Math.abs(currentIndex - data_length);

    let slicedArray = encoded_text.slice(currentIndex, pointerPosition);

    currentIndex += slicedArray.byteLength;
    // Add to offset MSS to point to the next segment in the array
    // manipulate pointerPosition to adjust to lower values using Math.min()

    // Is current data length enough to fit MSS?
    // if so add from current position + MSS
    // else get remaining of the currentDataLength
    pointerPosition += Math.min(MSS, currentDataLength);
    const payload = new Uint8Array(slicedArray.length);
    payload.set(slicedArray);
    segments.push(newSegment(i, segmentCount, Buffer.from(payload)));
  }
  return segments;
}

function newSegment(
  sequenceNum: number,
  segmentCount: number,
  payload: Buffer,
): ArrayBufferLike {
  // 4 bytes for sequenceNum 4 bytes for totalSegmentsCount
  const sequenceNumBuffer = convertIntToBuffer(sequenceNum);
  const segmentCountBuffer = convertIntToBuffer(segmentCount);
  const headerBuffer = new ArrayBuffer(8);
  const header = new Uint8Array(headerBuffer);
  header.set(Buffer.concat([sequenceNumBuffer, segmentCountBuffer]));
  return Buffer.concat([header, payload]);
}

function convertIntToBuffer(int: number): Buffer {
  const bytes = Buffer.alloc(4);
  bytes.writeIntLE(int, 0, 4);
  console.log(bytes);
  return bytes;
}

解析傳入的段

這種創(chuàng)建大型數(shù)據(jù)集的小片段的方法將有助于擴(kuò)展數(shù)據(jù)庫(kù)查詢，即使數(shù)據(jù)庫(kù)增長(zhǎng)也是如此。

現(xiàn)在搜索引擎如何解析緩沖區(qū)并將每個(gè)段轉(zhuǎn)換為網(wǎng)頁(yè)數(shù)組？

從段緩沖區(qū)讀取

首先提取段標(biāo)頭，因?yàn)闃?biāo)頭包含 2 個(gè)屬性，即序列號(hào)和總段數(shù)，

func GetSegmentHeader(buf []byte) (*SegmentHeader, error) {
    byteReader := bytes.NewBuffer(buf)
    headerOffsets := []int{0, 4}
    newSegmentHeader := SegmentHeader{}

    for i := range headerOffsets {
        buffer := make([]byte, 4)
        _, err := byteReader.Read(buffer)
        if err != nil {
            return &SegmentHeader{}, err
        }
        value := binary.LittleEndian.Uint32(buffer)

        // this feels disgusting but i dont feel like bothering with this
        if i == 0 {
            newSegmentHeader.SequenceNum = value
            continue
        }
        newSegmentHeader.TotalSegments = value
    }
    return &newSegmentHeader, nil
}

func GetSegmentPayload(buf []byte) ([]byte, error) {
    headerOffset := 8
    byteReader := bytes.NewBuffer(buf[headerOffset:])
    return byteReader.Bytes(), nil

}

處理段的重傳和重新排隊(duì)

序列號(hào)將用于分段的重傳/重新排隊(duì)，因此如果預(yù)期的序列號(hào)不是收到的序列號(hào)，則從當(dāng)前分段開始重新排隊(duì)每個(gè)分段。

    // for retransmission/requeuing
        if segmentHeader.SequenceNum != expectedSequenceNum {
            ch.Nack(data.DeliveryTag, true, true)
            log.Printf("Expected Sequence number %d, got %d\n",
                expectedSequenceNum, segmentHeader.SequenceNum)
            continue
        }

附加段有效負(fù)載

如果搜索引擎接收到的段總數(shù)等于數(shù)據(jù)庫(kù)服務(wù)要發(fā)送的總段長(zhǎng)度，則總段將用于中斷監(jiān)聽生產(chǎn)者（數(shù)據(jù)庫(kù)服務(wù)）然后中斷并解析聚合的段緩沖區(qū)，如果不是，則繼續(xù)偵聽并將段有效負(fù)載緩沖區(qū)附加到網(wǎng)頁(yè)緩沖區(qū)以保存來(lái)自所有傳入段的字節(jié)。

        segmentCounter++
        fmt.Printf("Total Segments : %d\n", segmentHeader.TotalSegments)
        fmt.Printf("current segments : %d\n", segmentCounter)
        expectedSequenceNum++
        ch.Ack(data.DeliveryTag, false)
        webpageBytes = append(webpageBytes, segmentPayload...)
        fmt.Printf("Byte Length: %d\n", len(webpageBytes))

        if segmentCounter == segmentHeader.TotalSegments {
            log.Printf("Got all segments from Database %d", segmentCounter)
            break
        }

順便說(shuō)一句，我使用 vim

感謝您參加我的 ted 演講，我將為 zensearch 實(shí)現(xiàn)更多功能和修復(fù)。

以上是擴(kuò)展 Zensearch 的能力來(lái)查詢整個(gè)數(shù)據(jù)庫(kù)的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！