kafka单机测试环境搭建

柔光的暖阳◎ 2024-04-17 21:54 85阅读 0赞

kafka单服务器搭建

kafka搭建需要jdk， zookeeper

因此搭建前需先准备好JDK环境  
由于kafka的安装包中已经内置一个zk，因此可以不用另外搭建， 当然如果想独立处理，或者已经拥有zk也可以接入使用

下载地址：  
`http://kafka.apache.org/downloads`

##### 1.下载 #####

下载带有scala的安装包

Binary downloads:
    Scala 2.11  - kafka_2.11-2.3.0.tgz (asc, sha512)
    Scala 2.12  - kafka_2.12-2.3.0.tgz (asc, sha512)

##### 2.安装JDK #####

mkdir /app
    
    tar -zxvf jdk-8u221-linux-x64.tar.gz

添加环境变量  
vi ~/.bash\_profile

export JAVA_HOME=/app/jdk1.8.0_221
    
    export JRE_HOME=${JAVA_HOME}/jre
    
    export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
    
    export PATH=.:${JAVA_HOME}/bin:$ZOOKEEPER_HOME/bin:$PATH

加载环境变量  
source ~/.bash\_profile

验证java环境

# java --version
    Unrecognized option: --version
    Error: Could not create the Java Virtual Machine.
    Error: A fatal exception has occurred. Program will exit.
    [root@kafka kafka_2.11-2.2.0]#

##### 3.解压安装 #####

tar -zxf kafka_2.11-2.2.0.tgz 
     
     cd kafka_2.11-2.2.0/

##### 4.启动zk #####

bin/zookeeper-server-start.sh config/zookeeper.properties

[root@kafka kafka_2.11-2.2.0]# bin/zookeeper-server-start.sh config/zookeeper.properties 
    [2019-09-01 15:42:13,338] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
    [2019-09-01 15:42:13,344] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
    [2019-09-01 15:42:13,344] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
    [2019-09-01 15:42:13,344] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
    [2019-09-01 15:42:13,344] WARN Either no config or no quorum defined in config, running  in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
    [2019-09-01 15:42:13,361] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
    [2019-09-01 15:42:13,361] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
     
     ....
    
    [2019-09-01 15:42:33,437] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
    [2019-09-01 15:42:33,444] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

##### 5.启动kafka #####

bin/kafka-server-start.sh config/server.propertie

[2019-09-01 15:58:04,457] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
    [2019-09-01 15:58:04,501] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
    [2019-09-01 15:58:04,504] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
    [2019-09-01 15:58:04,621] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
    [2019-09-01 15:58:04,631] INFO [SocketServer brokerId=0] Started data-plane processors for 1 acceptors (kafka.network.SocketServer)
    [2019-09-01 15:58:04,634] INFO Kafka version: 2.2.0 (org.apache.kafka.common.utils.AppInfoParser)
    [2019-09-01 15:58:04,635] INFO Kafka commitId: 05fcfde8f69b0349 (org.apache.kafka.common.utils.AppInfoParser)
    [2019-09-01 15:58:04,636] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

注意需要添加housts解析， 否则会报错：

[root@kafka config]# cat /etc/hosts
    127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    ::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
    
    192.168.137.51  kafka

错误信息

[2019-09-01 15:53:43,428] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
    [2019-09-01 15:54:03,445] ERROR [KafkaServer id=0] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
    java.net.UnknownHostException: kafka: kafka: Name or service not known
            at java.net.InetAddress.getLocalHost(InetAddress.java:1506)
            at kafka.server.KafkaServer$$anonfun$5.apply(KafkaServer.scala:410)
            at kafka.server.KafkaServer$$anonfun$5.apply(KafkaServer.scala:408)
            at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
            at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
            at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
            at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
            at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
            at scala.collection.AbstractTraversable.map(Traversable.scala:104)
            at kafka.server.KafkaServer.createBrokerInfo(KafkaServer.scala:408)
            at kafka.server.KafkaServer.startup(KafkaServer.scala:259)
            at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:38)
            at kafka.Kafka$.main(Kafka.scala:75)
            at kafka.Kafka.main(Kafka.scala)
    Caused by: java.net.UnknownHostException: kafka: Name or service not known
            at java.net.Inet6AddressImpl.lookupAllHostAddr(Native Method)
            at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:929)
            at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1324)
            at java.net.InetAddress.getLocalHost(InetAddress.java:1501)

##### 测试安装环境 #####

使用安装包中已经包含的命令行工具  
TOPIC  
可以查看下帮助

[root@kafka kafka_2.11-2.2.0]# bin/kafka-topics.sh --help
    This tool helps to create, delete, describe, or change a topic.
    Option                                   Description                            
    ------                                   -----------                            
    --alter                                  Alter the number of partitions,        
                                               replica assignment, and/or           
                                               configuration for the topic.         
    --bootstrap-server <String: server to    REQUIRED: The Kafka server to connect  
      connect to>                              to. In case of providing this, a     
                                               direct Zookeeper connection won't be 
                                               required.                            
    --command-config <String: command        Property file containing configs to be 
      config property file>                    passed to Admin Client. This is used 
                                               only with --bootstrap-server option  
                                               for describing and altering broker   
                                               configs.                             
    --config <String: name=value>            A topic configuration override for the 
                                               topic being created or altered.The   
                                               following is a list of valid         
                                               configurations:                      
                                                    cleanup.policy                        
                                                    compression.type                      
                                                    delete.retention.ms                   
                                                    file.delete.delay.ms                  
                                                    flush.messages                        
                                                    flush.ms                              
                                                    follower.replication.throttled.       
                                               replicas                             
                                                    index.interval.bytes                  
                                                    leader.replication.throttled.replicas 
                                                    max.message.bytes                     
                                                    message.downconversion.enable         
                                                    message.format.version                
                                                    message.timestamp.difference.max.ms   
                                                    message.timestamp.type                
                                                    min.cleanable.dirty.ratio             
                                                    min.compaction.lag.ms                 
                                                    min.insync.replicas                   
                                                    preallocate                           
                                                    retention.bytes                       
                                                    retention.ms                          
                                                    segment.bytes                         
                                                    segment.index.bytes                   
                                                    segment.jitter.ms                     
                                                    segment.ms                            
                                                    unclean.leader.election.enable        
                                             See the Kafka documentation for full   
                                               details on the topic configs.It is   
                                               supported only in combination with --
                                               create if --bootstrap-server option  
                                               is used.                             
    --create                                 Create a new topic.                    
    --delete                                 Delete a topic                         
    --delete-config <String: name>           A topic configuration override to be   
                                               removed for an existing topic (see   
                                               the list of configurations under the 
                                               --config option). Not supported with 
                                               the --bootstrap-server option.       
    --describe                               List details for the given topics.     
    --disable-rack-aware                     Disable rack aware replica assignment  
    --exclude-internal                       exclude internal topics when running   
                                               list or describe command. The        
                                               internal topics will be listed by    
                                               default                              
    --force                                  Suppress console prompts               
    --help                                   Print usage information.               
    --if-exists                              if set when altering or deleting or    
                                               describing topics, the action will   
                                               only execute if the topic exists.    
                                               Not supported with the --bootstrap-  
                                               server option.                       
    --if-not-exists                          if set when creating topics, the       
                                               action will only execute if the      
                                               topic does not already exist. Not    
                                               supported with the --bootstrap-      
                                               server option.                       
    --list                                   List all available topics.             
    --partitions <Integer: # of partitions>  The number of partitions for the topic 
                                               being created or altered (WARNING:   
                                               If partitions are increased for a    
                                               topic that has a key, the partition  
                                               logic or ordering of the messages    
                                               will be affected                     
    --replica-assignment <String:            A list of manual partition-to-broker   
      broker_id_for_part1_replica1 :           assignments for the topic being      
      broker_id_for_part1_replica2 ,           created or altered.                  
      broker_id_for_part2_replica1 :                                                
      broker_id_for_part2_replica2 , ...>                                           
    --replication-factor <Integer:           The replication factor for each        
      replication factor>                      partition in the topic being created.
    --topic <String: topic>                  The topic to create, alter, describe   
                                               or delete. It also accepts a regular 
                                               expression, except for --create      
                                               option. Put topic name in double     
                                               quotes and use the '\' prefix to     
                                               escape regular expression symbols; e.
                                               g. "test\.topic".                    
    --topics-with-overrides                  if set when describing topics, only    
                                               show topics that have overridden     
                                               configs                              
    --unavailable-partitions                 if set when describing topics, only    
                                               show partitions whose leader is not  
                                               available                            
    --under-replicated-partitions            if set when describing topics, only    
                                               show under replicated partitions     
    --zookeeper <String: hosts>              DEPRECATED, The connection string for  
                                               the zookeeper connection in the form 
                                               host:port. Multiple hosts can be     
                                               given to allow fail-over.

##### 创建topic #####

[root@kafka kafka_2.11-2.2.0]# bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 
    --topic mtest --partitions 1 --replication-factor 1
    
    Created topic mtest.

##### 查看TOPIC： #####

[root@kafka kafka_2.11-2.2.0]# bin/kafka-topics.sh --describe --zookeeper 127.0.0.1:2181 --topic mtest
    Topic:mtest     PartitionCount:1        ReplicationFactor:1     Configs:
            Topic: mtest    Partition: 0    Leader: 0       Replicas: 0     Isr: 0
    [root@kafka kafka_2.11-2.2.0]#

发送消息：

##### 启动生产者： #####

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mtest

打开另外一个窗口

##### 启动消费者： #####

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mtest --from-beginning

测试消息发送

>
    >
    >ok
    >this is test
    >12311
    >2424\
    >
    >
    >

消费者接收消息：

ok
    this is test
    12311
    2424\

安装完成