kafka-connect 虽然代码处于kafka中且占了很大的份量，但实际是Mq的一个应用了。kafka-connect是一个提供了从其他数据源到kafka的SourceConnector,以及从kafka到其他数据源的SinkConnector的rest服务。下面介绍单机版的基本情况:
一、启动方法
使用命令行bin/connect-standalone.sh config/connect-distributed.properties config/connect-file-source.properties 启动服务，配置参数中有多个文件路径,第一个是服务的配置，后面跟着n个connector的配置。
服务配置示例:
#服务端口
rest.port=8083
#插件搜索路径
plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
#kafka集群地址
bootstrap.servers=localhost:9092,localhost:9093
二、rest接口
kafka-connect提供了许多api操作connector,可以从confluent官网获取详细信息。下面罗列一部分:
1.GET /
获取kafka_cluster_id和版本信息
{
version: "unknown",
commit: "unknown",
kafka_cluster_id: "_MOY-UPhRC-Y-CwV59mm-g"
}
2.GET /connector-plugins
获取服务所有加载的connector。
[
{
class: "org.apache.kafka.connect.file.FileStreamSinkConnector",
type: "sink",
version: "unknown"
},
{
class: "org.apache.kafka.connect.file.FileStreamSourceConnector",
type: "source",
version: "unknown"
}
]
PUT /connector-plugins/{connecttype}/config/validate
4.GET /connectors
获取所有的connector实例，实际上是返回对象对应ClusterConfigState#connectorConfigs 的keyset.
[
"local-file-source"
]
5.GET /connectors/local-file-source
获取某一个connector实例的详细信息
{
"name": "local-file-source",
"config": {
"connector.class": "FileStreamSource",
"file": "test2.txt",
"tasks.max": "1",
"name": "local-file-source",
"topic": "t2t"
},
"tasks": [
{
"connector": "local-file-source",
"task": 0
}
],
"type": "source"
}
6.GET /connectors/{connector}/config
获取connector实例的配置
{
connector.class: "FileStreamSource",
file: "t7.txt",
tasks.max: "1",
name: "local-file-source",
topic: "t2t"
}
7.PUT /connectors/{connector}/config
设置connector实例的配置
请求:
{
"connector.class": "FileStreamSource",
"file": "t7.txt",
"tasks.max": "1",
"name": "local-file-source",
"topic": "t2t"
}
回复:
{
connector.class: "FileStreamSource",
file: "t7.txt",
tasks.max: "1",
name: "local-file-source",
topic: "t2t"
}
8.GET /connectors/{connector}/status
获取connector实例的运行状态
{
name: "local-file-source",
connector: {
state: "RUNNING",
worker_id: "172.16.152.106:8083"
},
tasks: [
{
id: 0,
state: "RUNNING",
worker_id: "172.16.152.106:8083"
}
],
type: "source"
}
3.代码结构图
image.png
4.简单代码分析
1.通过KafkaAdminClient从kafka中获取kafkaClusterId,
2.加载plugins，读取到plugins文件夹的所有connector类(包括source和sink)。注意每个plugin均配置了一个classLoader，这样能保证各个插件的独立性。
3.启动rest服务
RestServer服务内置了一个对象jetty的Server对象，并设置了一系列的handler(用了典型的装饰器模式,用于身份认证，数据统计,业务处理等)，当接受到客户端请求后，按照路径路由到相应的RootResource,ConnectorsResource,ConnectorPluginsResource处理.api中定义了一个Herder接口，单机模式下的实现是StandaloneHerder，分布式版本下是DistributedHerder。路由到各个处理器的请求最后几乎都是通过Herder来完成具体的业务逻辑,herder配置了一个worker，用于管理所有的connector实例，worker内部有一个无疆队列，每当有一个task产生时，启动一个线程来跑。
4.应用的输入参数有n个connector的配置,每个配置将生成一个connector实例，调用了herder的putConnectorConfig方法(和客户端调用接口PUT /connectors/{connector}/config 是一样的效果)。
1.connector实例配置变更接口的内部运行机制
每次调用PUT /connectors/{connector}/config会把原来的connector实例删掉，用新的配置在重新生成一个(这么干确实省事)。同理，如果挂在实例下面的task配置发生了改变(task的配置为ConnectorConfig#originals)，那么还会将原来的task对象全部删掉重新生成。
2.如何控制task的启停
在task任务线程中检测一个标记位即可,暂停的时候进入睡眠即可。
3.单机版和分布式版有什么区别
4.connector实例的配置和其下的task的配置有什么关系
5.假设task poll上来的数据发送失败，如何处理
WorkerSourceTask#toSend用于存储poll上来的消息，每次都是将单次poll上来的数据发送成功了才会再次调用connector 的poll方法。
6.是否会有消息丢失或者消息重复
首先需要知道kafka-connect是如何储存进度的，当准备发送消息时，WorkerSourceTask#outstandingMessages将会存储到这些消息（在成功推给kafka后清除），同时将消息connector传递上来的消息进度(key是个map，value也是个map)存储起来。当执行到后台定时任务，SourceTaskOffsetCommitter#commitExecutorService执行了一个后台任务最后调用FileOffsetBackingStore#set将生产进度刷新到磁盘(单机版默认是本地目录offset.storage.file.filename指定目录，结构是个hashmap,connector可以从SourceTask#context#offsetReader()#offsets()方法获取,每次全量刷新，重启时也全量载入内存，这样保证了历史记录是全的)
，刷新时机是outstandingMessages为空，这样就保证了刷盘的时候，消息是一定发送到kafka了。那么又有一个问题，如果数据源源不断的产生，会不会就不刷新了？这里kafka-connect也做了处理，当后台任务处于等待刷盘的时候，将消息存到WorkerSourceTask#outstandingMessagesBacklog用作备份。所以connector在实现的时候，消息必须有序，否则会产生消息重复。至于丢失，是不会的。那还有没有消息重复的场景呢？有的，如果producer发送成功了，但是刷到磁盘的时候宕机了，那么会有消息重复。
网友评论