kafka的安装以及基本操作

秒速五厘米 2023-07-09 02:17 29阅读 0赞

**一 kafka的安装**  
**1 名词解释：**

Topic(主题) 消息归类，给消息打上标记  
Producer(生产者) 消息的产生，发送  
Consumer(消费者) 接收使用消息者  
Consumer Group 每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）  
broker(kafak实例 ) 每个kafka节点就是broker  
Partition（分区 Partition是物理上的概念，每个Topic包含一个或多个Partition 是为提高kafka的吞吐量  
replica partition 的副本，保障 partition 的高可用  
leader replica 中的一个角色， producer 和 consumer 只跟 leader 交互  
follower replica 中的一个角色，从 leader 中复制数据  
controller Kafka 集群中的其中一个服务器，用来进行 leader election 以及各种 failover  
**2 kafka的两种模式**  
（1）点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除） 拉取模式

点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。

（2）发布/订阅模式（一对多，数据生产后，推送给所有订阅者） 也就是推送模式

发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。

**3 kafka安装**  
1.下载

Apache kafka 官方：http://kafka.apache.org/downloads.html

Scala 2.11 - kafka\_2.11-0.10.2.0.tgz (asc, md5)

注：当Scala用的是2.11那Kafka选择kafka\_2.11-0.10.2. 0.10.2才是Kafka的版本

Kafka集群安装:

1.  .安装JDK &配置JAVA\_HOME
2.  安装Zookeeper
    
    参照Zookeeper官网搭建一个ZK集群， 并启动ZK集群。
3.  解压Kafka安装包

[ambow@hadoopNode1 ambow]$ tar  -zxvf  kafka_2.11-0.10.2.1.tgz   -C  ~/app/

4.配置环境变量

export KAFKA_HOME=/home/ambow/app/kafka_2.11-0.10.2.1
    
    export PATH=$PATH:$KAFKA_HOME/bin

1.  修改配置文件config/server.properties  
    vi server.properties

#为依次增长的：0、1、2、3、4，集群中节点唯一id 
        broker.id=0    
         						
        #删除主题的配置，默认是false   生产环境设为false
        delete.topic.enable=true
        
        #监听的主机及端口号    各节点改为本机相应的hostName
        listeners=PLAINTEXT://hadoopNode1:9092   
        
        
        #Kafka的消息数据存储路径  
        log.dirs=/home/ambow/kafkaData/logs   
        
        #创建主题的时候，默认有1个分区
        num.partitions=3  
        
        #指定ZooKeeper集群列表，各节点以逗号分
        zookeeper.connect=hadoopNode1:2181,hadoopNode2:2181,hadoopNode3:2181,hadoopNode4:2181,hadoopNode5:2181

6.分发至各个节点

[ambow@hadoopNode1 app]$ scp -r   kafka_2.11-0.10.2.1     ambow@hadoopNode5:~/app
    [ambow@hadoopNode1 app]$ scp -r   kafka_2.11-0.10.2.1     ambow@hadoopNode4:~/app
    [ambow@hadoopNode1 app]$ scp -r   kafka_2.11-0.10.2.1     ambow@hadoopNode3:~/app
    [ambow@hadoopNode1 app]$ scp -r   kafka_2.11-0.10.2.1     ambow@hadoopNode2:~/app
    
    
    
    [ambow@hadoopNode1 app]$ scp -r   ~/.bash_profile     ambow@hadoopNode5:~
    [ambow@hadoopNode1 app]$ scp -r   ~/.bash_profile     ambow@hadoopNode4:~
    [ambow@hadoopNode1 app]$ scp -r   ~/.bash_profile     ambow@hadoopNode3:~
    [ambow@hadoopNode1 app]$ scp -r   ~/.bash_profile     ambow@hadoopNode2:~
    
    
    source   ~/.bash_profile

7.修改各个节点的配置文件：

#为依次增长的：0、1、2、3、4，集群中节点唯一id 
    broker.id=0  
    
    # 监听的主机及端口号    各节点改为本机相应的hostName
    listeners=PLAINTEXT://hadoopNode1:9092

8.各台节点上启动Kafka服务

[ambow@hadoopNode1 app]$ kafka-server-start.sh  $KAFKA_HOME/config/server.properties  &

注：要先启动各节点Zookeeper

zkServlet.sh start

[ambow@hadoopNode1 app]$ kafka-server-stop.sh

9.测试:

1)创建Topic名称为： test的主题

#语法:
    kafka-topics.sh --create --zookeeper 指定zookeeper集群节点用逗号分隔 --replication-factor 指定副本数  --partitions 指定分区数 --topic 指定主题名      
    
    [ambow@hadoopNode1 app]$ kafka-topics.sh --create --zookeeper hadoopNode1:2181,hadoopNode2:2181,hadoopNode3:2181,hadoopNode4:2181,hadoopNode5:2181 --replication-factor 3 --partitions 1 --topic test    
    
    
    
    
     kafka-topics.sh --create --zookeeper HadoopNode1:2181,HadoopNode2:2181,HadoopNode3:2181 --replication-factor 3 --partitions 1 --topic love    
     
     
      kafka-topics.sh --create --zookeeper hpNode1:2181 --replication-factor 2 --partitions 3 --topic it

注：replication-factor副本数必须小于等于节点数

如果只有一个副本，就会存在某一个节点上
    
     果设置两个副本，就会存在某两个节点上

kafka默认把消费存储7天

#删除指定的test主题
    [ambow@hadoopNode1 app]$ kafka-topics.sh --delete --zookeeper hadoopNode1:2181,hadoopNode2:2181,hadoopNode3:2181,hadoopNode4:2181,hadoopNode5:2181   --topic test

2)列出已创建的topic列表

#查看所有的主题:
    [ambow@hadoopNode1 app]$ kafka-topics.sh --list --zookeeper localhost:2181
    
    [ambow@hadoopNode1 app]$ kafka-topics.sh --list --zookeeper localhost:2181

3)查看Topic的详细信息

#查看指定主题的详细信息
    [ambow@hadoopNode1 app]$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

4)模拟生产者发送消息 向指定的主题发布消息

开启另一个终端去执行:

[ambow@hadoopNode1 app]$ kafka-console-producer.sh --broker-list hadoopNode1:9092,hadoopNode2:9092,hadoopNode3:9092,hadoopNode4:9092,hadoopNode5:9092 --topic test
    
    
    
     kafka-console-producer.sh --broker-list HadoopNode1:9092,HadoopNode2:9092,HadoopNode3:9092 --topic test

注：\[ambow@hadoopNode1 app\]$ kafka-console-producer.sh --broker-list (节点可以是一个，一般本配多个防止节点宕掉) --topic test

5）消费者:

模拟客户端去接受消息 消费指定的主题的消息

开启另一个终端去执行:

[ambow@hadoopNode1 app]$ kafka-console-consumer.sh --bootstrap-server hadoopNode1:9092,hadoopNode2:9092,hadoopNode3:9092,hadoopNode4:9092,hadoopNode5:9092 --from-beginning --topic test
    
    #定阅
    kafka-console-consumer.sh --bootstrap-server HadoopNode1:9092,HadoopNode2:9092,HadoopNode3:9092 --from-beginning --topic test