Это 6-й день моего участия в Gengwen Challenge, смотрите подробности мероприятия:Обновить вызов

Проект с открытым исходным кодом, представленный вам сегодня, — это Godis: сервер Redis, реализованный на языке Go. служба поддержки:

5 структур данных (строка, список, хэш, набор, отсортированный набор)
Автоматический срок действия (TTL)
Публикация и подписка, геолокация, постоянство и другие функции

Возможно, вам не нужно внедрять сервис Redis самостоятельно, но вы устали каждый день писать, добавлять, удалять, изменять и проверять бизнес-код, пытаясь повысить свой уровень программирования и пытаясь написать проект с нуля, чтобы открыть IDE, но не получается начать?

Создание колеса должно быть хорошим способом улучшить свои навыки программирования.Давайте напишем сервер Redis (Godis) с нуля на Go, из которого вы узнаете:

Как написать Go TCP-сервер
Разработайте и внедрите безопасный и надежный протокол связи (протокол Redis).
Как использовать язык Go для разработки программ с высокой степенью параллелизма
Проектировать и внедрять распределенные кластеры и распределенные транзакции
Знакомы с общими структурами данных, такими как связанный список, хэш-таблица, таблица пропуска и колесо времени.

Не волнуйтесь, если содержание слишком сложное, вы не можете выучить или не имеете языковой базы Go! ! Хотя пример кода написан на Go, он не повлияет на ваше понимание принципов и базовых протоколов Redis и секретов высокой производительности. Кроме того, чтобы позаботиться о читателях, автор оптимизировал техническое объяснение. Пример кода упрощен на основе оригинального проекта, а комментарии добавлены построчно. Если вы продвинутый игрок, посетите проект напрямую, чтобы прочитать исходный код:

GitHub.com/HDT3213/Боже…

Начинается следующий текст, давайте вместе рассеем туман Redis.

1. Напишите TCP-сервер

Как мы все знаем, Redis представляет собой модель C/S и использует для связи протокол TCP. Следующим шагом является реализация TCP-сервера. Как язык программирования, широко используемый на стороне сервера, Golang предоставляет очень лаконичный TCP-интерфейс, поэтому его очень удобно реализовывать. Образец кода:

func ListenAndServe(address string) {
    // 绑定监听地址
    listener, err := net.Listen("tcp", address)
    if err != nil {
        log.Fatal(fmt.Sprintf("listen err: %v", err))
    }
    defer listener.Close()
    log.Println(fmt.Sprintf("bind: %s, start listening...", address))

    for {
        // Accept 会一直阻塞直到有新的连接建立或者listen中断才会返回
        conn, err := listener.Accept()
        if err != nil {
            // 通常是由于listener被关闭无法继续监听导致的错误
            log.Fatal(fmt.Sprintf("accept err: %v", err))
        }
        // 开启新的 goroutine 处理该连接
        go Handle(conn)
    }
}

func Handle(conn net.Conn) {
    reader := bufio.NewReader(conn)
    for {
        // ReadString 会一直阻塞直到遇到分隔符 '\n'
        // 遇到分隔符后 ReadString 会返回上次遇到分隔符到现在收到的所有数据
        // 若在遇到分隔符之前发生异常, ReadString 会返回已收到的数据和错误信息
        msg, err := reader.ReadString('\n')
        if err != nil {
            // 通常遇到的错误是连接中断或被关闭，用io.EOF表示
            if err == io.EOF {
                log.Println("connection close")
            } else {
                log.Println(err)
            }
            return
        }
        b := []byte(msg)
        // 将收到的信息发送给客户端
        conn.Write(b)
    }
}

func main() {
    ListenAndServe(":8000")
}

👌 Пока для завершения сервера требуется всего 40 строк кода! После запуска вышеуказанной службы TCP введите в терминалеtelnet 127.0.0.1 8000Вы можете подключиться к серверу, который вы только что написали, и он вернет сообщение, которое вы отправили обратно, как есть (поэтому, пожалуйста, не ругайте его):

Этот TCP-сервер очень прост: основная сопрограмма вызывает функцию accept для прослушивания порта, а после принятия нового соединения открывается горутина для его обработки. Эта простая модель блокирующего ввода-вывода чем-то похожа на более ранние серверы Tomcat/Apache.

Модель блокирующего ввода-вывода заключается в использованииОдин поток обрабатывает одно соединение, прослушивающий поток блокируется, когда новые данные не получены, пока поток не будет активирован для обработки, когда данные будут готовы. Модель блокирующего ввода-вывода неэффективна, поскольку требует запуска большого количества потоков и частых переключений контекста. Технология epoll (мультиплексирование ввода-вывода), используемая Redis, используетОдин поток обрабатывает много соединений, что значительно повышает пропускную способность. Так будет ли наш TCP-сервер намного медленнее, чем Redis?

Конечно, нет, Golang использует то преимущество, что накладные расходы на планирование Goroutine намного меньше, чем накладные расходы на планирование потоков для инкапсуляцииgoroutine-per-connectionСтилизуйте минимальный интерфейс, а библиотека net/tcp инкапсулирует epoll как блокирующий ввод-вывод, что позволяет избежать сложного асинхронного кода, необходимого для собственного интерфейса epoll, при этом обеспечивая высокую производительность epoll.

На компьютере автора Redis может отвечать на 10,6 тыс. команд PING в секунду, в то время как Godis (полный код) имеет пропускную способность 9,2 тыс. запросов в секунду, что не сильно отличается. Чтобы узнать больше о секретах высокой производительности Golang, вы можете выполнить поискgo netpollerилиgo 语言网络轮询器ключевые слова

Кроме того, квалифицированный TCP-сервер не должен останавливаться при закрытии, но ему необходимо выполнить необходимую работу по очистке, такую как ответ на полученные запросы и освобождение TCP-соединений. Эту функцию обычно называют优雅关闭илиgraceful shutdown, шаги изящного завершения работы:

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

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

2. Перспективный протокол Redis

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

Начиная с Redis 2.0, связь объединена в протокол RESP (протокол сериализации REdis), который легко реализовать и который может не только эффективно анализироваться программами, но также может считываться и отлаживаться людьми.

RESP — это двоичный защищенный текстовый протокол, работающий по протоколу TCP. RESP использует линии в качестве единицы измерения, а команды или данные, отправляемые клиентом и сервером, всегда находятся в\r\n(CRLF) в качестве новой строки.

Двоичная безопасность означает, что в протоколе разрешено использовать произвольные символы, не вызывая сбоя. Например, строка в языке C начинается с\0Поскольку конец строки не может появляться в середине\0, а строка Go позволяет отображать\0, мы говорим, что строки в языке Go являются двоично-безопасными, а строки в языке C не являются двоично-безопасными.

Двоичная безопасность RESP позволяет нам включать в ключ или значение\rили\nтакие специальные символы. Двоичная безопасность особенно важна при использовании Redis для хранения двоичных данных, таких как protobuf, msgpack и т. д.

RESP определяет 5 форматов:

Простая строка: используется сервером для возврата простых результатов, таких как «ОК». Небезопасен для двоичного кода и не допускает новой строки
Ошибка: используется сервером для возврата простых сообщений об ошибках, таких как «ERR Invalid Synatx», который не является двоично-безопасным и не допускает новой строки.
Целое: возвращаемое значение таких команд, как llen и scar, 64-битное целое число со знаком.
Строка (массовая строка): безопасная двоичная строка, такая как возвращаемое значение таких команд, как get
Массив (Array, также известный как Multi Bulk Strings): массовый массив строк, формат команды, отправляемой клиентом, и ответ команды, такой как lrange.

RESP представляет формат первым символом:

Простые строки: начните с "+", например: "+ ok \ r \ n"
Ошибка: начинается с "-", например: "-ERR Invalid Synatx\r\n"
Целое число: начните с ":", например: ":1\r\n"
строка: с$Начинать
массив: с*Начинать

Давайте разберемся с протоколом на некоторых практических примерах.

2.1 Струны

String (Bulk String) состоит из двух строк, первая строка$+ Длина тела, вторая строка - собственно содержание. как:

$3\r\nSET\r\n

Строки (Bulk Strings) являются двоично-безопасными, что означает, что они могут содержать символы "\r\n" внутри Bulk String (CRLF в конце строки скрыт):

$4
a\r\nb

2.2 Пусто

$-1Указывает nil, например, при использовании команды get для запроса несуществующего ключа ответ$-1.

2.3 Массивы

Первая строка формата Array — это «*» + длина массива, за которой следует соответствующее количество Bulk Strings. Например["foo", "bar"]Сообщение (содержание на момент передачи):

*2
$3
foo
$3
bar

Клиент также отправляет команды на сервер, используя формат Array. Сама команда будет первым аргументом, напримерSET key valueСообщение RESP команды:

*3
$3
SET
$3
key
$5
value

Печать Newlines Out:

*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n

2.4 Подготовка к анализу

Как только вы узнаете содержимое часто используемых пакетов RESP, вы можете начать их разбор. Но следует отметить, что RESP二进制安全протокол, позволяющий использовать в организме\r\nперсонаж. Например, Redis может правильно получать и выполнятьSET "a\r\nb" hellogithubинструкции, правильное сообщение этой инструкции выглядит следующим образом:

*3  
$3
SET
$4
a\r\nb 
$11
hellogithub

когдаReadBytesПятая строка "a\r\nb\r\n" читается как две строки:

*3  
$3
SET
$4
a  // 错误的分行
b // 错误的分行
$11
hellogithub

Итак, когда четвертая строка прочитана$4, не следует продолжать использоватьReadBytes('\n')Чтобы прочитать следующую строку, вы должны использоватьio.ReadFull(reader, msg)метод для чтения содержимого указанной длины.

msg = make([]byte, 4 + 2) // 正文长度4 + 换行符长度2
_, err = io.ReadFull(reader, msg)

2.5 Написание парсера протокола RESP

После решения проблемы с тем, что вышеприведенный контент содержит «\r\n», мы можем приступить к написанию парсера протокола Redis!

type Payload struct {
	Data redis.Reply
	Err  error
}

// ParseStream 通过 io.Reader 读取数据并将结果通过 channel 将结果返回给调用者
// 流式处理的接口适合供客户端/服务端使用
func ParseStream(reader io.Reader) <-chan *Payload {
	ch := make(chan *Payload)
	go parse0(reader, ch)
	return ch
}

Поскольку код синтаксического анализатора относительно велик, здесь кратко представлен только основной процесс.

func parse0(reader io.Reader, ch chan<- *Payload) {
    // 初始化读取状态
    readingMultiLine := false
    expectedArgsCount := 0
    var args [][]byte
    var bulkLen int64
    for {
        // 上文中我们提到 RESP 是以行为单位的
        // 因为行分为简单字符串和二进制安全的 BulkString，我们需要封装一个 readLine 函数来兼容
        line, err = readLine(reader, bulkLen)
        if err != nil { 
            // 处理错误
            return
        }
        // 接下来我们对刚刚读取的行进行解析
        // 我们简单的将 Reply 分为两类:
        // 单行: StatusReply, IntReply, ErrorReply
        // 多行: BulkReply, MultiBulkReply

        if !readingMultiLine {
            if isMulitBulkHeader(line) {
                // 我们收到了 MulitBulkReply 的第一行
                // 获得 MulitBulkReply 中 BulkString 的个数
                expectedArgsCount = parseMulitBulkHeader(line)
                // 等待 MulitBulkReply 后续行
                readingMultiLine = true
            } else if isBulkHeader(line) {
                // 我们收到了 BulkReply 的第一行
                // 获得 BulkReply 第二行的长度, 通过 bulkLen 告诉 readLine 函数下一行 BulkString 的长度
                bulkLen = parseBulkHeader()
                // 这个 Reply 中一共有 1 个 BulkString
                expectedArgsCount = 1 
                // 等待 BulkReply 后续行
                readingMultiLine = true
            } else {
                // 处理 StatusReply, IntReply, ErrorReply 等单行 Reply
                reply := parseSingleLineReply(line)
                // 通过 ch 返回结果
                emitReply(ch)
            }
        } else {
            // 进入此分支说明我们正在等待 MulitBulkReply 或 BulkReply 的后续行
            // MulitBulkReply 的后续行有两种，BulkHeader 或者 BulkString
            if isBulkHeader(line) {
                bulkLen = parseBulkHeader()
            } else {
                // 我们正在读取一个 BulkString, 它可能是 MulitBulkReply 或 BulkReply 
                args = append(args, line)
            }
            if len(args) == expectedArgsCount { // 我们已经读取了所有后续行
                // 通过 ch 返回结果
                emitReply(ch)
                // 重置状态, 准备解析下一条 Reply
                readingMultiLine = false
                expectedArgsCount = 0
                args = nil
                bulkLen = 0
            }
        }
    }
}

3. Внедрите базу данных в памяти

Пока мы сделали часть приема и разбора данных, остальное — где хранить данные?

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

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

sync.map: Параллельная хеш-таблица, официально предоставляемая Golang, подходит для сценариев с большим количеством операций чтения и меньшим количеством операций записи. Но когдаm.dirtyСразу после повышения будетm.readкопировать в новыйm.dirty, в случае большого объёма данных операция копирования заблокирует все сопрограммы, и тут таится большая скрытая опасность.
juc.ConcurrentHashMap: параллельная хеш-таблица Java реализована с использованием сегментированных блокировок. Доступ к потоку хеш-таблицы во время расширения поможет в операции повторного хеширования, а все операции чтения и записи будут заблокированы до конца повторного хеширования. Поскольку количество пар ключ-значение в базе данных кеша огромно, а время отклика операций чтения и записи велико, стратегия использования juc не подходит.
memcached hashtable: когда фоновый поток выполняет операцию повторного хеширования, основной поток определяет, был ли повторно хеширован слот хэша, к которому нужно получить доступ, чтобы решить, следует ли использовать old_hashtable или new_hashtable. Эта конструкция называетсяпрогрессивная перефразировкаЕго преимущество в том, что операция перехеширования в основном не блокирует чтение и запись основного потока, что является наиболее идеальным решением.

Однако реализация прогрессивного повторного хеширования очень сложна, поэтому godis использует стратегию блокировки сегментации, широко используемую сообществом Golang (не три вышеперечисленных), которая заключается в распределении ключей по фиксированному количеству сегментов, чтобы избежать общей операции повторного хеширования. Шард — это карта, защищенная блокировкой. Когда сегмент выполняет перехеширование, он блокирует чтение и запись в сегменте, но не влияет на другие сегменты.

код показывает, как показано ниже:

type ConcurrentDict struct {
    table []*Shard
    count int32
}

type Shard struct {
    m     map[string]interface{}
    mutex sync.RWMutex
}

func (dict *ConcurrentDict) spread(hashCode uint32) uint32 {
	tableSize := uint32(len(dict.table))
	return (tableSize - 1) & uint32(hashCode)
}

func (dict *ConcurrentDict) getShard(index uint32) *Shard {
	return dict.table[index]
}

func (dict *ConcurrentDict) Get(key string) (val interface{}, exists bool) {
	hashCode := fnv32(key)
	index := dict.spread(hashCode)
	shard := dict.getShard(index)
	shard.mutex.RLock()
	defer shard.mutex.RUnlock()
	val, exists = shard.m[key]
	return
}

func (dict *ConcurrentDict) Put(key string, val interface{}) (result int) {
	if dict == nil {
		panic("dict is nil")
	}
	hashCode := fnv32(key)
	index := dict.spread(hashCode)
	shard := dict.getShard(index)
	shard.mutex.Lock()
	defer shard.mutex.Unlock()

	if _, ok := shard.m[key]; ok {
		shard.m[key] = val
		return 0
	} else {
		shard.m[key] = val
		dict.addCount()
		return 1
	}
}

ConcurrentDictБезопасность параллелизма для одной ключевой операции может быть гарантирована, но она все еще не может соответствовать требованиям безопасности параллелизма, например:

Необходимо выполнить команду Incr:读取 -> 做加法 -> 写入Трехэтапная операция, двухэтапные операции чтения и записи не являются атомарными
Команда MSETNX устанавливает значение всех заданных ключей тогда и только тогда, когда все заданные ключи не существуют, нам необходимо обеспечить атомарность двух операций «проверить, существует ли несколько ключей» и «записать несколько ключей».

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

выполнитьdb.LockerСамая прямая идея состоит в том, чтобы использовать одинmap[string]*sync.RWMutex

Процесс блокировки делится на два этапа: инициализация мьютекса -> блокировка
Процесс разблокировки также делится на два этапа: разблокировка -> освобождение мьютекса.

Тогда возникает неразрешимая проблема параллелизма:

время	Корутина А	Корутина Б
1		locker["a"].Unlock()
2	locker["a"] = &sync.RWMutex{}
3		delete(locker["a"])
4	locker["a"].Lock()

Поскольку сопрограмма B снимает блокировку в момент t3, попытка сопрограммы A заблокировать ее в момент t4 потерпит неудачу. Если сопрограмма B не выполняется при разблокировкеdelete(locker["a"])Этого исключения можно избежать, но это вызовет серьезные утечки памяти.

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

type Locks struct {
    table []*sync.RWMutex
}

func Make(tableSize int) *Locks {
    table := make([]*sync.RWMutex, tableSize)
    for i := 0; i < tableSize; i++ {
        table[i] = &sync.RWMutex{}
    }
    return &Locks{
        table: table,
    }
}

func (locks *Locks)Lock(key string) {
    index := locks.spread(fnv32(key))
    mu := locks.table[index]
    mu.Lock()
}

func (locks *Locks)UnLock(key string) {
    index := locks.spread(fnv32(key))
    mu := locks.table[index]
    mu.Unlock()
}

При блокировке нескольких ключей следует отметить, что если сопрограмма A удерживает блокировку ключа a и пытается получить блокировку ключа b, в это время сопрограмма B удерживает блокировку ключа b и пытается получить блокировку ключа a. , что приведет к тупику.

Решение состоит в том, что все сопрограммы блокируются в одном и том же порядке.Если две сопрограммы хотят получить блокировку ключа a и ключа b, они должны сначала получить блокировку ключа a, а затем блокировку ключа b, чтобы избежать циклического ожидание .

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

Наконец, приведенный выше код был упрощенным самописным проектом с открытым исходным кодом Godis: сервер Redis с языком Go. Ждем вашего внимания и звезды:

адрес проекта:GitHub.com/HDT3213/Боже…

Эпилог

Повседневная работа многих друзей состоит в основном в написании бизнес-кода, и они опасаются «архитектуры» и «низкоуровневого кода» фреймворков, баз данных и промежуточного ПО.

Но в этой статье мы написали только 3 компонента и в общей сложности сотни строк кода для реализации базового сервера Redis. Таким образом, базовая технология не сложна, пока вы интересуетесь технологией от более мелкой к более глубокой, от простой к сложной, «основной код» не является загадочным.

Интерес — лучший учитель, HelloGitHub открывает для себя радость программирования

обрати внимание наHelloGitHubОфициальный аккаунт получил обновление в кратчайшие сроки.

Есть больше представлений о проектах с открытым исходным кодом и ценных проектов, которые ждут вас.