• Классические реализации
• Two Lock Queue
• Отказываемся от Mutex совсем
• Где можно встретить подобный код?

В прошлом посте мы разобрались что такое lock-free алгоритм на примере задачи TransferMoney. В заметке я обмолвился что идеи из предложенного алгоритма имеют место быть в реальном мире.

Сегодня хочу это продемонстрировать на примере реализации структуры данных Queue. Пройдем путь от наивной реализации до очереди Майкла-Скотта.

Классические реализации

Очередь — это линейная структура данных, в которой элементы обрабатываются по принципу FIFO (First In, First Out — «первым пришел — первым вышел»)

Классическая реализация на языке Go:

type Node struct {
  val  int
  next *Node
}

type Queue struct {
  head *Node
  tail *Node
}

func NewQueue() *Queue {
  sentinel := &Node{}
  return &Queue{head: sentinel, tail: sentinel}
}

func (q *Queue) Push(val int) {
  node := &Node{val: val}
  q.tail.next = node
  q.tail = node
}

func (q *Queue) Pop() *int {
  next := q.head.next
  if next == nil {
    return nil
  }
  q.head = next
  val := next.val
  return &val
}

Если нам нужно работать из нескольких потоков:

type Node struct {
  val  int
  next *Node
}

type MutexQueue struct {
  mu   sync.Mutex
  head *Node
  tail *Node
}

func NewMutexQueue() *MutexQueue {
  sentinel := &Node{}
  return &MutexQueue{head: sentinel, tail: sentinel}
}

func (q *MutexQueue) Push(val int) {
  q.mu.Lock()
  defer q.mu.Unlock()

  node := &Node{val: val}
  q.tail.next = node
  q.tail = node
}

func (q *MutexQueue) Pop() *int {
  q.mu.Lock()
  defer q.mu.Unlock()

  next := q.head.next
  if next == nil {
    return nil
  }
  q.head = next
  val := next.val
  return &val
}

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

Two Lock Queue

В статье Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms от Maged M. Michael и Michael L. Scott был предложен вариант реализации очереди не на одном мьютексе а на двух - для головы и хвоста соответственно.

type Node struct {
  val  int
  next atomic.Pointer[Node]
}

type TwoLockQueue struct {
  head     *Node
  headLock sync.Mutex

  tail     *Node
  tailLock sync.Mutex
}

func NewTwoLockQueue() *TwoLockQueue {
  sentinel := &Node{}
  return &TwoLockQueue{head: sentinel, tail: sentinel}
}

func (q *TwoLockQueue) Push(val int) {
  node := &Node{val: val}

  q.tailLock.Lock()
  defer q.tailLock.Unlock()

  q.tail.next.Store(node)
  q.tail = node
}

func (q *TwoLockQueue) Pop() *int {
  q.headLock.Lock()
  defer q.headLock.Unlock()

  next := q.head.next.Load()
  if next == nil {
    return nil
  }
  q.head = next
  val := next.val
  return &val
}

Код совсем не отличается от того что мы рассмотрели ранее. Кроме одной вещи - в структуре Node атрибут next теперь не просто Node, а atomic.Pointer. Нужно это для того чтобы получить синхронизацию когда в очереди 0 и 1 элемент. Без атомика одновременные push и pop будут бить в одну и ту же область памяти.

Такую реализацию очереди можно встретить в Java - класс LinkedBlockingQueue

Итог - мы добились того что чтения и записи у нас не зависят друг от друга.

Отказываемся от Mutex совсем

А что насчет Lock-free реализации? Она существует и ее предложили в той же статье что я упомянул выше.

type AtomicNode struct {
  val  int
  next atomic.Pointer[AtomicNode]
}

type LockFreeQueue struct {
  head atomic.Pointer[AtomicNode]
  tail atomic.Pointer[AtomicNode]
}

func NewLockFreeQueue() *LockFreeQueue {
  q := &LockFreeQueue{}
  q.head = atomic.Pointer[AtomicNode]{}
  node := &AtomicNode{}

  q.head.Store(node)
  q.tail.Store(node)

  return q
}

func (q *LockFreeQueue) Push(val int) {
  node := AtomicNode{val: val}

  for {
    curr := q.tail.Load()
    currNext := curr.next.Load()

    // поймали незавершенный push, 
    // пытаемся завершить операцию "за того парня"
    if currNext != nil {
      q.tail.CompareAndSwap(curr, currNext)
      continue
    }

    // подвешиваемся на то место где раньше был nil
    if curr.next.CompareAndSwap(nil, &node) {
      q.tail.CompareAndSwap(curr, &node)
      return
    }
  }
}

func (q *LockFreeQueue) Pop() *int {
  for {
    curr := q.head.Load()
    currNext := curr.next.Load()
    if currNext == nil {
      return nil
    }
    val := currNext.val
    if ok := q.head.CompareAndSwap(curr, currNext); ok {
      return &val
    }
  }
}

В чем основные отличия?

• Каждый узел очереди - обязательно спрятан за atomic.
• Операции реализованы через бесконечный цикл (классика lock-free).
• У нас есть dummy узел. Если для классической реализации и версии с мьютексом это скорее украшательство для удобства и простоты кода. То для lock-free очереди это обязательно.
• В функции Push спрятана логика взаимопомощи потоков. Если при вставке мы видим что у tail есть ненулевой tail.next то в первую очередь мы протолкнем его, и уйдем на следующую итерацию цикла.
• В случае когда у нас честный tail и tail.next нам нужно обновить 2 атомика. И как вы помните, тут нужно быть осторожными. Поэтому мы в обязательном порядке линкуем к хвосту наш новый узел и Если операция прошла успешно, то уже по принципу best effort пытаемся подвинуть и сам tail, но уже не проверяем успешность - ведь если что, за нас это сделает другой поток.

Как итог - мы снова добились того что чтения и записи у нас не зависят друг от друга. На этот раз без мьютексов вообще.

P.S. Реализация подобная этой есть в Java - ConcurrentLinkedQueue.

Где можно встретить подобный код?

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

Почему спросите вы? Дело в том что для настоящего хайлоад прода мало взять lock-free, нужно еще и оптимизироваться под железо (которое стало и мощнее и сложнее). А это отдельный вид трюков, для каждого можно писать отдельную заметку. В своем же цикле заметок я стараюсь идти поступательно, от простого (насколько это возможно) к сложному.

В любом случае начало положено, мы уже научились писать код без мьютексов, в следующей заметке добьем последний кусочек теории - wait free алгоритмы и структуры данных.

• Multi-Word Compare-And-Swap
• Show me your code
• Выводы или "Зачем так сложно?"

В прошлых заметках мы прошлись по синтетической задаче TransferMoney вдоль и поперек. Сошлись на том что код с Mutex - самая простая и понятная реализация. Но есть ли у нее альтернативы? Мы попытались написать код на атомиках, он оказался сложнее и имел баги. В этой заметке я попробую написать код без найденных недостатков.

Ахиллесова пята нашего первого "блина комом" - наличие двух атомиков. Достичь атомарности и предсказуемости над двумя участками памяти в лоб невозможно, наше текущее железо к такому просто не приспособлено. Но потребность писать код подобный нашему синтетическому примеру была есть и будет. И раз уж железяка нам не может помочь - будем что-то придумывать на уровне алгоритма самой программы.

Multi-Word Compare-And-Swap

В 2002м году вышла статья A Practical Multi-Word Compare-and-Swap Operation by Timothy L. Harris, Keir Fraser and Ian A. Pratt. В ней был предложена реализация алгоритма CASN - гарантированное атомарное изменение N atomic переменных. Я не буду ее пересказывать, авторы постарались на славу и не только предложили алгоритм но и доказали что он рабочий и корректный (а это в теме concurrency самое главное).

Главные мысли которые нам нужны чтобы приступить к кодированию:

• Вводим примитив "дескриптор" - он содержит внутри себя операции которые нужно применить атомарно.
• Вместо мьютексов мы рядом с нашими атомиками хранящими ценные данные храним еще и ссылку на дескриптор - это сигнал программе что над этим кусочком данных кто-то уже начал работу.

Вместе эти два фактора дают нам возможность внедрить логику "доталкивания". Пример: если поток №1 начал операцию и его внезапно усыпил ОС то поток №2 которому нужна та же ячейка памяти для своей операции вместо того чтобы блокироваться и засыпать поток №2 пытается применить ту самую операцию. Вспоминаем определение lock-free - код гарантирует что в случае конкурентного исполнения кто-то обязательно достигнет прогресса.

Похоже что авторам удалось придумать то что мы так долго искали. Давайте попробуем приземлить идеи на практику.

Show me your code

TLDR: Весь код в формате Go Playground доступен здесь. Переходите и экспериментируйте. Перейдем к рассмотрению кода, он будет состоять из нескольких примитивов:

• Аккаунт - место где мы храним денежку.
• Транзакция - абстракция хранящая наши операции, гарантирующиая детерминированное исполнение.
• Операция - примитив хранящий данные о том в каком состоянии данные сейчас, и какой инкремент нужно произвести.

Начнем со структуры account

type Account struct {
	balance atomic.Pointer[balance]
}

type balance struct {
    amount int64
    tx *tx
}

func (b *balance) isInFlight() bool { return b.tx != nil }

func NewAccount(amount int64) *Account {
	a := &Account{}
	a.balance.Store(&balance{amount: amount})
	return a
}

func (a *Account) Balance() int64 {
	return a.readBalance().amount
}

// readBalance возвращает реальное значение ячейки,
// помогая завершить чужую транзакцию если ячейка захвачена.
func (a *Account) readBalance() *balance {
	for {
		b := a.balance.Load()
		if !b.isInFlight() {
			return b
		}
		b.tx.commit()
	}
}

// Transfer — lock-free перевод через Multi-word CAS.
func (from *Account) Transfer(to *Account, amount int64) {
	for {
		t := &tx{}
		t.firstOp = newOperation(from, t, -amount)
		t.secondOp = newOperation(to, t, +amount)

		// Фиксируем порядок операций по адресу, чтобы "доталкивание" работало корректно:
		// без порядка A помогает B, B помогает A → бесконечная рекурсия.
		if uintptr(unsafe.Pointer(from)) > uintptr(unsafe.Pointer(to)) {
			t.firstOp, t.secondOp = t.secondOp, t.firstOp
		}

		if t.commit(); t.succeeded() {
			return
		}
	}
}

Что мы видим:

• Бесконечный цикл внутри функции Transfer, классика non-blocking алгоритмов. Пробуем сделать что-то полезнон пока не достигнем успеха.
• Чтение баланса - не просто возвращает нам цифру - под капотом идет проверка на наличии inFlight операции, и если есть - мы ее "доталкиваем" как писали в статье.

Перейдем теперь на уровень ниже - реализация примитива транзакция:

type txStatus int32

const (
	txUnresolved txStatus = iota
	txSucceeded
	txFailed
)

// tx — дескриптор атомарной операции над двумя счетами.
type tx struct {
	done     atomic.Int32
	firstOp  operation
	secondOp operation
}

func (t *tx) status() txStatus { return txStatus(t.done.Load()) }
func (t *tx) resolved() bool   { return t.status() != txUnresolved }
func (t *tx) succeeded() bool  { return t.status() == txSucceeded }
func (t *tx) tryResolve(ok bool) {
	if ok {
		t.done.CompareAndSwap(int32(txUnresolved), int32(txSucceeded))
	} else {
		t.done.CompareAndSwap(int32(txUnresolved), int32(txFailed))
	}
}

func (t *tx) commit() {
	if !t.resolved() {
		t.tryResolve(t.prepare())
	}
	if t.succeeded() {
		t.applyProgress()
	} else {
		t.applyRollback()
	}
}

func (t *tx) prepare() bool {
	return t.firstOp.prepare() && t.secondOp.prepare()
}

func (t *tx) applyProgress() {
	t.firstOp.tryFinalize()
	t.secondOp.tryFinalize()
}

func (t *tx) applyRollback() {
	t.firstOp.tryRestore()
	t.secondOp.tryRestore()
}

Что мы видим:

• Транзакция может быт в статусе - unresolved, success, failure
• Транзация состоит из двух фаз - подготовка (prepare) и фиксация (commit)
• если по каким то причинам не удалось успешно подготовить все вложенные в транзакцию операции то мы падаем и пробуем заново (см код выше)

Перейдем к заключительной детали паззла - примитив операция:

 type operation struct {
	acc    *Account
	before *balance
	after  *balance
}

func newOperation(acct *Account, t *tx, delta int64) operation {
	current := acct.readBalance()
	return operation{
		acc:    acct,
		before: current,
		after:  &balance{amount: current.amount + delta, tx: t},
	}
}

func (op *operation) tx() *tx                     { return op.after.tx }
func (op *operation) current() *balance           { return op.acc.balance.Load() }
func (op *operation) isClaimed(cur *balance) bool { return cur == op.after }
func (op *operation) isStale(cur *balance) bool   { return cur != op.before }
func (op *operation) tryClaim() bool {
	return op.acc.balance.CompareAndSwap(op.before, op.after)
}

func (op *operation) tryFinalize() {
	op.acc.balance.CompareAndSwap(
	    op.after, 
	    &balance{amount: op.after.amount},
	)
}

func (op *operation) tryRestore() {
	op.acc.balance.CompareAndSwap(op.after, op.before)
}

func (op *operation) prepare() bool {
	for {
		if op.tx().resolved() {
			return op.tx().succeeded()
		}
		cur := op.current()
		switch {
		case op.isClaimed(cur):
			return true
		case cur.isInFlight():
			cur.tx.commit()
			continue
		case op.isStale(cur):
			return false
		default:
			op.tryClaim()
		}
	}
}

Что мы видим:

• Бесконечный цикл, крутимся пока либо сами не допушим транзакцию либо нам не поможет сосед. Ну или мы настолько отстали что нужно падать с false чтобы управляющий код сделал ретрай.

Опытные ребята наверняка сейчас поймают эффект легкого дежавю. Код что-то напоминает. Действительно, то что у нас получилось это оптимистичный 2 phase commit. Еще раз убеждаемся что Concurrency и Distributed Systems идут рука об руку.

Завершаем разбор исходного кода, функция main:

func main() {
    // простой пример, без concurrency
	a := NewAccount(1000)
	b := NewAccount(0)
	fmt.Printf("before: a=%d, b=%d\n", a.Balance(), b.Balance())
	a.Transfer(b, 300)
	fmt.Printf("after:  a=%d, b=%d\n", a.Balance(), b.Balance())

	// много параллельных операций, убедимся что 
	// не теряем и не делаем леньги из воздуха :)
	const (
		numAccounts   = 10
		initialFunds  = 1000
		numGoroutines = 50
		numTransfers  = 200
	)

	accounts := make([]*Account, numAccounts)
	for i := range accounts {
		accounts[i] = NewAccount(initialFunds)
	}
	wantTotal := int64(numAccounts * initialFunds)

	var wg sync.WaitGroup
	for range numGoroutines {
		wg.Add(1)
		go func() {
			defer wg.Done()
			rng := rand.New(rand.NewSource(rand.Int63()))
			for range numTransfers {
				from := accounts[rng.Intn(numAccounts)]
				to := accounts[rng.Intn(numAccounts)]
				if from == to {
					continue
				}
				from.Transfer(to, int64(rng.Intn(10)+1))
			}
		}()
	}
	wg.Wait()

	var total int64
	for _, acc := range accounts {
		total += acc.Balance()
	}
	fmt.Printf("conservation check: want=%d got=%d ok=%v\n", wantTotal, total, total == wantTotal)
}

Выводы или "Зачем так сложно?"

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

Если вы дочитали до этого места у вас наверняка возник вопрос вынесенный в заголовок. И он абсолютно справедливый и здравый. Скорее всего в продакшне напрямую такой никто из нас никогда не писал и даже не видел. Скажу больше - именно в таком виде встретить реализацию структуры данных и алгоритма практически невозможно - слишком наивно и дорого по перформансу.

Но так или иначе идеи которые мы рассмотрели это фундамент на основе которого построены например:

• Software Transactional Memory подход который очень классно себя показал в функциональных языках программирования. Про него сделаю отдельный пост в канале или заметку.
• Lock-free структуры данных (идеи MCAS нашли отклик в реализации дереьев поиска, а идея "доталкивания" присутствует дефакто стандартной lock-free реализации очереди - Michael-Scott lock-free queue.

На этом всё, надеюсь вам было интересно и не скучно. В следующих постах мы будем уходить подальше от академических изысков и будем искать ответы на вопросы вида:

• А что по перформансу? Lock-Free уделывает мьютексы?
• Почему мы все еще пользуемся мьютексами если можно без них?

Спасибо, что дочитали, до встречи в новых постах!

🟢 Backend, Software Engineering

• Проблема развития Junior / Middle Backend разработчиков
• Цикл постов о ресурсах по прокачке в Backend #0 #1 #2 #3
• Хорошие практики разработки типичных back-end приложений
• Как практиковаться начинающему разработчику?
• Подборка ресурсов по Git
• Полезные технические ресурсы программисту-стартаперу
• Научные статьи по Computer Science
• Саморазвитие в Computer Science через практику языка Си

🟢 Docker, K8s, Microservices

• Цикл постов о виртуализации и контейнерах #1 #2 #3 #4
• TOP 3 ресурса по погружению в Docker, Kubernetes, Microservices, Cloud Native.
• Подборка для изучения и практики Kubernetes
• What happens when ... Kubernetes edition!
• Создать K8s кластер в одну команду
• Монолитная архитектура
• Прагматичный взгляд на Микросервисную архитектуру.

🟢 Linux, DevOps, SRE

• Подборка бесплатных ресурсов по изучению командной строки и утилит Linux #1 #2 #3
• Изучение DevOps через упражнения
• Бесплатный видеокурс по основам SRE от инженеров Google Cloud
• SRE Interview Preparation Guide
• Обзор сетевых очередей Linux

🟢 Concurrency, Performance

• Concurrency. Multiprocess, Multithreading и Async
• Concurrency. Synchronization и Consistency
• Продуманная оптимизация
• SIMD

🟢 Языки программирования

🔵 Golang

• Как я изучал язык Go
• Тулинг для Go разработчика

🔵 Ruby

• 💎 Learn Ruby. My way
• 💎 Learn Rails. My way
• 💎Complete Guide to setup VS Code for Ruby on Rails (Debugger, Linter, Completion, Formatting)

🔵 Python

• 🐍 Как я изучал Python. Step 1 - Изучение языка.
• 🐍 Как я изучал Python. Step 2 - Инструменты и библиотеки.
• 🐍 Как я изучал Python. Step 3 - Concurrency
• 🐍Практический гайд по процессам и потокам (и не только) в Python
• 🐍 Многопоточность в Python: очевидное и невероятное

🟢 OLTP СУБД

• Подборка книг по СУБД. От новичка до эксперта.
• Путеводитель по базам данных
• Use The Index, Luke
• Подборка классных и бесплатных ресурсов по PostgreSQL
• Партиционирование в PostgreSQL
• Пагинация на примере PostgreSQL. От простого к сложному

🟢 OLAP. Data Engineering

• Fundamentals of Data Engineering
• 7 баз данных за 7 недель, Clickhouse

🟢 Kafka

• Простым языком об Apache Kafka, как, зачем и почему
• Kafka Connect Tutorial

🟢 Big Tech Interview

• Всё, что нужно знать о System Design для прохождения интервью и не только
• Бесплатный, минимально необходимый план подготовки к алгоритмическому интервью в компанию мечты.

🟢 Soft Skills, Mindset

• 1 to 1 - что это, зачем, как подготовиться?
• Какие вопросы стоит задать перед тем как принять оффер? #1 #2
• Зарплатные переговоры
• Как правильно задавать вопросы, если ты начинающий айтишник
• Наглядный жизненный урок
• Почему обучение — это новая прокрастинация

🟢 Личное

• Первые шаги в профессии. Ruby
• Рост в Ruby до уровня Senior
• Как я в первый раз попробовал покорить Big Tech
• Вторая попытка свичнуться из Ruby или как уволиться отработав 2х недели и научиться слышать себя.
• Как я менял стек. Success story после череды неудач.

Плоская оргструктура - хороша тем что все решения принимаются быстро а минус в том что руководитель в такой системе станет бас фактором. Так как руководить в одиночку всеми специалистами напрямую задачка для человека с высоким уровнем компетенций.

Линейная / иерархическая оргструктура - добавляет иерархии, появляется несколько уровней управления, руководители взаимозаменяемы и все довольно прозрачно с точки зрения полномочий и ответственности. К минусам стоит отнести что скорость принятия решений низкая.

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

Проектная оргструктура. Хороша тем, что каждый департамент сфокусирован на своем проекте и приносит конкретную измеримую пользу бизнесу. К минусам можно отнести проблему что в такой оргструктуре с ростом компании будет проблемой "выравнивание" культуры между проектами, выравнивание экспертизы специалистов. Сфера применимости - outsource компании.

Матричная оргструктура - одна из самых популярных оргструктур, к ее плюсам можно отнести высокую гибкость, а также то что менеджмент уделяет внимание всем возможным аспектам (delivery / people management). Основные минусы такой оргструктуры - двоевластие и размытие ответственности.

Дивизионная оргструктура. Плюсы: масштабируемся быстро, просто создаем новый дивизион. Четкие уровни управления. К минусам можно отнести высокий уровень дезинтеграции и дублирование функций в каждом дивизионе.

• овладеть основными компонентами фреймворка.
• углубиться в них и решать задачи эффективно с точки зрения времени.
• писать производительный код.
• вырости до уровня Senior и решать сложные задачи.

Аналогичный пост по Ruby - здесь.

💎Getting Started

У Rails просто офигенная документация, я очень часто использовал ее для решения повседневных задач. Для меня это один из примеров того как должна выглядеть дружелюбная к разработчику дока.

• https://rusrails.ru/
• http://guides.rubyonrails.org/

Ну и само собой я не могу не отметить классику - туториал Майкла Хартла по написанию клона твиттера

Книги

• Agile Web Development with Rails 7 - тот самый туториал про интернет магазин
• The Rails 7 Way - энциклопедия по Ruby on Rails

💎 Best practices

Репозитории компании разрабатывающих на Ruby в которых они ведут свои соглашения и правила по написанию кода на Ruby on Rails.

Если вы хотите максимально быстро прокачаться лучше ресурса не найти - учитесь на чужих ошибках и впитывайте чужой опыт.

• Best practices for running Rails in production
• Describing the development process used by the Infinum Rails Team.
• A guide for programming in style.

💎 Templates

Для того чтобы с нуля стартовать с крутым набором инструментом и не тратить время на настройку локального окружения можно воспользоваться готовыми сборками:

• https://github.com/ledermann/docker-rails
• https://github.com/bullet-train-co/bullet_train

Статья о том из каких библиотек строят проекты Evil Martians - Gemfile of dreams: the libraries we use to build Rails apps

💎 Performance

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

• Design Rails JSON API with performance in mind - JTWay
• JSON Serialization in Rails: A Complete Guide
• Архитектура сервера PUMA
• Configuring Puma, Unicorn and Passenger for Maximum Efficiency
• Threads are evil. Async Ruby in action - Подробный гайд о тредах, процессах и асинхронности
• Ruby CPU and Memory - Pareto principle - как профилировать производительность приложений

💎 Active Record

Всё что нужно знать чтобы хорошо уметь готовить ActiveRecord и выжимать из него максимум

• самый популярный вопрос на собеседовании - Making sense of ActiveRecord joins, includes, preload, and eager_load
• Rails Database Best Practices (About queries)
• Getting the Most out of your Database with Ruby on Rails (About configuration)
• Understanding Locking in Rails ActiveRecord

PostgreSQL

• Indexes on Rails: How to Make the Most of Your Postgres Database
• Как правильно готовить Ruby on Rails для работы с PostgreSQL

💎 Архитектура / Паттерны

Rails довольно гибкий и мощный фреймворк в котором многое держится на соглашениях, и если бездумно ими пользоваться то проект может превратиться в запутанное спагетти

• Rails hurts, because we're using it wrong. Let's fix that! - в чём проблема с архитектурой Rails?
• Rails design patterns - простые варианты как улучшить кодовую базу

💎 Rails Security

Работая в production важно не только делать фичу но и писать безопасный код и защищать данные клиентов от вредоносного вмешательства.

• Checklist of security precautions for Ruby on Rails applications.
• SQL Injections Guide

💎 Блоги компаний

• https://thoughtbot.com/blog
• https://evilmartians.com/chronicles

• Кем и зачем она была создана.
• Доступно рассмотрим из каких абстракций и примитивов состоит.
• За счет чего достигается высокая пропускная способность.
• Поговорим про отказоустойчивость и за счет чего она достигается.

Многие новички часто задают вопрос "Почему?" касаемо любой новой технологии. Надеюсь что этой статьей я смогу ответить на большинство “Почему?” связанных с Apache Kafka.

Краткий экскурс в историю. Причины появления Kafka

Apache Kafka — распределенный, реплицируемый журнал с открытым исходным кодом, разрабатываемый в рамках фонда Apache. Написан на Java и Scala. Сложно, непонятно, но пойдем дальше.

Разработан в компании LinkedIn в начале 2010х, далее был передан в фонд Apache. Сейчас авторы проекта работают в компании Confluent которую сами и основали.

Kafka зародилась в LinkedIn как решение для задач про передаче больших объемов данных между сервисами, в частности для построения моделей машинного обучения. Классические брокеры сообщений не могли обеспечить нужную пропускную способность (throughput) и LinkedIn решил создать “свой велосипед”

Из чего состоит Apache Kafka?

Что же придумали разработчики Kafka чтобы достигнуть высокой скорости передачи данных? Ответ: сделали брокер максимально простым и топорным: он хранит все сообщения на диске и позволяет делать с ними только 2 простых но очень эффективных с точки зрения железа операции:

• добавить запись в конец файла
• прочитать файл с позиции X до позиции Y

Как следствие, нельзя:

• искать, фильтровать данные на стороне кластера, только на стороне клиента
• удалять данные (за удаление данных отвечает Kafka)

За чтение сообщений отвечают потребители (consumers).За отправку сообщений в Kafka отвечают производители (producers). Писатель и читатель при взаимодействии с брокером указывают обязательный параметр - topic или тема. За счет него логически идет разделение потоков данных на стороне kafka.

Чуть подробнее о потребителях

Потребитель по порядку читает сообщения из темы и направляет брокеру Kafka специальные сообщения о том что он прочитал. В этом самом сообщении брокеру передается специальный идентификатор offset - позиция на которой остановился читатель.

Если читатель перезапускается то при повторном подключении к брокеру он получит последнее актуальное состояние offset и продолжит читать сообщения с того места на котором остановился. Offset в каком то смысле похож на индекс в массиве. Зная индекс мы можем сразу обратиться к нужному элементу и продолжить обработку всех последующих за ним.

И это всё? Просто файлы?

На самом деле нет. Представим, у нас есть топик X, в который пишется очень много сообщений. Мы понимаем что в одного потребителя не справляемся и хотим распаралелить процесс. Если наш topic это просто файл то читать из него несколькими потребителями становится проблематично, так как нужно потребителей синхронизировать между собой, чтобы не получилось так что каждый потребитель обрабатывает одно и то же. Синхронизация это время, и оно очень ценно, с ростом количества потребителей время синхронизации будет линейно расти.

Что же делать? Разработчики Kafka решили разделить топик на N независимых сущностей (по факту это те же файлы) назвав партициями. Теперь нас ничто не останавливает от того чтобы создать N потребителей и быстро вычитывать данные, так как потребителей не нужно синхронизировать, каждый читает только свой кусочек, свою партицию.

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

Набор потребителей в Kafka именуется Consumer Group (группа потребителей). При добавлении или выходе потребителей из группы kafka требуется ребалансировка, чтобы либо выделить новому потребителю свою партицию либо назначить уже работающему консьюмеру партицию которую читал умерший консьюмер.

Что насчет отказоустойчивости?

Репликация

Что может пойти не так с Kafka брокером? Да всё что угодно, например сервер на котором он развернут может выйти из строя. В таком случае мы не сможем ни писать сообщения ни читать их, а это грусть печаль, бизнес стопорится, деньги не зарабатываются. Что спасает в таких случаях? Правильно, избыточность.

Если базы данных умеют восстанавливаться от сбоев за счет репликации то почему бы и Kafka не перенять эту механику? Так посчитали и разработчики Kafka и добавили механизм репликации данных между узлами. Теперь вместо одного брокера получаем кластер брокеров.

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

Репликация конфигурируется для каждого топика через переменную Replication Factor - сколько копий для партиции нужно иметь в кластере.

Controller

У одного узла в кластере Kafka есть специальная роль - controller, в его обязанности входит как раз таки на основании состояния кластера назначать лидеры для партиций топиков. Лидер партиция - партиция с которой взаимодействуют потребители и произовдители.  Аналог мастера в мире БД. Остальные копии лидер-партиции являются ведомыми и их обязанность - участвовать в репликации.

Для того чтобы гарантировать в кластере уникальность роли controller а также хранить и отслеживать состояние брокеров кластера Kafka использует kraft / Apache Zookeper.

Итог

В данной статье я постарался описать максимально простым и понятным языком как устроен брокер Apache Kafka и почему именно так а не как иначе. Многие вещи в этой статье не удалось рассмотреть, например семантики доставки, это было сделано осознанно чтобы не растягивать статью.

Надеюсь у меня получилось доступно донести концепции и теперь вас не испугать популяным определением которую дают Kafka на просторах сети - распределенный, реплицируемый журнал