#messagebroker

Apache Kafka – это высокопроизводительная распределенная потоковая платформа с открытым исходным кодом, разработанная изначально в LinkedIn, которая позволяет приложениям публиковать, подписываться, хранить и обрабатывать потоки записей в реальном времени, обеспечивая при этом высокую пропускную способность, масштабируемость и отказоустойчивость для широкого круга сценариев использования, включая обработку больших данных и микросервисные архитектуры. Эта платформа функционирует как распределенный журнал транзакций. Она способна обрабатывать триллионы событий в день. Таким образом, Kafka стала стандартом де-факто для построения потоковых конвейеров данных. Она используется многими крупными компаниями. Ее надежность и производительность делают ее идеальным решением для современных распределенных систем.  

Что такое Apache Kafka и почему это не RabbitMQ

Многие новички часто путают Kafka с классическими очередями сообщений. Традиционные брокеры обычно удаляют сообщение сразу после его успешного прочтения клиентом. В Apache Kafka данные сохраняются на диске в течение заданного периода времени. Вы можете перечитывать одни и те же события многократно разными приложениями. Это делает платформу идеальным хранилищем для событийной архитектуры и аналитики. Главное архитектурное различие кроется в способе доставки данных потребителям. В очередях брокер сам «толкает» сообщения в сторону подписчиков. Kafka использует модель вытягивания (pull), где потребители сами запрашивают нужные данные. Такой подход предотвращает перегрузку медленных клиентов при резких всплесках трафика. Система легко масштабируется горизонтально путем добавления новых серверов в кластер. Kafka хранит данные в виде упорядоченной последовательности записей. Сообщения в топиках не удаляются сразу после прочтения клиентом. Высокая производительность достигается за счет линейной записи на диск. Поддержка нескольких подписчиков на один и тот же поток данных. Apache Kafka предоставляет разработчикам гарантии порядка сообщений внутри одной партиции. Это критически важно для финансовых систем и обработки транзакций в реальном времени.    

Ключевые аспекты и архитектура Apache Kafka

Архитектура системы строится вокруг концепции распределенного кластера из нескольких узлов. Каждый такой узел называется брокером и отвечает за хранение части данных. Клиентские приложения взаимодействуют с кластером через стандартизированный протокол обмена данными. Разделение ответственности позволяет системе работать без простоев даже при выходе серверов из строя. Архитектура Kafka включает несколько основных компонентов. Все они работают вместе для обеспечения её функциональности.

  Брокеры, Топики и Партиции Данные в системе организуются в логические категории, которые называются топиками. Каждый топик физически разделяется на несколько более мелких частей — партиций. Партиция представляет собой упорядоченный лог, в который данные только дописываются в конец. Именно разбиение на партиции позволяет распределять нагрузку между разными брокерами кластера. Партиции позволяют параллельно считывать данные из одного большого топика. Каждая запись в партиции получает свой уникальный номер (offset). Количество партиций напрямую влияет на максимальную пропускную способность системы. Брокеры равномерно распределяют партиции по дисковым подсистемам серверов.

Разделение данных на партиции является ключом к горизонтальному масштабированию всей платформы.   Продюсеры и Консьюмеры Продюсеры (Producers) — это приложения, которые отправляют данные в топики кластера. Они могут выбирать, в какую именно партицию направить конкретное сообщение. Обычно это делается на основе ключа сообщения с помощью специального алгоритма хеширования. Консьюмеры (Consumers) считывают данные и обрабатывают их согласно бизнес-логике приложения. Продюсеры могут работать в асинхронном режиме для повышения скорости. Консьюмеры объединяются в группы для совместной обработки одного топика. Система автоматически балансирует нагрузку между участниками одной группы потребителей. Такое разделение позволяет независимо масштабировать генерацию и потребление потоков данных.   Consumer Groups и Offset Механизм групп потребителей позволяет эффективно распределять чтение данных между несколькими экземплярами приложения. Каждая партиция внутри группы закрепляется только за одним конкретным потребителем. Это гарантирует, что одно сообщение не будет обработано дважды разными участниками группы. Offset (смещение) — это позиция последнего прочитанного сообщения в каждой отдельной партиции. Kafka хранит информацию о смещениях в специальном системном топике. При перезагрузке потребитель продолжает чтение с последнего зафиксированного места. Группы позволяют легко реализовать как модель очереди, так и модель публикации-подписки. Правильное управление смещениями обеспечивает гарантию доставки сообщений «как минимум один раз».  

Как Kafka обеспечивает надежность и скорость

Скорость работы платформы часто вызывает удивление у инженеров, привыкших к медленным БД. Секрет кроется в использовании последовательного ввода-вывода и системного кэша операционной системы. Вместо случайного доступа к данным, Kafka просто дописывает байты в конец файлов. Современные жесткие диски показывают отличные результаты при такой линейной нагрузке на шпиндель.   Репликация и ISR (In-Sync Replicas) Надежность хранения данных достигается путем создания копий партиций на разных брокерах. Одна реплика назначается лидером, а остальные становятся ведомыми последователями (followers). Все операции записи и чтения проходят исключительно через лидера для обеспечения согласованности. Ведомые брокеры просто копируют данные к себе в фоновом режиме.  

Список синхронных реплик (ISR) содержит узлы, которые полностью догнали лидера. При падении лидера новым главным узлом становится один из участников ISR. Параметр acks в настройках продюсера определяет степень подтверждения записи. Репликация позволяет системе выживать даже при потере нескольких физических серверов одновременно.   Zero-copy и Sequential I/O Для передачи данных по сети Kafka использует технологию zero-copy. Это позволяет передавать данные из дискового кэша напрямую в сетевой сокет. Процессор при этом не тратит циклы на копирование байтов в пользовательское пространство. Данная оптимизация радикально снижает нагрузку на CPU и увеличивает общую пропускную способность. Данные передаются в том же формате, в котором лежат на диске. Отсутствие десериализации на стороне брокера экономит ценные ресурсы системы. Операционная система эффективно кэширует горячие данные в оперативной памяти. Благодаря этим техникам Kafka способна обрабатывать миллионы сообщений в секунду  

  Паттерны проектирования топиков и ключей

Выбор ключа сообщения определяет, в какую партицию попадут ваши данные. Если вы используете null в качестве ключа, сообщения распределяются по кругу (Round Robin). Однако для многих задач критически важно сохранять порядок связанных событий. Например, все транзакции одного пользователя должны обрабатываться в строгой последовательности. Сообщения с одинаковым ключом всегда попадают в одну и ту же партицию. Плохой выбор ключа может привести к перекосу данных (Data Skew). Некоторые партиции могут стать слишком тяжелыми и замедлить работу кластера. Проектирование схемы данных требует понимания того, как консьюмеры будут группировать информацию.   Admin API Для управления объектами кластера, такими как топики, брокеры и конфигурации, используется Admin API. Этот API предоставляет программный доступ к административным функциям. Он позволяет автоматизировать операции развертывания и мониторинга. Это делает управление кластером более эффективным.

  ZooKeeper и KRAFT Традиционно, Apache ZooKeeper использовался для хранения метаданных кластера Kafka. Он отвечал за управление брокерами, топиками и группами потребителей. Однако, начиная с версии Kafka 2.8, и особенно с полной реализацией в версии 4.0, платформа переходит на новый режим консенсуса — KRAFT (Kafka Raft Metadata). KRAFT полностью устраняет зависимость от ZooKeeper. Это упрощает архитектуру и развертывание кластера. Такая миграция делает управление более простым и производительным. Режим KRAFT встроен непосредственно в брокеры Kafka.      

Сценарии использования Apache Kafka в реальных проектах

Одним из самых популярных сценариев является агрегация технических логов из тысяч сервисов. Вместо записи в локальные файлы, приложения отправляют события напрямую в централизованный кластер. Это позволяет анализировать ошибки и производительность системы в режиме реального времени. Команда эксплуатации может быстро реагировать на инциденты еще до жалоб пользователей. Другой важный кейс — архитектура Event Sourcing. Здесь Kafka выступает в роли единственного источника истины для всех микросервисов. Каждое изменение состояния системы записывается как событие в неизменяемый лог. Любой сервис может воссоздать свою базу данных, просто перечитав историю событий с самого начала.  

Развертывание Apache Kafka 4.2.0 в режиме KRaft на Ubuntu

Для работы Kafka требуется среда исполнения Java. Мы будем использовать OpenJDK 17 или 21, которые обеспечивают отличную стабильность и производительность. Все действия выполняются в консоли сервера.

Подготовка системы и установка

Первым делом обновите индексы пакетов и установите Java. После этого скачайте дистрибутив с официального зеркала Apache мы будем использовать последнюю доступную нам сейчас версию Kafka KRAFT 4.2.0 .

#--- Установите Java sudo apt install default-jdk -y #--- Скачайте архив wget https://dlcdn.apache.org/kafka/4.2.0/kafka_2.13-4.2.0.tgz #--- Распакуйте его tar -xzf kafka_2.13-4.2.0.tgz #--- Перейдите в рабочую директорию cd kafka_2.13-4.2.0

Установка в домашнюю директорию удобна для тестов. Для продакшн-среды рекомендуется перенести файлы в /opt/kafka.

Конфигурация KRaft для Kafka кластера

В режиме KRaft один из брокеров берет на себя роль контроллера. В одноюнитовой (single-node) конфигурации наш сервер будет совмещать роли брокера и контроллера. Нам нужно сгенерировать уникальный идентификатор кластера и отформатировать хранилище.

#--- Генерация UUID - Скопируйте полученную строку. bin/kafka-storage.sh random-uuid #--- Форматирование логов bin/kafka-storage.sh format -t xIX_NIddScSTq7AjG3mmVQ -c config/server.properties --standalone #--- Запуск сервера в режиме Демона ( страшно ) bin/kafka-server-start.sh -daemon config/server.properties #--- Создаем свой первый Топик на кластере Apache Kafka в KRAFT режиме bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic BigDataSchool #--- Подключаемся к топику Продюсером и продюсируем в него то есть пишем в него сообщения bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic BigDataSchool #--- Выичтываем данные из топика Консьюмером ( не забудьте ключ --from-beginning) bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic BigDataSchool --from-beginning

  Флаг - daemon запустит процесс в фоновом режиме. Теперь ваша Kafka готова к приему сообщений.

Базовые команды для работы с данными  

Все стандартные утилиты Kafka находятся в директории bin. Для локальной работы мы будем использовать адрес брокера localhost:9092.

Создание топика

Топик — это логическое хранилище ваших сообщений. При создании важно указать количество партиций и фактор репликации. Для одного узла репликация всегда равна единице.

Отправка сообщений (Продюсер)

Консольный продюсер позволяет отправлять текстовые строки в топик в интерактивном режиме. Каждая новая строка после нажатия Enter станет отдельным сообщением.

После запуска введите текст, например: Привет ты молодец!!! . Для выхода используйте Ctrl+C.

Чтение сообщений (Консьюмер)

Чтобы увидеть данные, которые вы только что отправили, нужно запустить консольный потребитель. Флаг --from-beginning заставит Kafka отдать все сообщения с самого начала истории топика.

Эти команды являются фундаментом для любого администратора или разработчика Big Data. Они позволяют быстро проверить доступность кластера и корректность прохождения данных.

Не отчаивайся если получилось не все - в Школе Больших Данных у нас есть для тебя целый курс по KAFKA:"Администрированию Apache Kafka кластера" - приходи не пожалеешь! А если ты разработчик то еще DEVKI:"Apache Kafka для develeopers"

Экосистема за пределами ядра (Connect и Streams)

Сегодня Apache Kafka — это не просто брокер, а полноценная экосистема инструментов. Фреймворк Kafka Connect позволяет подключать внешние базы данных без написания кода. Вы можете настроить автоматическую выгрузку данных из PostgreSQL прямо в топики Kafka. Существуют сотни готовых коннекторов для облачных хранилищ и поисковых движков вроде Elasticsearch. Kafka Streams — это библиотека для построения приложений потоковой обработки данных на Java. Она позволяет выполнять агрегацию, фильтрацию и объединение потоков внутри вашего приложения. Для тех, кто предпочитает декларативный подход, существует ksqlDB. Этот инструмент дает возможность писать SQL-запросы прямо к потокам событий в реальном времени. Connect упрощает интеграцию с Legacy-системами и внешними API. Streams обеспечивает отказоустойчивую обработку состояний (stateful processing). ksqlDB превращает поток событий в динамически обновляемые таблицы. Использование этих инструментов сокращает время вывода новых аналитических продуктов на рынок.    

Основные дистрибутивы и расширения Apache Kafka

Хотя Kafka является проектом с открытым исходным кодом, существует несколько коммерческих дистрибутивов и платформ, которые добавляют дополнительные функции, инструменты управления и поддержку. Эти дистрибутивы значительно упрощают развертывание и эксплуатацию Kafka в производственной среде. - Confluent Platform: Это одна из самых популярных коммерческих платформ. Confluent Platform предоставляет расширенные возможности. Среди них — Confluent Schema Registry, Confluent Control Center. Она также включает ksqlDB и дополнительные коннекторы. Confluent предлагает полную экосистему вокруг Kafka. - Arenadata Streaming (ADS): Российский дистрибутив Kafka. Arenadata Streaming предоставляет корпоративную поддержку. Он включает дополнительные инструменты для управления и мониторинга. ADS ориентирован на использование в российских IT-инфраструктурах.  

Основные конкуренты Apache Kafka

На рынке потоковой обработки данных у Kafka есть несколько ключевых конкурентов. Каждый из них имеет свои сильные стороны. Выбор зависит от конкретных требований проекта. - RabbitMQ: Это популярный брокер сообщений. RabbitMQ использует протокол AMQP. Он часто применяется для обмена сообщениями между микросервисами. Однако RabbitMQ лучше подходит для очередей сообщений. Он менее масштабируем для потоковой передачи больших объемов данных. - Apache Pulsar: Еще одна распределенная потоковая платформа. Apache Pulsar поддерживает как очереди, так и потоки. Он имеет многоуровневую архитектуру хранения. Pulsar предлагает гибкую масштабируемость. Он является серьезным конкурентом в области потоковых данных. - Amazon Kinesis: Управляемый сервис потоковой передачи данных от AWS. Amazon Kinesis предоставляет возможности для сбора, обработки и анализа потоковых данных. Он полностью интегрирован с другими сервисами AWS. Однако Kinesis является проприетарным решением.  

Заключение

An Introduction to RabbitMQ:

RabbitMQ is a message broker software in Ruby on Rails development where web applications connect to the queue and transfer a message onto it. When one application server sends the request to another application server, RabbitMQ works as a queue manager where requests handled and served one by one. Message transferring is done from one application to another through RabbitMQ.

RabbitMQ uses the standard Advanced Message Queuing Protocol (AMQP) to keep message request in queues and serve easily. The message may contain any information like simple text transfer or task that need to process on another server. This RabbitMQ software saves messages in a queue and keeps in a queue until receiving application get connected and start processing the messages.

Why should we use RabbitMQ in Ruby on Rails Development?

Message broker works on the producer, broker, and consumer logic. Message broker server allows consumer and producer server to respond request fast instead of using resources heavily at one time. The message broker is used to optimize the load on the server and provide services until application executes the task on the requested server. The requested server can be any third party application which performs the task.

Queued messages will be used to send the messages or task to multiple recipients from consumption that leads to the load balancing on the producer and consumer server.

Let take an example of ruby on rails server and bitcoin exchange server (i.e. bitcoind) for btc transaction working.

I have rails server running as producer and bitcoind server as a consumer.

Rails server send multiple request to bitcoind server continuously. There may be chances to slow down bitcoind because of multiple requests send to bitcoind server to process task.

In that case, RabbitMQ can fulfill the role of load balancing and messages passing by keeping the request coming from rails server in a queue and serve request one by one to bitcoind server.

Bitcoind server will take a task from the queue and start processing.

                                                       AMQP role in RabbitMQ:

AMQP store and forward messages to consumers, same as SMTP protocol which is used to deliver mail to mailboxes. Messages start from producer application to the exchanges that help to redirect in queues. Lastly, queues deliver task to the consumer application. It's now consumer's server responsibility to take the messages from queues to process as required.

A virtual channel is used to establish the connection between publishing messages from the queue. All process exchanges done over the channels.

How to use RabbitMQ in Rails application?

Using Bunny ruby RabbitMQ client, we can integrate RabbitMq with rails application. There are some gems available to start with RabbitMQ in ruby like

gem “bunny”

Bunny help rails application to interoperate with the consumer application, it can be built in any technologies irrespective of the size and complexity of the application. Also, add anampq gem in rails Gemfile to use AMQP protocol services.

  gem  ‘amqp’

RabbitMQ installation commands in Ubuntu:

1] Step 1: Update system  

sudo apt-get update

2] Step 2: Install Erlang

RabbitMQ is written in Erlang, you need to install Erlang before you can use RabbitMQ:

a] wget http://packages.erlang-solutions.com/site/esl/esl-erlang/FLAVOUR_1_general/esl-erlang_20.1-1~ubuntu~xenial_amd64.deb

b] sudo dpkg -i esl-erlang_20.1-1\~ubuntu\~xenial_amd64.deb

3] Step 3: Install RabbitMQ

     sudo apt-get installrabbitmq-server

4] Step 4: Start the Server

  sudo systemctl start rabbitmq-server.service

  sudo systemctl enable rabbitmq-server.service

It is one of those methods of the MbJavaComputeNode class that can be overridden

It is called either during deployment (just before the deployment is committed) or on broker startup

If this method throws an exception, then either the deployment of the flow fails or the it fails to start

Perform those tasks that will always work in this method. Here is one example: 

Since the user defined properties cannot be accessed in the constructor. To access these properties at initialization time, implement onInitialize and use this implementation to access the user defined properties

Note: The code that depends on an external resource that might not be there or becomes unavailable then it should not be executed inside an implementation of onInitialize() because the broker will not retry calls to this method unless the broker itself is restarted. If such requirements, like for example, initializing an external connection, need to handled then such steps can be taken during the first call to the evaluate() method. If an exception occurs during the initialization, then it can be thrown so that its handled like any other flow exception (and from then onwards the usual message exception handling algorithm/procedure kicks in)

References:

- http://pic.dhe.ibm.com/infocenter/wmbhelp/v8r0m0/index.jsp?topic=%2Fcom.ibm.etools.mft.doc%2Fac20805_.htm

- http://publib.boulder.ibm.com/infocenter/wmbhelp/v7r0m0/index.jsp?topic=%2Fcom.ibm.etools.mft.doc%2Fac30220_.htm

Rational ClearCase

1- Click Window > Preferences to open the Preferences window.

2- Expand General in the left pane, and click Capabilities.

3- In the Capabilities pane, click Advanced. The Advanced window opens.

4- Expand Team, and ensure that ClearCase SCM Adapter is selected.

5- Click OK to close the Advanced window.

6- Click OK or Apply to apply your changes.

Concurrent Versions System (CVS)

1- Configure CVS by carrying out the following tasks:

Create a directory on your computer, for example, on Windows - c:\CVSRepository.

Start the CVSNT control panel. Select Start > Programs > CVSNT to see the icon on the desktop.

Stop both the CVS Service and the CVS Lock Service.

Select the Repositories tag, click Add and create a new repository. Note that no entry appears on the screen the first time that you do this.

Use the ... button on the next window to select the directory that you created in step 1.a and click OK. Note that when CVS has finished formatting its repository the backslash in the directory name is changed to a forward slash.

Select the Service Status tab and restart both the CVS Service and the CVS Lock Service.

2- Enable the CVS Revision tag to be populated in the Eclipse Version fields in the Message Broker Toolkit. To do this on Windows:

Select Window->Preferences

Expand the Team section and click CVS

Use the drop down in the Default keyword substitution: field and set the value to ASCII with keyword expansion(-kkv)

3- Add the WebSphere® Message Broker file types to the Eclipse CVS configuration. To do this:

Select File Content in the Team section of the window you opened in step 2

Click Add and add msgflow as an allowable file extension. Ensure that the value is set to ASCII.

Repeat the above procedure for the following file extensions that the broker uses: esql, mset, mxsd etc.

Click OK when you have finished.

Rational Team Concert

1- Start the WebSphere Message Broker Toolkit. 2- Open the Work Items perspective and click Window > Open  Perspective > Other > Work Items. 3- Click the Create Repository Connection link in the Team Artifacts view. 4- Follow the dialog and enter the information given to you by your Jazz™ administrator. 5- To add your project to the repository, right-click the project and select Team > Share Project. 6- In the Share Project window, select Jazz Source Control as the repository type.

Source:

http://publib.boulder.ibm.com/infocenter/wmbhelp/v6r1m0/index.jsp?topic=%2Fcom.ibm.etools.mft.doc%2Fac38010_.htm

http://publib.boulder.ibm.com/infocenter/wmbhelp/v6r1m0/index.jsp?topic=%2Fcom.ibm.etools.mft.doc%2Fac26530_.htm

To create a message body from the JMS message payload, JMSInput has to determine the appropriate parser. The entity that leads JMSInput node to this decision is the JMS message type. This value is picked off by JMSInput node with the use of the JMS API. It has to be kept in mind that different JMS message types require different processing/parsing which the following can give some idea about:

mqsicreatebar: compiles all the resources into one bar file before its deployment to the broker