kafka入门综合概要介绍

旧城等待， 2024-02-18 16:12 46阅读 0赞

~ kafka

*  设计初衷为统一信息收集平台
    
     *  分布式收集，统一处理
        
         *  可以实时处理反馈信息
         *  支持大数据
         *  高容错
     *  分布式、分区的、多副本的、多订阅者的日志系统（分布式MQ系统）
     *  消息的发布（publish）称作producer，消息的订阅（subscribe）称作consumer，中间的存储阵列称作broker。
     *  ii. 多个broker协同合作，producer、consumer和broker三者之间通过zookeeper来协调请求和转发。
     *  iii. producer产生和推送(push)数据到broker，consumer从broker拉取(pull)数据并进行处理。
     *  iv. broker端不维护数据的消费状态，提升了性能。
     *  v. 直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。
     *  vi. Kafka使用scala编写，可以运行在JVM上。
 *  java编程需要引入相关jar包
 *  详细
    
     *  操作
        
         *  启动
            
             *  启动zookeeper server
                
                 *  bin/zookeeper-server-start.sh config/zookeeper.properties &
             *  启动kafka server
                
                 *  bin/kafka-server-start.sh config/server.properties &
     *  组件
        
         *  zookeeper
            
             *  Topic/logs
                
                 *  分类，一类消息
                 *  Product
                    
                     *  发送者，生产者
                 *  Consumer
                    
                     *  接收者
                 *  partition
                    
                     *  区
                     *  append log文件
                     *  Offset
                        
                         *  消息在partition中的位置
                     *  消息队列
     *  broker
        
         *  kafka集群中的kafka实例
     *  场景
        
         *  Message Queue
 *  设计原理
    
     *  持久化
        
         *  文件存储+缓存批量IO
     *  性能
        
         *  可压缩
     *  Product发送者，生产者
        
         *  负载均衡
         *  异步发送
     *  Consumer
        
         *  接收者
         *  API
            
             *  high-lever
             *  simple
     *  消息传送机制
        
         *  最多一次
         *  最少一次
         *  只有一次
     *  复制备份
        
         *  leader
            
             *  负责所有读写以及监控follow
         *  follow
            
             *  以及与leader同步
         *  所有的follow都保存成功消息才能被消费
            
             *  即使只有一个replicas实例存活仍然可以实现消息的正常发送接收
     *  日志
     *  zookeeper
 *  主要配置
    
     *  Broker
     *  Product
     *  Consumer
     *  Broker部署
        
         *  单机
         *  集群
 *  安装
    
     *  liunx环境
     *  安装jdk
        
         *  vi /etc/profile在末尾追加
            
             *  JAVA\_HOME
             *  JRE\_HOME
             *  PATH=$PATH:$JAVA\_HOME/bin:$JRE\_HOME/bin
             *  CLASSPATH=.:$JAVA\_HOME/lib/dt.jar:$JAVA\_HOME/lib/tools.jar:$JRE\_HOME/lib
         *  source /etc/profile使环境变量立刻生效
         *  ~ jdk安装
     *  安装kafka
        
         *  解压
            
             *  tar -xzvf kafka\_2.8.0-0.8.1.tgz ;rm kafka\_2.8.0-0.8.1.tgz; cd kafka\_2.8.0-0.8.1
         *  修改配置
            
             *  Kafka默认开启JVM压缩指针，但只是在64位的HotSpot VM受支持，如果安装了32位的HotSpot VM，需要修改/bin/kafka-run-class.sh文件
                
                 *  KAFKA\_JVM\_PERFORMANCE\_OPTS="-server -XX:+UseCompressedOops -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true"
                    
                     *  去除-XX:+UseCompressedOops参数
         *  启动
            
             *  bin/zookeeper-server-start.sh config/zookeeper.properties &
             *  bin/kafka-server-start.sh config/server.properties &
         *  停止
            
             *  bin/kafka-server-stop.sh config/server.properties &
             *  bin/zookeeper-server-stop.sh config/zookeeper.properties &
         *  主机topic
            
             *  create a topic
                
                 *  创建单partitions单replication的topic
                    
                     *  bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
                        
                        bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test2
                 *  查看topic
                    
                     *  bin/kafka-topics.sh --list --zookeeper localhost:2181
             *  查看topic的分区
                
                 *  bin/kafka-topics.sh --describe --zookeeper localhost:2181
                 *  bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
         *  启动生产者
            
             *  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
         *  启动消费者
            
             *  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
         *  produce启动的时候参数使用的是kafka的端口而consumer启动的时候使用的是zookeeper的端口；
         *  单机连通性能测试
            
             *  运行producer
                
                 *  bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
             *  运行consumer
                
                 *  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
             *  在producer端输入字符串并回车，查看consumer端是否显示。
         *  分布式连通性测试
 *  配置修改
    
     *  bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my-topic --config max.message.bytes=128000
        
        bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my-topic --deleteConfig max.message.bytes
 *  查看消费者的位置
    
     *  bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zkconnect localhost:2181 --group test
 *  删除topic与数据
    
     *  1) kafka-topics.sh --delete --zookeeper host:port --topic topicname
        
        2) 删除kafka存储目录（server.properties文件log.dirs配置，默认为"/tmp/kafka-logs"）相关topic目录删除zookeeper "/brokers/topics/"目录下相关topic节点
 *  高可用
 *  重发
    
     *  如何确保消息的安全