0%

kafka自定义组件

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

自定义分区器

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

  • 实现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接口