Kafka的安装与使用

男娘i 2022-01-21 08:13 302阅读 0赞

### 文章目录 ###

*  资源链接
 *  软件安装
 *   *  配置环境变量
     *  配置ZooKeeper
     *  配置Kafka
 *  源码
 *  使用Java连接Kafka
 *   *  启动Kafka相关的服务器
     *  运行结果

# 资源链接 #

官网：

*  [Kafka][]

Kafka的官方下载地址：

*  [Kafka Download][]

Kafka的官方文档地址：

*  [Kafka Documentation][]

# 软件安装 #

这里所安装的Kafka版本为`kafka_2.11-2.2.0`。在此，我需要说明的是，由于Kafka本身需要ZooKeeper提供相应的支持，因而在安装Kafka时一定要确保优先安装ZooKeeper。

不过在`kafka_2.11-2.2.0`版本的Kafka中，由于有一个内置的ZooKeeper，因而我们就不必再去下载相应的ZooKeeper软件了，直接使用这里现成的即可。

## 配置环境变量 ##

本来Kafka和ZooKeeper是属于两个软件，可以分别配置环境变量的，但是由于这里采用的是Kafka中含有内置的ZooKeeper，所以我们现在也就省事了，直接配置一个Kafka的环境变量即可（此时它们共用一个环境变量）。

![在这里插入图片描述][20190609101036182.png]

然后在Path中做以下配置：

%KAFKA_HOME%\bin\windows

## 配置ZooKeeper ##

ZooKeeper的配置文件所在位置：

安装目录 + config\zookeeper.properties

其默认的端口号如下：

clientPort=2181

其数据存储目录：

dataDir=D:/Dev/Kafka/kafka_2.11-2.2.0/zookeeper

其中`dataDir`指的是该软件的数据存储位置，此时根据我们的实际情况自定义相应的存储位置。

## 配置Kafka ##

Kafka的配置文件所在位置：

安装目录 + config\server.properties

在这里，我们指定相应的日志文件存储位置：

log.dirs=D:/Dev/Kafka/kafka_2.11-2.2.0/kafka-logs

由于Kafka需要依赖ZooKeeper，而我们的Kafka又内置了ZooKeeper，因而此处的Kafka监听的ZooKeeper端口号直接使用默认的即可，默认的Kafka监听的ZooKeeper端口号为：

zookeeper.connect=localhost:2181

# 源码 #

**引入Kafka依赖**

由于我这里使用了SpringCloud微服务架构，而在该架构中的`spring-cloud-dependencies`已经指定了相应的依赖版本号，因而此处我就没必要再次给出了。

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

**Java源码**

OrderApplication类

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.CommandLineRunner;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
    import org.springframework.cloud.client.loadbalancer.LoadBalanced;
    import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
    import org.springframework.cloud.netflix.hystrix.EnableHystrix;
    import org.springframework.cloud.openfeign.EnableFeignClients;
    import org.springframework.context.annotation.Bean;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.client.RestTemplate;
    
    /** * @author: 张侦毅 * @date: 2019/6/4 10:38 * @description: 订单Bootstrap启动类 */
    @Slf4j
    @SpringBootApplication
    @EnableEurekaClient
    public class OrderApplication implements CommandLineRunner { 
    
        public static void main(String[] args) { 
            SpringApplication.run(OrderApplication.class,args);
        }
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        /** * 发送信息 * @param args * @throws Exception */
        @Override
        public void run(String... args) throws Exception { 
            kafkaTemplate.send("KafkaQueue","Hello World!");
        }
    
        /** * 接收信息 * @param message 接收的信息 */
        @KafkaListener(topics = "KafkaQueue", groupId = "consumer-group")
        public void listener(String message) { 
            log.info("Received Message：{}",message);
        }
    
    }

**配置文件**

application.yml部分配置文件信息

#kafka
      kafka:
        bootstrap-servers: localhost:9092
        consumer:
              group-id: test-consumer-group
              auto-offset-reset: earliest
              key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
              value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

# 使用Java连接Kafka #

## 启动Kafka相关的服务器 ##

进入Kafka的安装目录，按住【Shift】键，鼠标右击，在弹出的菜单中选择：在此处打开命令窗口

启动ZooKeeper

zookeeper-server-start config\zookeeper.properties

启动Kafka

kafka-server-start config\server.properties

当我们最后看到下述信息时，这说明我们的Kafka软件已经启动成功

...
    [2019-06-09 10:46:47,012] INFO Kafka version: 2.2.0 (org.apache.kafka.common.utils.AppInfoParser)
    [2019-06-09 10:46:47,012] INFO Kafka commitId: 05fcfde8f69b0349 (org.apache.kafka.common.utils.AppInfoParser)
    [2019-06-09 10:46:47,014] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

## 运行结果 ##

启动微服务项目，当我们在IDEA控制台中看到下述信息时，这说明我们的软件已经能够正常使用Kafka了。

2019-06-09 10:51:15.281  INFO 13064 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=consumer-group] (Re-)joining group
    2019-06-09 10:51:15.321  INFO 13064 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=consumer-group] Successfully joined group with generation 1
    2019-06-09 10:51:15.322  INFO 13064 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=consumer-group] Setting newly assigned partitions [KafkaQueue-0]
    2019-06-09 10:51:15.337  INFO 13064 --- [ntainer#0-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-2, groupId=consumer-group] Resetting offset for partition KafkaQueue-0 to offset 0.
    2019-06-09 10:51:15.338  INFO 13064 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [KafkaQueue-0]
    2019-06-09 10:51:15.380  INFO 13064 --- [ntainer#0-0-C-1] com.**.order.OrderApplication         : Received Message：Hello World!

[Kafka]: http://kafka.apache.org
[Kafka Download]: http://kafka.apache.org/downloads
[Kafka Documentation]: http://kafka.apache.org/documentation/
[20190609101036182.png]: /images/20220121/9864d03432d54f53a9d4fd2047f44224.png