没有合适的资源?快使用搜索试试~ 我知道了~
资源推荐
资源详情
资源评论




















Spring Boot集群管理工具集群管理工具KafkaAdminClient使用方法解析使用方法解析
主要介绍了Spring Boot集群管理工具KafkaAdminClient使用方法解析,文中通过示例代码介绍的非常详细,对大
家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
原理介绍原理介绍
在Kafka官网中这么描述AdminClient:The AdminClient API supports managing and inspecting topics, brokers, acls, and
other Kafka objects. 具体的KafkaAdminClient包含了一下几种功能(以Kafka1.0.0版本为准):
创建Topic:createTopics(Collection<NewTopic> newTopics)
删除Topic:deleteTopics(Collection<String> topics)
罗列所有Topic:listTopics()
查询Topic:describeTopics(Collection<String> topicNames)
查询集群信息:describeCluster()
查询ACL信息:describeAcls(AclBindingFilter filter)
创建ACL信息:createAcls(Collection<AclBinding> acls)
删除ACL信息:deleteAcls(Collection<AclBindingFilter> filters)
查询配置信息:describeConfigs(Collection<ConfigResource> resources)
修改配置信息:alterConfigs(Map<ConfigResource, Config> configs)
修改副本的日志目录:alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
查询节点的日志目录信息:describeLogDirs(Collection<Integer> brokers)
查询副本的日志目录信息:describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas)
增加分区:createPartitions(Map<String, NewPartitions> newPartitions)
其内部原理是使用Kafka自定义的一套二进制协议来实现,详细可以参见Kafka协议。主要实现步骤:
客户端根据方法的调用创建相应的协议请求,比如创建Topic的createTopics方法,其内部就是发送CreateTopicRequest请
求。
客户端发送请求至Kafka Broker。
Kafka Broker处理相应的请求并回执,比如与CreateTopicRequest对应的是CreateTopicResponse。
客户端接收相应的回执并进行解析处理。
和协议有关的请求和回执的类基本都在org.apache.kafka.common.requests包中,AbstractRequest和AbstractResponse是这
些请求和回执类的两个基本父类。
代码如下
@Component
public class KafkaConfig{
// 配置Kafka
public Properties getProps(){
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
/* props.put("retries", 2); // 重试次数
props.put("batch.size", 16384); // 批量发送大小
props.put("buffer.memory", 33554432); // 缓存大小,根据本机内存大小配置
props.put("linger.ms", 1000); // 发送频率,满足任务一个条件发送*/
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
return props;
}
}
@RestController
public class KafkaTopicManager {
@Autowired
private KafkaConfig kafkaConfig;
@GetMapping("createTopic")
public void createTopic(){
AdminClient adminClient = KafkaAdminClient.create(kafkaConfig.getProps());
NewTopic newTopic = new NewTopic("test1",4, (short) 1);
Collection<NewTopic> newTopicList = new ArrayList<>();
newTopicList.add(newTopic);
adminClient.createTopics(newTopicList);
adminClient.close();
}
资源评论


weixin_38703669
- 粉丝: 8
- 资源: 880
上传资源 快速赚钱
我的内容管理 展开
我的资源 快来上传第一个资源
我的收益
登录查看自己的收益我的积分 登录查看自己的积分
我的C币 登录后查看C币余额
我的收藏
我的下载
下载帮助


安全验证
文档复制为VIP权益,开通VIP直接复制
