架构►mq►kafka

spring kafka

连接 Kafka

1
2
3
4
5
6
7
// 连接 kafka
@Bean
public KafkaAdmin admin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    return new KafkaAdmin(configs);
}

管理 Topic

1
2
3
4
5
6
7
8
@Bean
public NewTopic divinerInternalStrategyOperation() {
    return TopicBuilder.name("example-topic")
            .partitions(1)
            .replicas(1)
            .compact()
            .build();
}

订阅 topic

1
2
3
4
@KafkaListener(id = "consumer-group", topics = "diviner_internal_strategy_operation")
public void listen(String in) {
    // ...
}

发送消息

1
2
3
4
@Autowired
private KafkaTemplate<Object, Object> template;
 
template.send(key, value);

参考