有网站建设费科目吗,企业展厅设计专业品牌,网页特效制作,深圳微网站一、简介
Kafka 是最初由 Linkedin 公司开发#xff0c;是一个分布式、分区的、多副本的、多订阅者#xff0c;基于 zookeeper 协调的分布式日志系统#xff08;也可以当做 MQ 系统#xff09;#xff0c;常见可以用于 web/nginx 日志、访问日志#xff0c;消息服务等等…一、简介
Kafka 是最初由 Linkedin 公司开发是一个分布式、分区的、多副本的、多订阅者基于 zookeeper 协调的分布式日志系统也可以当做 MQ 系统常见可以用于 web/nginx 日志、访问日志消息服务等等Linkedin于2010年12月贡献给了 Apache基金会 并成为顶级开源项目。
应用特性
分布式存储数据被自动分区并分布在集群的节点中。消息有序性Kafka 能确保从生产者传到消费者的记录都是有序的。高容错性允许集群中节点失败若副本数量为n,则允许n-1个节点失败。高吞吐量Kafka 支持单机每秒至少处理10万以上消息通常可以达到数百万条消息。易扩展性支持集群热扩展。高并发支持数千个客户端同时读写。持久性支持消息数据持久化到本地磁盘 并支持数据备份和灵活配置数据的持久化时间。实时处理/低延迟在数据写入的同时对进行处理消息延迟最低只有几毫秒。
应用场景
Kafka 本质是 支持分布式的消息系统/消息中间件 。分析 Kafka 的应用场景等同于分析 消息中级件 的应用场景。通常使用 消息系统 的 发布/订阅模型 功能来连接 生产者 和 消费者。实现以下三大功能
生产者和消费者的解耦消息持久化 / 消息冗余消息缓冲 / 流量消峰
具体应用场景有
日志收集或数据管道作为日志收集系统或数据处理管道的一部分以处理大量的日志数据或实时数据流。负载均衡如果系统收到大量请求或数据流可以使用消息队列把这些任务平均分配给多个处理器或服务从而实现负载均衡。系统解耦消息队列经常用作不同服务间的通信机制以解耦系统的不同部分。分布式事务如果一个事务需要跨多个服务进行可以使用消息队列来协调不同服务之间的通信确保事务的原子性。实时流数据处理比如实时日志分析或者实时数据报警。Kafka 能接收实时数据流并保证它的可靠性和持久性这样就可以在上游源源不断生产数据的同时下游可以实时地进行分析。通知和实时更新消息队列可以用作通知的中介比如告知用户完成某个任务或者在后端数据更新时实时通知前端。
设计目标
高性能以时间复杂度为O(1)的方式提供消息持久化能力即使对TB级以上数据也能保证常数时间的访问性能。高吞吐率即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。消息系统支持Kafka Server间的消息分区及分布式消费同时保证每个partition内的消息顺序传输。横向扩展支持在线水平热扩展
二、kafka安装和配置
1. zookeeper安装配置
需要说明一下 为了支持 Kafka 的集群功能 Zookeeper 必须使用集群模式部署。 本文以部署 3 个Zookeeper 实例的伪集群为例。具体安装步骤参阅之前的文章Zookeeper 安装教程和使用指南
2. kafka安装配置
下载链接Kafka Downloads 下载页面中包含两种下载方式 : kafka-[version]-src.tgz包含 Kafka 源码和API源码需要自己编译 a) 安装
[rootAli ~]# wget https://downloads.apache.org/kafka/3.6.2/kafka_2.12-3.6.2.tgz
[rootAli ~]# tar xzvf kafka_2.12-3.6.2.tgz
[rootAli ~]# mv /usr/local/kafka_2.12-3.6.2 /usr/local/kafkab) 配置实例
配置第一个 Kafka 实例
# broker 编号集群内必须唯一
broker.id1
# 监听所有ip的9091端口PLAINTEXT表示明文传输
listenersPLAINTEXT://:9091
# 相当于listenersPLAINTEXT://0.0.0.0:9091
# 消息日志存放地址
log.dirs/usr/local/kafka/logs
# ZooKeeper 地址多个用,分隔 /kafka指定在zk上的目录
zookeeper.connectlocalhost:12181/kafka,localhost:22181/kafka配置第二个 Kafka 实例
# broker 编号集群内必须唯一
broker.id1
# 监听所有ip的9092端口PLAINTEXT表示明文传输
listenersPLAINTEXT://:9092
# 消息日志存放地址
log.dirs/opt/kafka/logs
# ZooKeeper 地址多个用,分隔
zookeeper.connectlocalhost:12181/kafka,localhost:22181/kafka注两个客户端的listeners中的port不能一样 4) 服务管理
# 启动服务 -daemon 表示后台启动
$KAFKA_HOME/bin/kafka-server-start.sh -daemon config/server.properties# 查看服务
jps -l43330 org.apache.zookeeper.server.quorum.QuorumPeerMain14356 org.elasticsearch.bootstrap.Elasticsearch14583 org.logstash.Logstash45976 kafka.Kafka # kafka服务进程netstat -anlpt | grep 9091tcp6 0 0 :::9091 :::* LISTEN 45976/javatcp6 0 0 192.168.18.128:9091 192.168.18.128:49356 TIME_WAIT -# 关闭服务
$KAFKA_HOME/bin/kafka-server-stop.sh3. 常用操作
1) 创建topic #两条命令效果一样
bin/kafka-topics.sh --create --bootstrap-server localhost:9091 --partitions 2 --replication-factor 2 --topic yumu
bin/kafka-topics.sh --create --zookeeper localhost:2181/kafka --partitions 2 --replication-factor 2 --topic yumu 在kafka1上创建一个topic会自动同步到其他客户端 --create表示创建操作--zookeeper 指定了 Kafka 连接的 ZooKeeper--partitions 表示每个主题4个分区--replication-factor 表示创建每个分区创建2个副本副本因子--topic 表示主题名称。 注副本因子不能超过存活的broker数量否则报错Replication factor: 20 larger than available brokers: xxx. 2) 查看topic
# 查看topic列表 #两条命令效果一样
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
bin/kafka-topics.sh --list --zookeeper localhost:2181/kafka __consumer_offsetstopic-demoyumu# 查看topic详细信息 #两条命令效果一样
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic yumu
bin/kafka-topics.sh --zookeeper localhost:2181/kafka --describe --topic yumu Topic: yumu PartitionCount: 2 ReplicationFactor: 2 Configs:Topic: yumu Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2Topic: yumu Partition: 1 Leader: 1 Replicas: 2,1 Isr: 1,23) 测试通信
# 窗口1启动生产者向yumu主题发送消息
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic yumu# 窗口2启动消费者订阅yumu主题
bin/kafka-console-consumer.sh --bootstrap-server localhost:9091 --topic yumu# 窗口3启动消费者订阅yumu主题
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic yumu结果
# 生产者
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic yumu
hello, kafka!
once again.# 消费者1
bin/kafka-console-consumer.sh --bootstrap-server localhost:9091 --topic yumu
hello, kafka!
once again.# 消费者2
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic yumu
hello, kafka!
once again.# 查看所有消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic yumu --from-beginning# 删除topic
bin/kafka-topics.sh --delete --bootstrap-server localhost:9091 --topic yumu三、遇到的问题
1. 第一次启动kafka成功后关闭kafka并修改配置再次启动失败报错如下
[2020-11-07 20:43:00,866] INFO Cluster ID MChFWWMBT9GJClVEriND5A (kafka.server.KafkaServer)
[2020-11-07 20:43:00,873] ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.InconsistentClusterIdException: The Cluster ID MChFWWMBT9GJClVEriND5A doesnt match stored clusterId Some(c6QPfvqlS6C3gtsYZptQ8Q) in meta.properties. The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.at kafka.server.KafkaServer.startup(KafkaServer.scala:235)at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)at kafka.Kafka$.main(Kafka.scala:82)at kafka.Kafka.main(Kafka.scala)
[2020-11-07 20:43:00,875] INFO shutting down (kafka.server.KafkaServer)
[2020-11-07 20:43:00,877] INFO [ZooKeeperClient Kafka server] Closing. (kafka.zookeeper.ZooKeeperClient)
[2020-11-07 20:43:00,986] INFO Session: 0x1000da0dde2000c closed (org.apache.zookeeper.ZooKeeper)
[2020-11-07 20:43:00,986] INFO EventThread shut down for session: 0x1000da0dde2000c (org.apache.zookeeper.ClientCnxn)
[2020-11-07 20:43:00,987] INFO [ZooKeeperClient Kafka server] Closed. (kafka.zookeeper.ZooKeeperClient)
[2020-11-07 20:43:00,992] INFO shut down completed (kafka.server.KafkaServer)[2020-11-07 20:43:00,992] ERROR Exiting Kafka. (kafka.server.KafkaServerStartable)
[2020-11-07 20:43:00,993] INFO shutting down (kafka.server.KafkaServer)原因 kafka启动之后会生成一些日志和配置导致这个问题的原因是第一次启动之后生成了log/meta.properties文件 cat meta.properties
#
#Sat Nov 07 21:43:51 CST 2020
broker.id1
version0
cluster.idMChFWWMBT9GJClVEriND5A第二次改完配置后再去启动的时候生成应该会生成一个新的id新的id和旧的ID不一致导致无法启动删除log/meta.properties文件后重新启动即可疑问是不是我关闭的方法不对呢 推荐阅读
Kafka介绍ELK介绍Kafka安装C语言操作kafka以及安装librdkafka库
下一篇Kafka消息系统原理
