# 前言
Apache Kafka 是一款开源的消息系统，在开发各类系统的时候，我们经常会选择使用Kafka来帮助我们削峰、做异步处理、解耦，那么正好借此机会跟大家说说kafka的安装部署、应用场景以及简单的使用。

## kafka的安装部署
想在个人电脑搭建kafka的环境，我们需要三个东西，分别是JDK、Zookeeper和Kafka。（JDK的安装在这里就不说了，我们直接说重点）。kafka的运行是依赖于zookeeper的，所以我们需要先装一个zookeeper。

有些小伙伴会说了，kafka不是内置了一套zookeeper嘛？我是不是就不用重新安装zookeeper了？
说的没错，我们确实可以直接使用内置的zookeeper来启动kafka，不过我们这里讲解的详细一些，所以就单独安装一个zookeeper。
#### 部署zookeeper
咱们先去[Zookeeper官网](http://https://zookeeper.apache.org/)上下载一个安装包（最好下载稳定版的）
我这里下载的是apache-zookeeper-3.5.9-bin.tar.gz，把文件保存在本地的目录下并解压，解压后我们需要先在根目录下新建一个data文件夹，目录结构如下图所示：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/59827758074216b7e8460f82f79cc4bf.png#pic_center)
接下来我们进入操conf文件夹下，复制一份zoo_sample.cfg文件，并将复制出来的文件改名为zoo.cfg，打开zoo.cfg文件找到dataDir这一行，将这一行的路径改成上面新建的data文件夹的路径。
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/92b07c32c62e864e73bd0ed544d6cc25.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/f97a45a61fb4088eea5ccbdf57ec4061.png)
最后我们把zookeeper添加到环境变量

 - 添加系统变量：ZOOKEEPER_HOME = D:\kafkaTools\zookeeper\apache-zookeeper-3.5.9-bin
 - 添加环境变量：%ZOOKEEPER_HOME%\bin
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/de002e3310127359c34140bf6caa87f5.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/af8350c3f1eb26589dd3f92c9a0a32a8.png)
这时候我们的zookeeper就部署完成了，在DOS下运行：zkserver 即可启动zookeeper（默认端口：2181）。

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/deb046b137b069f6a2bee653e69f6fc4.png#pic_center)
#### 部署kafka
zookeeper搭建成功后，我们就可以开始部署kafka了，我们依然是先去官网下载一个安装包。
官网下载地址：http://kafka.apache.org/downloads.html
这里我下载的是kafka_2.13-2.7.0.tgz，把文件保存在本地的目录下并解压，解压后我们需要先在根目录下新建一个logs文件夹，目录结构如下图所示：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/605abadf74bb226ed028fe9f94688344.png#pic_center)
接下来我们进入到config文件夹下，编辑server.properties文件。
log.dirs= D:\\kafkaTools\\kafka\\kafka_2.13-2.7.0\\logs（刚刚新建的logs文件夹的目录）
zookeeper.connect=localhost:2181（表示连接本地的zookeeper服务）
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/965cdc0bfbd852a0901212e035fa4be6.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/928e5b11f7af85d9eee41543e4eb79d2.png)
接下来我们就可以启动kafka了，==这里需要注意的一点是，启动kafka之前要确保zookeeper已经成功启动。==
我们进到kafka的根目录下，在这里打开DOS窗口，运行.\bin\windows\kafka-server-start.bat .\config\server.properties 即可启动kafka。
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/9a877206e620a56cd8aa66c2b56e28fc.png)
至此，我们本机的kafka环境就已经部署完成了。

P.S. 这里有可能会报错：“找不到或无法加载主类 Files\java\jdk-9.0.1\lib;C:\Program”
解决办法：在kafka安装目录中找到bin/windows目录中的kafka-run-class.bat文件，把setCOMMAND这一行的%CLASSPATH%前后都加上双引号。也就是将%CLASSPATH%改为"%CLASSPATH%"


## kafka的应用场景
 - 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer
 - 消息系统：解耦生产者和消费者、缓存消息等
 - 用户活动跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后消费者通过订阅这些topic来做实时的监控分析，亦可保存到数据库
 - 运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告
 - 流式处理：比如spark streaming和storm

## kafka的简单示例
在kafka中我们经常会看到两个名词“生产者”和“消费者”，那么我们就写两个方法，模拟一下“生产”和“消费”的流程。

我们先新建一个RunKafkaProduce.java文件来模拟一下生产者【向主体（Topic）发布新消息】

```java
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

/**
 * @program: ziye
 * @description: RunKafkaProduce
 * @author: 庄霸.liziye
 * @create: 2021-03-25 15:38
 **/
public class RunKafkaProduce {
    private final Producer<String, String> producer;
    public final static String TOPIC = "logstest";

    private RunKafkaProduce(){
        Properties props = new Properties();
        // 此处配置的是kafka的broker地址:端口列表
        props.put("metadata.broker.list", "127.0.0.1:9092");
        //配置value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");

        //配置key的序列化类
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");

        //request.required.acks
        //0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees (some data will be lost when a server fails).
        //1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
        //-1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be lost as long as at least one in sync replica remains.
        props.put("request.required.acks","-1");

        producer = new Producer<String, String>(new ProducerConfig(props));
    }

    void produce() {
        int messageNo = 1;
        final int COUNT = 11;

        int messageCount = 0;
        while (messageNo < COUNT) {
            String key = String.valueOf(messageNo);
            String data = "Hello kafka message :" + key;
            producer.send(new KeyedMessage<String, String>(TOPIC, key ,data));
            System.out.println(data);
            messageNo ++;
            messageCount++;
        }
        System.out.println("Producer端一共产生了" + messageCount + "条消息！");
    }

    public static void main( String[] args )
    {
        new RunKafkaProduce().produce();
    }
}
```
这里我们简单讲解一下request.required.acks参数。

Kafka producer的ack有3中机制,初始化producer时的producerconfig可以通过配置request.required.acks不同的值来实现：

 - 0：这意味着生产者producer不等待来自broker同步完成的确认继续发送下一条（批）消息。此选项提供最低的延迟但最弱的耐久性保证（当服务器发生故障时某些数据会丢失，如leader已死，但producer并不知情，发出去的信息broker就收不到）。
 - 1：这意味着producer在leader已成功收到的数据并得到确认后发送下一条message。此选项提供了更好的耐久性为客户等待服务器确认请求成功（被写入死亡leader但尚未复制将失去了唯一的消息）。
 - -1：这意味着producer在follower副本确认接收到数据后才算一次发送完成。 此选项提供最好的耐久性，我们保证没有信息将丢失，只要至少一个同步副本保持存活。

==三种机制，性能依次递减 (producer吞吐量降低)，数据健壮性则依次递增。==

接下来我们再写一个“消费者”【从主体（Topic）订阅新消息】

```java
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @program: ziye
 * @description: RunKafkaConsumer
 * @author: 庄霸.liziye
 * @create: 2021-03-25 15:41
 **/
public class RunKafkaConsumer {
    private final ConsumerConnector consumer;

    private final static  String TOPIC="logstest";

    private RunKafkaConsumer(){
        Properties props=new Properties();
        //zookeeper连接地址
        props.put("zookeeper.connect","127.0.0.1:2181");
        //topic
        props.put("group.id","logstest");
        //Zookeeper 超时
        props.put("zookeeper.session.timeout.ms", "50000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");

        props.put("serializer.class", "kafka.serializer.StringEncoder");
        ConsumerConfig config=new ConsumerConfig(props);
        consumer= kafka.consumer.Consumer.createJavaConsumerConnector(config);
    }

    void consume(){
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(TOPIC, new Integer(1));

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
        KafkaStream<String, String> stream = consumerMap.get(TOPIC).get(0);
        ConsumerIterator<String, String> it = stream.iterator();

        int messageCount = 0;
        while (it.hasNext()){
            System.out.println(it.next().message());
            messageCount++;
            if(messageCount == 10){
                System.out.println("Consumer端一共消费了" + messageCount + "条消息！");
            }
        }
    }

    public static void main(String[] args) {
        new RunKafkaConsumer().consume();
    }
}
```
我们运行分别RunKafkaProduce.java和RunKafkaConsumer.java结果如下图所示（==注意：需要先启动生产者==）：
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/8f82ac2dd9eb29a1b7e5ee88d8f324a1.png)
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-4287693/2407058bc6bbbdb452941fdb1846d0ea.png)
这里我们再对kafka的一些名词做一个解释：
- 消息：Record。Kafka是消息引擎，这里的消息就是Kafka处理的主要对象。
- 主体：Topic。主题就是承载消息的逻辑容器，在实际应用中多用于区分具体业务。
- 消息位移：Offset。表示分区中每条消息的位置信息，是一个单调递增不变的值。
- 副本：Replica。Kafka中一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随者副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可配置多个副本实现高可用。
- 生产者：Producer 。 向主体发布新消息的应用程序。
- 消费者：Consumer。从主体订阅新消息的应用程序。
- 消费者位移：Consumer Offset 。表示消费者消费进度，每个消费者都有自己的消费者位移。
- 消费者组：Consumer Group 。多个消费者实例共同组成的一个组，同时消费多个分区实现高吞吐。
- 重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其它消费者实例自动重新分配订阅主题分区的过程。Rebalance 是kafka
消费者端实现高可用的重要手段。

# 小结
Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。简单地说，Kafka就相比是一个快递代收点，生产者是发送快递的人，消费者是接收快递的人，Kafka就是用来存快递的，只不过它提供了一些处理快递的机制。

Kafka有很多优势：
1. **高吞吐量、低延迟：** kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒
2. **可扩展性：** kafka集群支持热扩展
3. **持久性、可靠性：** 消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
4. **容错性：** 允许集群中节点故障（若副本数量为n,则允许n-1个节点故障）
5. **高并发：** 支持数千个客户端同时读写

当然kafka也有着不可忽视的缺点，消息是批量发送的，也就意味着数据并非是真正的实时，而且会出现消息丢失、消息重复、消息乱序、消息堆积的情况。
所以在我们使用kafka的时候一定要考虑到kafka的利与弊，在适合的场景选择合适的工具。

如果文章中有错误，欢迎大家留言指正；若您有更好、更独到的理解，欢迎您在留言区留下您的宝贵想法。

> *你在被打击时，记起你的珍贵，抵抗恶意；*
*你在迷茫时，坚信你的珍贵，抛开蜚语；*
**爱你所爱  行你所行  听从你心  无问东西**


Apache Kafka 是一款开源的消息系统，在开发各类系统的时候，我们经常会选择使用Kafka来帮助我们削峰、做异步处理、解耦，那么正好借此机会跟大家说说kafka的安装部署、应用场景以及简单的使用。

大聪明教你学Java | Windows10系统下kafka安装及使用

高级Java开发工程师

后端

编程语言

数据库

Apache Kafka是一款高性能分布式消息系统，支持日志收集、消息队列、用户行为跟踪等场景。本文详细介绍Kafka安装部署流程，包括Zookeeper配置、Kafka环境搭建及常见问题解决，并提供Java代码示例演示生产者消费者模式。Kafka具有高吞吐、低延迟、可扩展等优势，但也存在消息丢失、重复等问题，需根据业务场景合理选用。

数据备份

数据丢失

数据冗余

服务器

流数据

容器

Kafka

Java

4核4G3M云服务器 新用户低至38元/年！

2026新春采购季

tencentdb-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Apache Kafka是一款高性能分布式消息系统，支持日志收集、消息队列、用户行为跟踪等场景。本文详细介绍Kafka安装部署流程，包括Zookeeper配置、Kafka环境搭建及常见问题解决，并提供Java代码示例演示生产者消费者模式。Kafka具有高吞吐、低延迟、可扩展等优势，但也存在消息丢失、重复等问题，需根据业...

大聪明教你学Java | Windows10系统下kafka安装及使用-腾讯云开发者社区-腾讯云

大聪明教你学Java | Windows10系统下kafka安装及使用

大聪明教你学Java | Windows10系统下kafka安装及使用

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐