Быстрее, напишите мне клиент Redis

Красивый протокол упрощает реализацию

Redis - это сетевое хранилище ключей и значений в памяти с дополнительной надежностью, поддерживающее различные виды абстрактных структур данных. Redis можно использовать для реализации различных архитектурных шаблонов на стороне сервера.

Вы взаимодействуете с Redis, используя протокол клиент / сервер. Протокол Redis очень прост, что упрощает его реализацию.

В этой статье показано, как быстро реализовать минимальный, но полностью функциональный клиент Redis в Pharo.

Протокол

RESP (REdis Serialization Protocol) - это протокол запроса / ответа по TCP-соединению. Всего 5 типов данных: простые строки, ошибки, целые числа, массивные строки и массивы. Первый символ указывает на тип.

Простые строки начинаются с + и продолжаются до конца строки (CRLF, \ r \ n). Они не могут содержать конец строк.

+OK\r\n

Ошибки аналогичны, но начинаются с - и заканчиваются CRLF.

-ERR unknown command ‘GETT’\r\n

Целые числа начинаются с : и заканчиваются CRLF

:1\r\n

Групповые строки могут содержать все, что угодно, с префиксом $ и явным счетчиком байтов, заканчивающимся CRLF. Затем следует фактическое содержание. В конце есть дополнительный CRLF.

$2\r\nOK\r\n

Массивы - это составной тип. Они начинаются с *, а количество элементов заканчивается на CRLF. Далее следуют сами элементы.

*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n

Команды (запросы) от клиента к серверу отправляются в виде массива групповых строк. Например, команда для установки ключа в значение называется SET и принимает два аргумента, имя ключа и значение, обе строки. Итак, «SET foo 100» будет выглядеть следующим образом.

*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\n100\r\n

Для удобства также поддерживается более простой формат ввода. Эти встроенные команды можно записать в одну строку с аргументами, разделенными пробелами.

SET foo 100\r\n

Ответы могут быть любого из 5 типов данных и зависят от команды.

Модульные тесты

В духе разработки через тестирование (TDD) мы записываем наши ожидания для двух простых команд. Первая - это простая команда PING, которая дает ответ PONG. Вторая - это команда ECHO, которая отвечает на свой аргумент.

TestCase subclass: #SimpleRedisClientTests
  instanceVariableNames: ‘client’
  classVariableNames: ‘’
  package: ‘SimpleRedisClient’

Для каждого теста нужен функциональный, подключенный клиент.

SimpleRedisClientTests>>#setUp
  client := SimpleRedisClient new.
  client open
SimpleRedisClientTests>>#tearDown
  client close

Вот наш первый тест с использованием упрощенной встроенной команды.

SimpleRedisClientTests>>#testPing
  self assert: (client executeInline: #PING) equals: #PONG

В нашем втором тесте используется общая команда с аргументами.

SimpleRedisClientTests>>#testEcho
  | string |
  string := ‘STR-’ , 99 atRandom asString.
  self assert: (client execute: { #ECHO. string }) equals: string

Клиент

Чтобы подключиться к серверу, нашему клиенту необходимо знать хост и порт. После подключения у нас будет соединение, представляющее двоичный поток TCP. Для плавной работы с символьным вводом и выводом нам потребуются потоки in и out.

Это определение класса для объекта, реализующего наш клиент Redis.

Object subclass: #SimpleRedisClient
  instanceVariableNames: ‘host port connection in out’
  classVariableNames: ‘’
  package: ‘SimpleRedisClient’

Сначала мы добавляем аксессоры для хоста и порта, принимая во внимание некоторые полезные значения по умолчанию.

SimpleRedisClient>>#host
  ^ host ifNil: [ host := ‘localhost’ ]
SimpleRedisClient>>#host: string
  host := string
SimpleRedisClient>>#port
  ^ port ifNil: [ port := 6379 ]
SimpleRedisClient>>#port: integer
  port := integer

Открытие и закрытие клиента означает работу с двоичным потоком сокетов TCP и потоками символов для чтения и записи.

SimpleRedisClient>>#open
  self close.
  connection := ZdcSocketStream 
    openConnectionToHostNamed: self host 
    port: self port.
  in := ZnCharacterReadStream on: connection.
  out := ZnCharacterWriteStream on: connection
SimpleRedisClient>>#close
  connection
    ifNotNil: [ 
      [ connection close ] on: Error do: [ ].
      in := out := connection := nil ]

Вызов #close из #open делает его более надежным и способным функционировать как повторный вызов. Необработанный поток сокета TCP имеет дело с байтами, а не символами. Символьные потоки добавляют кодировку и декодирование UTF-8.

Обычное выполнение команды состоит из двух этапов: записи команды и чтения ответа.

SimpleRedisClient>>#execute: commandArgs
  self writeCommand: commandArgs.
  ^ self readReply
SimpleRedisClient>>#executeInline: command
  self writeInlineCommand: command.
  ^ self readReply

Остается реализовать сам протокол: писать команды и читать ответы в соответствии со спецификацией.

Написание команд

SimpleRedisClient>>#writeCommand: args
  out nextPut: $*; print: args size; crlf.
  args do: [ :each |
    | string byteCount |
    string := each asString.
    byteCount := out encoder encodedByteCountForString: string.
    out 
      nextPut: $$; print: byteCount; crlf; 
      nextPutAll: string; crlf ].
  out flush
SimpleRedisClient>>#writeInlineCommand: string
  out nextPutAll: string; crlf; flush

Обычные команды - это массивы массивных строк. Сначала записываем количество элементов. Каждый аргумент команды преобразуется в строку. Протокол Redis не указывает, какую кодировку использовать для строк, это зависит от клиента. Для Redis это просто набор байтов. Мы решили использовать UTF-8, кодировку переменной длины. Длина строки после $ - это количество байтов, а не количество символов. Поэтому мы используем кодировщик, чтобы вычислить, сколько байтов необходимо для кодирования строки.

Встроенные команды записываются в одну строку.

За командой следует #flush, чтобы передать все данные по сети.

Чтение ответов

Чтение ответа начинается с просмотра первого символа.

SimpleRedisClient>>#readReply
  | first |
  first := in next.
  first = $+ ifTrue: [ ^ in nextLine ].
  first = $: ifTrue: [ ^ in nextLine asInteger ].
  first = $- ifTrue: [ ^ self error: in nextLine ].
  first = $* ifTrue: [ ^ self readArray ].
  first = $$ ifTrue: [ ^ self readBulkString ].
  self error: ‘Unknown reply type’

Простые строки, ошибки и числа обрабатываются напрямую. Массивы и массовые строки обрабатываются вспомогательными методами.

SimpleRedisClient>>#readArray
  | length array |
  length := in nextLine asInteger.
  length = -1 ifTrue: [ ^ nil ].
  array := Array new: length streamContents: [ :elements |
    length timesRepeat: [ elements nextPut: self readReply ] ].
  ^ array

Длина массива, равная -1, является особым случаем и отличается от пустого массива. Правильное количество элементов считывается путем рекурсивного вызова #readReply. Массивы могут содержать элементы разных типов.

SimpleRedisClient>>#readBulkString
  | byteCount bytes |
  byteCount := in nextLine asInteger.
  byteCount = -1 ifTrue: [ ^ nil ].
  bytes := in wrappedStream next: byteCount.
  in nextLine.
  ^ in encoder decodeBytes: bytes

Этот последний метод завершает нашу реализацию. Опять же, длина -1 рассматривается как особый случай. Поскольку длина - это количество байтов, мы обращаемся к #wrappedStream (то есть к потоку двоичного сокета) и читаем необработанные байты в одном блоке. Затем мы используем кодировщик для декодирования байтов.

Примерно через 10 важных методов мы реализовали минимальный, но полностью функциональный клиент Redis.

Больше модульных тестов

Помимо этого обманчиво простого протокола, Redis реализует множество функций, и все они доступны с помощью нашего простого клиента. Вот пара примеров, написанных в виде дополнительных юнит-тестов.

Подробную информацию о каждой команде можно найти в интерактивной документации по командам Redis.

Хранение ключей и значений

Вот некоторые основные способы использования хранилища ключей.

SimpleRedisClientTests>>#testStringGetSetSimple
  | string |
  string := ‘STR-’ , 99 atRandom asString.
  self assert: (client execute: #(DEL foo)) >= 0.
  self assert: (client execute: #(EXISTS foo)) equals: 0.
  self assert: (client execute: #(GET foo)) isNil.
  self assert: (client execute: { #SET. #foo. string }) equals: #OK.
  self assert: (client execute: #(GET foo)) equals: string.
  self assert: (client execute: #(EXISTS foo)) equals: 1.
  self assert: (client execute: #(DEL foo)) > 0.

Счетчики

Значение ключа можно интерпретировать как целочисленный счетчик с атомарными операциями. Неопределенные счетчики начинаются с нуля.

SimpleRedisClientTests>>#testSimpleCounter
  client execute: #(DEL mycounter).
  self assert: (client execute: #(INCR mycounter)) equals: 1.
  self assert: (client execute: #(INCR mycounter)) equals: 2.
  self assert: (client execute: #(GET mycounter)) equals: ‘2’.
  self assert: (client execute: #(DECR mycounter)) equals: 1.
  self assert: (client execute: #(INCRBY mycounter 10)) equals: 11.
  client execute: #(DEL mycounter).

Другие структуры данных

Списки, наборы, отсортированные наборы и хеш-таблицы - это некоторые из поддерживаемых структур данных. Вы можете найти больше модульных тестов в распространяемом исходном коде (см. Инструкции по загрузке в конце).

Очереди

Наш клиент даже достаточно гибок, чтобы поддерживать некоторые особые варианты использования. Списки можно использовать как очереди с блокирующим вызовом для ожидания новых данных.

SimpleRedisClientTests>>#testQueueSimple
  | string semaphore |
  string := ‘STR-’ , 99 atRandom asString.
  semaphore := Semaphore new.
  client execute: #(DEL myqueue).
  [
    | anotherClient |
    anotherClient := SimpleRedisClient new.
    anotherClient open.
    semaphore signal.
    "Block waiting for data entering the queue"
    self
      assert: (anotherClient execute: #(BRPOP myqueue 0)) 
      equals: { #myqueue. string }.
    semaphore signal.
    anotherClient close
  ] 
    forkAt: Processor userSchedulingPriority 
    named: ‘testQueueSimple’.
  semaphore wait.
  self assert: (client execute: { #LPUSH. #myqueue. string }) > 0.
  semaphore wait.
  client execute: #(DEL myqueue).

Основной поток ожидает, пока разветвленный поток будет готов, координируя его с помощью семафора. Затем он помещает новый элемент в начало очереди.

Разветвленный поток выдает блокирующий всплывающий сигнал в конце очереди. Он получит и удалит строку, проталкиваемую основным потоком.

Pub / Sub

Другой вариант использования - это механизм pub / sub. Данные публикуются в канале. На этот канал могут подписаться несколько сторон. Каждый подписчик будет получать уведомления об опубликованных данных.

SimpleRedisClientTests>>#testPubSubSimple
  | semaphore string |
  string := ‘STR-’ , 99 atRandom asString.
  semaphore := Semaphore new.
  client execute: #(DEL mychannel).
 
  [
    | anotherClient |
    semaphore wait.
    anotherClient := SimpleRedisClient new.
    anotherClient open.
    self assert: (anotherClient execute:
      { #PUBLISH. #mychannel. string }) > 0.
    anotherClient close
  ] 
    forkAt: Processor userBackgroundPriority 
    named: ‘testPubSubSimple’.
  self 
    assert: (client execute: #(SUBSCRIBE mychannel))
    equals: #(subscribe mychannel 1).
  semaphore signal.
  "Block waiting for data distributed over the channel" 
  self 
    assert: client readReply 
    equals: { #message. #mychannel. string }.
  self 
    assert: (client execute: #(UNSUBSCRIBE mychannel)) 
    equals: #(unsubscribe mychannel 0).
  client execute: #(DEL mychannel).

Здесь основной поток блокируется при чтении входящего уведомления после подписки на канал.

Разветвленный поток публикует строку на канале. Основной поток получит эту строку как сообщение от канала.

Заключение

Разработчики Redis сделали умный ход - предложить такой элегантный и простой протокол. Это упрощает внедрение новых клиентов, что приводит к увеличению числа пользователей их программного обеспечения.

Для динамического интерактивного языка, такого как Pharo, важно иметь возможность взаимодействовать с как можно большим количеством сервисов и серверов. К счастью, написание новых клиентов может быть не только забавным, но и легким занятием.

Исходный код

Исходный код для SimpleRedisClient и SimpleRedisClientTests можно найти на GitHub по адресу https://github.com/svenvc/SimpleRedisClient.

Чтобы загрузить код в Pharo 6.1 или более поздней версии, откройте Мир ›Инструменты› Iceberg, нажмите +, чтобы добавить репозиторий, выберите параметр Клонировать с github.com и введите svenvc в качестве имени владельца и SimpleRedisClient в качестве имени проекта. Выбрав новый репозиторий, выберите Metacello ›Установить базовый план SimpleRedisClient.