|NO.Z.00059|——————————|BigDataEnd|——|Hadoop&kafka.V44|-------------------------------------------|ka

2022-04-09 14:35:08 阅读：189 来源： 互联网

标签：userData PartitionAssignor 自定义 NO kafka 分配 Subscription

一、自定义分配策略

### --- 自定义分配策略

~~~     自定义的分配策略必须要实现
~~~     org.apache.kafka.clients.consumer.internals.PartitionAssignor接口
~~~     PartitionAssignor接口的定义如下：

Subscription subscription(Set<String> topics);

String name();

Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions);

void onAssignment(Assignment assignment);
class Subscription {
    private final List<String> topics;
private final ByteBuffer userData;
}

class Assignment {
    private final List<TopicPartition> partitions;
    private final ByteBuffer userData;
}

### --- PartitionAssignor接口中定义了两个内部类：Subscription和Assignment。
~~~     Subscription类用来表示消费者的订阅信息，
~~~     # 类中有两个属性：
~~~     topics和userData，分别表示消费者所订topic列表和用户自定义信息
~~~     PartitionAssignor接口通过subscription()方法来设置消费者自身相关的Subscription信息，
~~~     注意到此方法中只有一个参数topics，与Subscription类中的topics的相互呼应，
~~~     但是并没有有关userData的参数体现。
~~~     为了增强用户对分配结果的控制，
~~~     可以在subscription()方法内部添加一些影响分配的用户自定义信息赋予userData，
~~~     比如：权重、ip地址、host或者机架（rack）等等。
~~~     再来说一下Assignment类，它是用来表示分配结果信息的，

~~~     # 类中也有两个属性：
~~~     partitions和userData，分别表示所分配到的分区集合和用户自定义的数据。
~~~     可以通过PartitionAssignor接口中的onAssignment()方法是在每个消费者收到消费组leader分配结果时的回调函数，
~~~     例如在StickyAssignor策略中就是通过这个方法保存当前的分配方案，
~~~     以备在下次消费组再平衡（rebalance）时可以提供分配参考依据。

~~~     # 接口中的name()方法用来提供分配策略的名称，
~~~     对于Kafka提供的3种分配策略而言，RangeAssignor对应的protocol_name为“range”，RoundRobinAssignor对应的protocol_name为“roundrobin”，
~~~     StickyAssignor对应的protocol_name为“sticky”，
~~~     所以自定义的分配策略中要注意命名的时候不要与已存在的分配策略发生冲突。
~~~     # 这个命名用来标识分配策略的名称，在后面所描述的加入消费组以及选举消费组leader的时候会有涉及。
~~~     真正的分区分配方案的实现是在assign()方法中，方法中的参数metadata表示集群的元数据信息，
~~~     而subscriptions表示消费组内各个消费者成员的订阅信息，最终方法返回各个消费者的分配信息。

### --- Kafka中还提供了一个抽象类
~~~     org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor，
~~~     它可以简化PartitionAssignor接口的实现，对assign()方法进行了实现，
~~~     其中会将Subscription中的userData信息去掉后，在进行分配。
~~~     Kafka提供的3种分配策略都是继承自这个抽象类。
~~~     如果开发人员在自定义分区分配策略时需要使用userData信息来控制分区分配的结果，
~~~     那么就不能直接继承AbstractPartitionAssignor这个抽象类，而需要直接实现PartitionAssignor接口。

package org.apache.kafka.clients.consumer;

import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

public class MyAssignor extends AbstractPartitionAssignor {

}

### --- 在使用时，消费者客户端需要添加相应的Properties参数，示例如下：

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,

MyAssignor.class.getName());

===============================END===============================

Walter Savage Landor:strove with none,for none was worth my strife.Nature I loved and, next to Nature, Art:I warm'd both hands before the fire of life.It sinks, and I am ready to depart ——W.S.Landor

来自为知笔记(Wiz)

标签：userData,PartitionAssignor,自定义,NO,kafka,分配,Subscription
来源： https://www.cnblogs.com/yanqivip/p/16121904.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

|NO.Z.00059|——————————|BigDataEnd|——|Hadoop&kafka.V44|-------------------------------------------|ka