GoでのRedisプロトコルの読み取りと書き込み

この投稿では、Redisprotocolがどのように機能し、何が優れているかを理解する方法として、GoでのRedisクライアントの2つのコンポーネントのシンプルでわかりやすい実装の概要を説明します。

Goでフル機能の本番環境に対応したRedisクライアントをお探しの場合は、GaryBurdのredigoライブラリをご覧になることをお勧めします。

始める前に 、Redisプロトコルの穏やかな紹介を必ずお読みください。このガイドで理解する必要のあるプロトコルの基本について説明しています。

GoのRESPコマンドライター

架空のRedisクライアントの場合、書き込む必要のあるオブジェクトは1種類だけです。それは、Redisにコマンドを送信するためのバルク文字列の配列です。コマンドからRESPへのライターの簡単な実装は次のとおりです。

package redis

import (
  "bufio"
  "io"
  "strconv"     // for converting integers to strings
)

var (
  arrayPrefixSlice      = []byte{'*'}
  bulkStringPrefixSlice = []byte{'$'}
  lineEndingSlice       = []byte{'\r', '\n'}
)

type RESPWriter struct {
  *bufio.Writer
}

func NewRESPWriter(writer io.Writer) *RESPWriter {
  return &RESPWriter{
    Writer: bufio.NewWriter(writer),
  }
}

func (w *RESPWriter) WriteCommand(args ...string) (err error) {
  // Write the array prefix and the number of arguments in the array.
  w.Write(arrayPrefixSlice)
  w.WriteString(strconv.Itoa(len(args)))
  w.Write(lineEndingSlice)

  // Write a bulk string for each argument.
  for _, arg := range args {
    w.Write(bulkStringPrefixSlice)
    w.WriteString(strconv.Itoa(len(arg)))
    w.Write(lineEndingSlice)
    w.WriteString(arg)
    w.Write(lineEndingSlice)
  }

  return w.Flush()
}

net.Connに書き込むのではなく オブジェクト、RESPWriter io.Writerに書き込みます 物体。これにより、netに緊密に結合することなくパーサーをテストできます。 スタック。他のioと同じように、ネットワークプロトコルをテストするだけです。 。

たとえば、bytes.Bufferを渡すことができます 最終的なRESPを検査するには：

var buf bytes.Buffer
writer := NewRESPWriter(&buf)
writer.WriteCommand("GET", "foo")
buf.Bytes() // *2\r\n$3\r\nGET\r\n$3\r\nfoo\r\n

GoのシンプルなRESPリーダー

RESPWriterを使用してRedisにコマンドを送信した後 、クライアントはRESPReaderを使用します 完全なRESPreplyを受信するまでTCP接続から読み取ります。まず、受信データのバッファリングと解析を処理するためのいくつかのパッケージが必要です。

package redis

import (
  "bufio"
  "bytes"
  "errors"
  "io"
  "strconv"
)

また、いくつかの変数と定数を使用して、コードを少し読みやすくします。

const (
  SIMPLE_STRING = '+'
  BULK_STRING   = '$'
  INTEGER       = ':'
  ARRAY         = '*'
  ERROR         = '-'
)

var (
  ErrInvalidSyntax = errors.New("resp: invalid syntax")
)

RESPWriterのように 、RESPReader RESPを読み取るオブジェクトの実装の詳細は気にしません。完全なRESPオブジェクトを読み取るまで、バイトを読み取る機能が必要です。この場合、io.Readerが必要です 、bufio.Readerでラップします 着信データのバッファリングを処理します。

オブジェクトと初期化子は単純です：

type RESPReader struct {
  *bufio.Reader
}

func NewReader(reader io.Reader) *RESPReader {
  return &RESPReader{
    Reader: bufio.NewReaderSize(reader, 32*1024),
  }
}

bufio.Readerのバッファサイズ 開発中の単なる推測です。実際のクライアントでは、サイズを構成可能にし、おそらくテストして最適なサイズを見つける必要があります。 32KBは開発に問題なく機能します。

RESPReader メソッドは1つだけです：ReadObject() 、呼び出しごとに完全なRESPオブジェクトを含むバイトスライスを返します。 io.Readerで発生したエラーはすべて返されます 、また、無効なR​​ESP構文が検出された場合にもエラーが返されます。

RESPのプレフィックスの性質は、最初のバイトを読み取るだけで、次のバイトの処理方法を決定できることを意味します。ただし、常に少なくとも最初の完全な行を読み取る必要があるため（つまり、最初の\r\nまで） ）、最初の行全体を読むことから始めることができます：

func (r *RESPReader) ReadObject() ([]byte, error) {
  line, err := r.readLine()
  if err != nil {
    return nil, err
  }

  switch line[0] {
  case SIMPLE_STRING, INTEGER, ERROR:
    return line, nil
  case BULK_STRING:
    return r.readBulkString(line)
  case ARRAY:
    return r.readArray(line) default:
    return nil, ErrInvalidSyntax
  }
}

読み取った行に単純な文字列、整数、またはエラープレフィックスがある場合、これらのオブジェクトタイプは完全に1行に含まれているため、受信したRESPオブジェクトとして完全な行になります。

readLine()で 、\nが最初に出現するまで読み上げます 次に、その前に\rが付いていることを確認します。 行をバイトスライスとして返す前：

func (r *RESPReader) readLine() (line []byte, err error) {
  line, err = r.ReadBytes('\n')
  if err != nil {
    return nil, err
  }

  if len(line) > 1 && line[len(line)-2] == '\r' {
    return line, nil
  } else {
    // Line was too short or \n wasn't preceded by \r.
    return nil, ErrInvalidSyntax
  }
}

readBulkString()で バルク文字列の長さの指定を解析して、読み取る必要のあるバイト数を確認します。完了したら、そのバイト数と\r\nを読み取ります ラインターミネーター：

func (r *RESPReader) readBulkString(line []byte) ([]byte, error) {
  count, err := r.getCount(line)
  if err != nil {
    return nil, err
  }
  if count == -1 {
    return line, nil
  }

  buf := make([]byte, len(line)+count+2)
  copy(buf, line)
  _, err = io.ReadFull(r, buf[len(line):])
  if err != nil {
    return nil, err
  }

  return buf, nil
}

getCount()を取得しました 長さの指定は配列にも使用されるため、別のメソッドに変換します。

func (r *RESPReader) getCount(line []byte) (int, error) {
  end := bytes.IndexByte(line, '\r')
  return strconv.Atoi(string(line[1:end]))
}

配列を処理するには、配列要素の数を取得してから、ReadObject()を呼び出します。 再帰的に、結果のオブジェクトを現在のRESPbufferに追加します：

func (r *RESPReader) readArray(line []byte) ([]byte, error) {
  // Get number of array elements.
  count, err := r.getCount(line)
  if err != nil {
    return nil, err
  }

  // Read `count` number of RESP objects in the array.
  for i := 0; i < count; i++ {
    buf, err := r.ReadObject()
    if err != nil {
      return nil, err
    }
    line = append(line, buf...)
  }

  return line, nil
}