Shikong/springcloud-study
1
0
Fork 0
mirror of https://gitee.com/shikong-sk/springcloud-study synced 2024-11-09 01:51:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

调整

Browse Source
This commit is contained in:
Shikong 2023-04-25 14:59:34 +08:00
parent c2f6bac38d
commit e4ebff30b9
1 changed files with 25 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
kafka-01/kafka-01-producer/src/test/java/cn/skcks/study/springcloud/kafka/tests/KafkaConfigTest.java
View File

@ -5,6 +5,7 @@ import cn.skcks.study.springcloud.utils.JsonUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@ -22,13 +23,13 @@ public class KafkaConfigTest {
@Autowired
private KafkaProperties kafkaProperties;
private Map<String,Object> adminProperties;
private Map<String,Object> producerProperties;
private Map<String,Object> consumerProperties;
private Map<String,Object> streamsProperties;
private Map<String, Object> adminProperties;
private Map<String, Object> producerProperties;
private Map<String, Object> consumerProperties;
private Map<String, Object> streamsProperties;
@PostConstruct
void init(){
void init() {
adminProperties = kafkaProperties.buildAdminProperties();
producerProperties = kafkaProperties.buildProducerProperties();
consumerProperties = kafkaProperties.buildConsumerProperties();
@ -36,20 +37,33 @@ public class KafkaConfigTest {
}
@Test
void properties () {
void properties() {
log.info("AdminProperties {}", JsonUtils.toJson(adminProperties));
log.info("ProducerProperties {}",JsonUtils.toJson(producerProperties));
log.info("ProducerProperties {}", JsonUtils.toJson(producerProperties));
log.info("ConsumerProperties {}", JsonUtils.toJson(consumerProperties));
log.info("StreamsProperties {}", JsonUtils.toJson(streamsProperties));
}
@Test
@SneakyThrows
void topics(){
void topics() {
try (AdminClient adminClient = AdminClient.create(kafkaProperties.buildAdminProperties())) {
ListTopicsResult topicsResult = adminClient.listTopics();
topicsResult.listings().get().forEach(topic -> {
log.info("topic uuid={} name={}",topic.topicId(),topic.name());
ListTopicsOptions options = new ListTopicsOptions();
// 包含内置主题
options.listInternal(true);
ListTopicsResult topicsResult = adminClient.listTopics(options);
topicsResult.listings().get().stream().sorted((a,b) -> {
if(a.isInternal() && b.isInternal()){
return 0;
}
if (a.isInternal()){
return -1;
} else {
return 1;
}
}).forEach(topic -> {
log.info("topic isInternal=>{}, uuid=>{}, name=>{}", topic.isInternal(), topic.topicId(), topic.name());
});
}
}