1 confluent介绍
Confluent是用来管理和组织不同数据源的流媒体平台，可以实时地把不同源和位置的数据集成到一个中心的事件流平台。并且很可靠、性能很高。
Confluent目前提供了社区版（免费）和商业版（收费）两个版本，社区版提供了Connectors、REST Proxy、KSQL、Schema-Registry等基础服务。商业版为企业提供了控制面板、负载均衡，跨中心数据备份、安全防护等高级特性。
1.2 服务功能介绍
1.2.1 Zookeeper
Zookeeper是一个开放源码的分布式应用程序协调服务，主要功能包扩：维护配置信息、命名、提供分布式同步、组管理等集中式服务 。Kafka使用ZooKeeper对集群元数据进行持久化存储，如果ZooKeeper丢失了Kafka数据，集群的副本映射关系以及topic等配置信息都会丢失，最终导致Kafka集群不再正常工作，造成数据丢失的后果。
1.2.2 Kafka
Kafka是一个分布式流处理平台，基于zookeeper协调并支持分区和多副本的分布式消息系统，是一种高吞吐量的分布式发布订阅消息系统，消息队列中间件，主要功能是负责消息传输，Confluent就是依赖Kafka来进行消息传输。Kafka最大的特性就是可以实时的处理大量数据以满足各种需求场景。
1.2.3 Control Center
control center可以很容易地管理kafka的连接，创建，编辑，和管理与其他系统的连接。我们可以从producer到consumer监控data streams，保证我们的每一条消息都被传递，还能测量出消息的传输耗时多久。使用confluent control center能让开发人员不写一句代码，也能构建基于kafka的数据生产管道。
1.2.4 Kafka-rest
Kafka-rest是Kafka RESTful接口服务组件，可以通过Restful接口而不是本机Kafka协议或客户端的情况下，生成和使用消息,而且还可以查看集群状态以及执行管理操作。
1.2.5 Schema-Registry
Schema-Registry是为元数据管理提供的服务，同样提供了RESTful接口用来存储和获取schemas，它能够保存数据格式变化的所有版本，并可以做到向下兼容。Schema-Registry还为Kafka提供了Avro格式的序列化插件来传输消息。Confluent主要用Schema-Registry来对数据schema进行管理和序列化操作。
1.2.6 Connect
Kafka Connect是 Kafka的一个开源组件，是用来将Kafka与数据库、key-value存储系统、搜索系统、文件系统等外部系统连接起来的基础框架。通过使用Kafka Connect框架以及现有的连接器可以实现从源数据读入消息到Kafka，再从Kafka读出消息到目的地的功能。
1.2.7 ksql-server
KSQL是使用SQL语句对Apache Kafka执行流处理任务的流式SQL引擎，Confluent 使用KSQL对Kafka的数据提供查询服务.
2 confluent下载
使用的开源的confluent的5.2.4版本
下载链接：http://packages.confluent.io/archive/5.2/confluent-5.2.4-2.11.tar.gz
3 环境准备
分布式搭建建议至少3个节点，但是由于用于测试及节点紧张这里使用2个节点
| 节点 | zookeeper | kafka | control-center | kafka-reset | schema-registry | connector | ksql-server |
|---|---|---|---|---|---|---|---|
| 10.0.165.8 | √ | √ | √ | √ | √ | √ | √ |
| 10.0.165.9 | √ | √ | √ | √ | √ | √ | √ |
| 2181 | 9092 | 9021 | 8082 | 8081 | 8083 | 8088 |
4 安装
4.1 解压
将下载的文件上传至linux，然后解压至相应的目录下
tar -zxvf /opt/package/confluent-5.2.4-2.11.tar.gz -C /home/kafka/.local/
修改文件名并进入到相应的目录下
mv /home/kafka/.local/confluent-5.2.4 /home/kafka/.local/confluent
cd /home/kafka/.local/confluent
4.2 修改配置
修改10.0.165.8节点的相应配置
4.2.1 zookeeper配置
(1)vim /home/kafka/.local/confluent/etc/kafka/zookeeper.properties
##数据存放目录，默认为/tmp/zookeepe存在删除风险
dataDir=/data/confluent/zookeeper
clientPort=2181
maxClientCnxns=0
initLimit=5
syncLimit=2
##多个zookeeper server，server的编号1、2等要与myid中的一致
server.1=10.0.165.8:2888:3888
server.2=10.0.165.9:2888:3888
(2)生成myid
echo 1 > /home/kafka/.local/confluent/etc/kafka/myid
(3)修改confluent服务启动脚本，将myid发布到confluent运行目录下。
bin/confluent start会启动confluent的各服务，且会将etc下的各配置，复制到confluent运行目录下。
vim /home/kafka/.local/confluent/bin/confluent
在config_zookeeper()方法块最后一行，添加
cp ${confluent_conf}/kafka/myid $confluent_current/zookeeper/data/
目的是将etc/kafka/myid拷贝到confluent运行目录下，否则会报myid is no found，zookeeper启动失败。
4.2.2 Kafka配置
vim /home/kafka/.local/confluent/etc/kafka/server.properties
broker.id=0
#listeners与advertised.listeners可以只配一个，与当前机器网卡有关系，请注意。advertised.listeners可能通用性更强，值为当前机器的ip与端口，其他机器ip无需配置
advertised.listeners=PLAINTEXT://10.0.165.8:9092
##根据实际情况调整
num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100
#log.dirs是最重要的配置，kafka数据所在
log.dirs=/data/confluent/kafka-logs
num.partitions=12
num.recovery.threads.per.data.dir=1
message.max.bytes=10000000
replica.fetch.max.bytes= 10485760
auto.create.topics.enable=true
auto.leader.rebalance.enable = true
##备份因子数<=kafka节点数，若大于会报错
default.replication.factor=2
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=20000
log.flush.interval.ms=10000
log.flush.scheduler.interval.ms=2000
log.retention.check.interval.ms=300000
log.cleaner.enable=true
##log失效时间，单位小时
log.retention.hours=48
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000
confluent.metrics.reporter.bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
confluent.metrics.reporter.topic.replicas=2
confluent.support.metrics.enable=true
confluent.support.customer.id=anonymous
delete.topic.enable=true
group.initial.rebalance.delay.ms=0
4.2.3 kafka-rest
vim /home/kafka/.local/confluent/etc/kafka-rest/kafka-rest.properties
id=kafka-rest-server-001
schema.registry.url=http://10.0.165.8:8081
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
bootstrap.servers=PLAINTEXT://10.0.165.8:9092
port=8082
consumer.threads=8
access.control.allow.methods=GET,POST,PUT,DELETE,OPTIONS
access.control.allow.origin=*
4.2.4 ksql
confluent-4没有这个
vim /home/kafka/.local/confluent/etc/ksql/ksql-server.properties
ksql.service.id=default_
bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
listeners=http://0.0.0.0:8088
ksql.schema.registry.url=http://10.0.165.8:8081,http://10.0.165.9:8081
ksql.sink.partitions=4
4.2.5 confluent-control-center
vim /home/kafka/.local/confluent/etc/confluent-control-center/control-center-dev.properties
bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
confluent.controlcenter.rest.listeners=http://0.0.0.0:9021
#每个id要唯一，不然只能启动一个
confluent.controlcenter.id=1
confluent.controlcenter.data.dir=/data/confluent/control-center
confluent.controlcenter.connect.cluster=http://10.0.165.8:8083,http://10.0.165.9:8083
##每台都配置各自的ip
confluent.controlcenter.ksql.url=http://10.0.165.8:8088
confluent.controlcenter.schema.registry.url=http:/10.0.165.8:8081,http://10.0.165.9:8081
confluent.controlcenter.internal.topics.replication=2
confluent.controlcenter.internal.topics.partitions=2
confluent.controlcenter.command.topic.replication=2
confluent.monitoring.interceptor.topic.partitions=2
confluent.monitoring.interceptor.topic.replication=2
confluent.metrics.topic.replication=2
confluent.controlcenter.streams.num.stream.threads=30
4.2.6 schema-registry
vim /home/kafka/.local/confluent/etc/schema-registry/schema-registry.properties
listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=PLAINTEXT://10.0.165.8:9092,10.0.165.9:9092
kafkastore.topic=_schemas
debug=false
4.2.7 connect
vim /home/kafka/.local/confluent/etc/schema-registry/connect-avro-distributed.properties
bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statuses
config.storage.replication.factor=2
offset.storage.replication.factor=2
status.storage.replication.factor=2
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
rest.port=8083
rest.advertised.port=8083
plugin.path=/home/kafka/.local/confluent/share/java
4.2.8 将confluent发送到其他节点
scp -r confluent/ kafka@10.0.165.9:/home/kafka/.local/
然后修改其他节点的配置
vi myid
2
vi /home/kafka/.local/confluent/etc/kafka/server.properties
broker.id=1
advertised.listeners=PLAINTEXT://10.0.165.9:9092
vi /home/kafka/.local/confluent/etc/kafka-rest/kafka-rest.properties
id=kafka-rest-server-002
schema.registry.url=http://10.0.165.9:8081
bootstrap.servers=PLAINTEXT://10.0.165.9:9092
vi /home/kafka/.local/confluent/etc/confluent-control-center/control-center-dev.properties
confluent.controlcenter.id=2
confluent.controlcenter.ksql.url=http://10.0.165.9:8088
然后在两个节点的/data目录下新建confluent并修改权限
sudo mkdir /data/confluent
sudo chown kafka:kafka /data/confluent
4.3 服务启动与停止
4.3.1 全部服务启动
启动：bin/confluent start
查看状态：bin/confluent status
停止：bin/confluent stop
4.3.2 单独启动服务
服务单独启动
启动kafka-rest
bin/kafka-rest-start etc/kafka-rest/kafka-rest.properties
上面的这种方式是前台启动，也可以以后台方式启动。
nohup bin/kafka-rest-start etc/kafka-rest/kafka-rest.properties &
启动zookeeper
bin/zookeeper-server-start -daemon etc/kafka/zookeeper.properties
启动kafka broker
bin/kafka-server-start -daemon etc/kafka/server.properties
启动schema registry
bin/schema-registry-start -daemon etc/schema-registry/schema-registry.properties
5 安装过程常见报错
5.1 KafkaServer启动失败
[2020-06-27 04:28:15,713] FATAL [KafkaServer id=2] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.KafkaException: Socket server failed to bind to 10.0.165.8:9092: Cannot assign requested address.
at kafka.network.Acceptor.openServerSocket(SocketServer.scala:331)
at kafka.network.Acceptor.<init>(SocketServer.scala:256)
at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:97)
at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:89)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at kafka.network.SocketServer.startup(SocketServer.scala:89)
at kafka.server.KafkaServer.startup(KafkaServer.scala:229)
at io.confluent.support.metrics.SupportedServerStartable.startup(SupportedServerStartable.java:112)
at io.confluent.support.metrics.SupportedKafka.main(SupportedKafka.java:58)
Caused by: java.net.BindException: Cannot assign requested address
at sun.nio.ch.Net.bind0(Native Method)
at sun.nio.ch.Net.bind(Net.java:433)
at sun.nio.ch.Net.bind(Net.java:425)
at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67)
at kafka.network.Acceptor.openServerSocket(SocketServer.scala:327)
... 9 more
[2020-06-27 04:28:15,715] INFO [KafkaServer id=2] shutting down (kafka.server.KafkaServer)
[2020-06-27 04:28:15,717] INFO [SocketServer brokerId=2] Shutting down (kafka.network.SocketServer)
[2020-06-27 04:28:15,718] INFO [SocketServer brokerId=2] Shutdown completed (kafka.network.SocketServer)
[2020-06-27 04:28:15,721] INFO Shutting down. (kafka.log.LogManager)
[2020-06-27 04:28:15,760] INFO Shutdown complete. (kafka.log.LogManager)
[2020-06-27 04:28:15,761] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2020-06-27 04:28:15,762] INFO Session: 0x27297ff0225a5a9 closed (org.apache.zookeeper.ZooKeeper)
[2020-06-27 04:28:15,764] INFO EventThread shut down for session: 0x27297ff0225a5a9 (org.apache.zookeeper.ClientCnxn)
[2020-06-27 04:28:15,765] INFO [KafkaServer id=2] shut down completed (kafka.server.KafkaServer)
[2020-06-27 04:28:15,766] INFO [KafkaServer id=2] shutting down (kafka.server.KafkaServer)
自己copy了server.properties文件到各个节点没有修改下面的配置 监听器的配置，应该指向节点本身的主机名和端口，我全部四台机器都指向了10.0.165.8，所以导致了只有节点1是正常的
advertised.listeners=PLAINTEXT://10.0.165.9:9092
5.2 Confluent schema-registry启动失败
[2020-06-27 16:09:39,872] WARN The replication factor of the schema topic _schemas is less than the desired one of 3. If this is a production environment, it's crucial to add more brokers and increase the replication factor of the topic. (io.confluent.kafka.schemaregistry.storage.KafkaStore:242)
[2020-06-27 16:09:50,095] ERROR Server died unexpectedly: (io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain:51)
java.lang.IllegalArgumentException: Unable to subscribe to the Kafka topic _schemas backing this data store. Topic may not exist.
at io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread.<init>(KafkaStoreReaderThread.java:125)
at io.confluent.kafka.schemaregistry.storage.KafkaStore.init(KafkaStore.java:130)
at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.init(KafkaSchemaRegistry.java:199)
at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.setupResources(SchemaRegistryRestApplication.java:64)
at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.setupResources(SchemaRegistryRestApplication.java:42)
at io.confluent.rest.Application.createServer(Application.java:157)
at io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain.main(SchemaRegistryMain.java:43)
因为kafkaserver没有启动
6 常用操作
（1）启动
confluent start
（2）查看日志文件目录
confluent current
（3）列出连接
confluent list connectors
（4）查看加载的连接器
confluent status connectors
[
"file-source"
]
（5）查看具体连接器状态
confluent status file-source