Структура потребления

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

Режим push сложно адаптировать к потребителям с разной скоростью потребления, поскольку скорость отправки сообщений определяется брокером. Его цель — доставить сообщения как можно быстрее, но это может легко привести к тому, что у потребителя не будет времени обработать сообщение. Типичными проявлениями являются отказ в обслуживании и перегрузка сети. Режим извлечения может потреблять сообщения с соответствующей скоростью в зависимости от способности потребителя потреблять сообщения.

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

Стратегия распределения разделов

потребитель В группе есть несколько потребителей, а тема имеет несколько разделов, поэтому обязательно потребуется распределение разделов, то есть определение того, какой раздел каким потребителем используется. Kafka предоставляет 3 типа потребителей Стратегия распределения разделов：RangeAssigor、RoundRobinAssignor、StickyAssignor。

Интерфейс PartitionAssignor используется пользователями для определения и реализации алгоритмов распределения разделов для достижения распределения разделов между потребителями. Члены группы потребителей подписываются на интересующие их темы и передают эти отношения подписки Брокеру как координатору группы подписки. Координатор выбирает одного из потребителей для выделения разделов для этой группы потребителей и передает результаты выделения всем потребителям в группе потребителей. Kafka по умолчанию использует алгоритм распределения RangeAssignor.

RangeAssignor

RangeAssignor назначает независимые разделы каждой теме. Для каждой Темы сначала сортируются разделы по идентификатору раздела, затем сортируются потребители, подписавшиеся на группу потребителей этой Темы, а затем разделы распределяются между потребителями как можно более равномерно. Это может быть только максимально сбалансированным, поскольку количество разделов может не делиться на количество потребителей, поэтому некоторые потребители будут распределены по большему количеству разделов. Схема распределения следующая:

Алгоритм выделения разделов следующий:

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, Subscription> subscriptions) {
    Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());
    //Цикл for обрабатывает несколько подписанных тем отдельно.
    for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
        String topic = topicEntry.getKey();
        List<String> consumersForTopic = topicEntry.getValue();
 
        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        if (numPartitionsForTopic == null)
            continue;
        //Провести сортировку по потребителям
        Collections.sort(consumersForTopic);
        //Рассчитываем среднее количество разделов, выделенных каждому потребителю
        int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
        //Рассчитываем количество дополнительных разделов после среднего распределения
        int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
 
        List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
        for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
            //Вычисляем i-го потребителя и выделяем начальную позицию раздела
            int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
            //Рассчитываем количество разделов, назначенных i-му потребителю
            int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
            assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
        }
    }
    return assignment;
}

Очевидная проблема с этим методом распределения заключается в том, что по мере увеличения количества тем, на которые подписаны потребители, проблема дисбаланса становится все более серьезной. Например, в приведенном выше сценарии с 4 разделами и 3 потребителями C0 выделит еще один раздел. Если вы в данный момент подписываетесь на тему с номером раздела 4, C0 будет выделен на один раздел больше, чем C1 и C2. Таким образом, C0 будет выделено на два раздела больше, чем C1 и C2 в целом. увеличивается, эта ситуация будет становиться все более серьезной. Результаты задания:

Вставьте сюда описание изображения

Как показано на рисунке, подпишитесь на 2 темы, каждая тема имеет 4 раздела и всего 3 потребителя.

C0：[T0P0,T0P1,T1P0,T1P1] C1：[T0P2,T1P2] C2：[T0P3,T1P3]

RoundRobinAssignor

Стратегия распределения RoundRobinAssignor состоит в том, чтобы отсортировать разделы всех тем и всех потребителей, подписанных в группе потребителей, и распределить их как можно более равномерно (RangeAssignor сортирует и выделяет разделы одной темы). Если список Тем, на которые подписаны потребители в группе потребителей, одинаков (каждый потребитель подписан на одну и ту же Тему), то результат распределения будет максимально сбалансированным (разница в количестве разделов, выделяемых между потребителями, не будет превышать 1). . Если списки тем, на которые оформлена подписка, различаются, результаты распространения не гарантируются «максимально сбалансированными», поскольку некоторые потребители не участвуют в распространении некоторых тем.

Вставьте сюда описание изображения

В случае двух вышеупомянутых тем, по сравнению с предыдущей стратегией распределения RangeAssignor, распределение разделов может быть более сбалансированным. Однако, учитывая эту ситуацию, предположим, что есть три потребителя с именами C0, C1 и C2, и есть 3 темы T0, T1 и T2 с 1, 2 и 3 разделами соответственно, и C0 подписывается на T0, а C1 подписывается на T0 и T1, то C2 подписывается на T0, T1 и T2, то результаты выделения RoundRobinAssignor будут следующими:

Вставьте сюда описание изображения

Вроде бы раздача была максимально сбалансирована, но можно обнаружить, что С2 отвечает за потребление 4-х разделов, а С1 подписывается на Т1. Можно ли сделать более сбалансированным, если Т1Р1 отдать на потребление С1?

StickyAssignor

Целью алгоритма выделения разделов StickyAssignor является как можно меньшая корректировка изменений распределения разделов на основе результатов предыдущего выделения при выполнении нового выделения, тем самым сохраняя накладные расходы, вызванные изменениями в распределении разделов. Sticky — это «липкий», что можно понимать как «липкий» результат распределения — каждое изменение распределения требует минимальных изменений по сравнению с предыдущим распределением. Его цели двояки:

Распределение разделов должно быть максимально сбалансированным. Результат каждого перераспределения должен соответствовать результату предыдущего распределения. Когда эти две цели конфликтуют, первая цель имеет приоритет. Первая цель — это то, чего пытается достичь каждый алгоритм назначения, но вторая цель действительно отражает характеристики StickyAssignor.

Алгоритм StickyAssignor относительно сложен. Следующий пример иллюстрирует эффект выделения (по сравнению с предварительным условием RoundRobinAssignor).

Существует 4 темы: T0, T1, T2, T3, и каждая тема имеет 2 раздела. Есть 3 потребителя: C0, C1 и C2. Все потребители подписаны на эти 4 раздела.

Красная стрелка выше представляет измененное распределение разделов. Видно, что стратегия распределения StickyAssignor претерпела небольшие изменения.