0%

kafka自定义组件

kafka对于部分配置提供了统一接口,允许用户自定义接口实现,根据业务需求进行定制,如自定义分区器以及序列化反序列化

自定义分区器

kafka的默认分区策略可能不能很傲的满足业务需求,可以需要根据kafka提供的API开发定制满足业务场景的分区策略,自定义一个分区器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public interface Partitioner extends Configurable, Closeable {
// 分区,自定义所传入的分区
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);


public void close();

default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
}
}

public interface Configurable {
//获取配置信息和初始化数据时调用
void configure(Map<String, ?> configs);

}
  • 实现org.apache.kafka.clients.producer.Partitioner接口,重写该接口的partition方法,在该方法中实现分区分配的算法

  • 实例化KafkaProducer的配置中指定partitioner.class为自定义的分区器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
// 自定义分区器
public class CustomPartition implements Partitioner {

private static final Integer PARTITIONS = 6;

@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// 如果key为null,采用默认分区
if (key == null) {
System.out.println("key为null,采用默认分区");
return new DefaultPartitioner().partition(topic, key, keyBytes, value, valueBytes, cluster);
} else {
String code = String.valueOf(key);
try {
int partitionId = code.hashCode() % PARTITIONS;
System.out.println("-----------" + code.hashCode() + "-----------------" + partitionId);
return partitionId;
} catch (Exception e) {
System.out.println("分区出现异常,采用分区0" + e.getMessage());
return 0;
}
}
}

@Override
public void close() {

}

@Override
public void configure(Map<String, ?> configs) {

}
}



public class TestPartition {
/**
* 设置生产消息的总数
*/
private static final int MSG_SIZE = 10;
/**
* 主题名称
*/
private static final String TOPIC_NAME = "test-topic";

private static KafkaProducer<String, String> producer;

static {
Properties properties = initConfig();
producer = new KafkaProducer<String, String>(properties);
}

/**
* 初始化kafka配置
*
* @return
*/
private static Properties initConfig() {
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 配置分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CustomPartition.class.getName());
return properties;
}

public static void main(String[] args) {
ProducerRecord<String, String> record = null;
try {
for (int i = 0; i < MSG_SIZE; i++) {
record = new ProducerRecord<>(TOPIC_NAME,"分区消息key"+i,"分区消息体"+i);
producer.send(record);
Thread.sleep(1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
producer.close();
}
}
}

自定义序列化

需要实现org.apache.kafka.common.serialization.Serializer接口

自定义拦截器

对于 producer 而言,interceptor 使得用户在消息发送前以及 producer 回调逻辑前有机会对消息做一些定制化需求,比如修改消息等。同时,producer 允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。Intercetpor 的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public interface ProducerInterceptor<K, V> extends Configurable {
// 该方法会封装进 KafkaProducer.send 方法中,即它运行在用户主线程中。Producer 确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作,但最好保证不要修改消息所属的 topic 和分区,否则会影响目标分区的计算
public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);

// 该方法会在消息从 RecordAccumulator 成功发送到 Kafka Broker 之后,或者在发送过程中失败时调用。并且通常都是在 producer 回调逻辑触发之前。onAcknowledgement 运行在producer 的 IO 线程中,因此不要在该方法中放入很重的逻辑,否则会拖慢 producer 的消息发送效率
public void onAcknowledgement(RecordMetadata metadata, Exception exception);

//关闭 interceptor,主要用于执行一些资源清理工作
public void close();
}

public interface Configurable {

//获取配置信息和初始化数据时调用
void configure(Map<String, ?> configs);

}

执行顺序

拦截器->序列化器->分区器