一、安装Kafka

下载地址：Apache Kafka 以kafka版本3.3.1为例

  • 在windows环境下先解压下载的压缩包，kafka压缩包中包含了需要的zookeeper

  • 将kafka解压，如解压后文件名为kafka并放在D盘下

  • 编写两个bat，一个用于启动zookeeper，一个用于启动kafka

    创建启动zookeeper的bat如zookeeper.startup.bat

1

D:\kafka\bin\windows\zookeeper-server-start.bat D:\kafka\config\zookeeper.properties

        创建启动kafka的bat如kafka.startup.bat

1

D:\kafka\bin\windows\kafka-server-start.bat D:\kafka\config\server.properties
  • 先启动zookeeper，再启动kafka。

二、服务配置

配置文件位于config文件夹下的server.properties，如D:\kafka\config\server.properties

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

//当前机器在集群中的唯一标识，和zookeeper的myid性质一样
broker.id=0
//当前kafka对外提供服务的端口默认是9092
port=9092
//这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
# host.name=node01
//这个是borker进行网络处理的线程数
num.network.threads=3
//这个是borker进行I/O处理的线程数
num.io.threads=8
//发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
socket.send.buffer.bytes=102400
//kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
socket.receive.buffer.bytes=102400
//这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
socket.request.max.bytes=104857600
//消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，
//如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
log.dirs=/home/hadoop/log/kafka-logs
//默认的分区数，一个topic默认1个分区数
num.partitions=1
//每个数据目录用来日志恢复的线程数目
num.recovery.threads.per.data.dir=1
//默认消息的最大持久化时间，168小时，7天
log.retention.hours=168
//这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
log.segment.bytes=1073741824
//每隔300000毫秒去检查上面配置的log失效时间
log.retention.check.interval.ms=300000
//是否启用log压缩，一般不用启用，启用的话可以提高性能
log.cleaner.enable=false
//设置zookeeper的连接端口
zookeeper.connect=192.168.123.102:2181,192.168.123.103:2181,192.168.123.104:2181
//设置zookeeper的连接超时时间
zookeeper.connection.timeout.ms=6000

三、常用命令

下面示例采用的kafka版本是：3.3.1

创建topic

创建一个名字为test的主题， 有三个分区，有两个副本：

1

kafka-topics.bat --bootstrap-server localhost:9092 --create --topic test --partitions 3 --replication-factor 2

查看topic副本信息

1

kafka-topics.bat --bootstrap-server localhost:9092 --describe --topic test

查看kafka当中存在的topic

1

kafka-topics.bat --bootstrap-server localhost:9092 --list

删除topic

1

kafka-topics.bat --bootstrap-server localhost:9092 --delete --topic test

生产者生产数据

1

kafka-console-producer.bat --broker-list localhost:9092 --topic test

消费者消费数据

1

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --group my-group --from-beginning

查看消费者组的偏移量

1

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group my-group

增加topic分区

1

kafka-topics.bat --bootstrap-server localhost:9092 --alter --topic test --partitions 5

特定偏移量重新消费

  • 停止当前的消费者应用程序。
  • 使用 Kafka 提供的命令行工具来重置消费者组的偏移量到指定的偏移量位置：
1

kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group my-group --topic test --reset-offsets --to-offset 100 --execute

上述命令将将消费者组 my-group 在名为 test 的主题上的偏移量重置为偏移量 100。

四、Springboot中使用kafka

添加 Maven 依赖

1
2
3
4

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

配置kafka

在 application.yml 配置文件中添加 Kafka 配置

  • 生产者配置
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092 #kafka 集群地址,多个kafka节点用逗号隔开  例：127.0.0.1:9092,127.0.0.1:9093
    producer: #生产者配置
      client-id: necor-kafka-producer-dev # 客户端ID,发出请求时传递给服务器;⽤于服务器端⽇志记录
      key-serializer: org.apache.kafka.common.serialization.StringSerializer # Key 序列化类
      value-serializer: org.apache.kafka.common.serialization.StringSerializer # Value 序列化类
      retries: 0 #重试次数,如果该值⼤于零时，表⽰启⽤重试失败的发送次数
      batch-size: 4096   #批量处理的最大大小 单位 byte
      buffer-memory: 33554432 # ⽣产者可⽤于缓冲等待发送到服务器的记录的内存总字节数，默认值为33554432。发送延时,当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
      compression-type: gzip # 消息压缩：none、lz4、gzip、snappy，默认为 none。
      # 应答级别 默认为 all
      # acks=0 把消息发送到kafka就认为发送成功
      # acks=1 把消息发送到kafka leader分区，并且写入磁盘就认为发送成功
      # acks=all/-1 把消息发送到kafka leader分区，并且leader分区的副本follower对消息进行了同步就任务发送成功
      acks: 1
      properties:
        retry.backoff.ms: 100 #重试时间间隔，默认100
        linger.ms: 0 #默认为0，表示批量发送消息之前等待更多消息加入batch的时间 设置为0则buffer-memory没用
        max.request.size: 1048576 #默认1MB，表示发送消息最大值
        connections.max.idle.ms: 540000 #默认9分钟，表示多久后关闭限制的连接
        receive.buffer.bytes: 32768 #默认32KB，表示socket接收消息缓冲区的大小，为-1时使用操作系统默认值
        send.buffer.bytes: 131072 #默认128KB，表示socket发送消息缓冲区大小，为-1时使用操作系统默认值
        request.timeout.ms: 30000 #默认30000ms，表示等待请求响应的最长时间
  • 消费者配置
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

spring:
  kafka:
    listener:
      type: single #设置是否批量消费，默认 single（单条），batch（批量）
    bootstrap-servers: 127.0.0.1:9092 #kafka 集群地址,多个kafka节点用逗号隔开  例：127.0.0.1:9092,127.0.0.1:9093
    consumer:
      group-id: default-group # 指定默认消费者group id
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 键的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 值的反序列化方式
      # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量。
      # 手动提交还需要设置 spring.kafka.listener.ack-mode为 manual
      enable-auto-commit: true
      # 自动提交的时间间隔。每隔一段时间提交一次 如果'enable.auto.commit'为true，则消费者偏移⾃动提交给Kafka的频率（以毫秒为单位），默认值为5000。
      # 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
      auto-commit-interval: 1S
      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
      auto-offset-reset: earliest

编写代码

  • 创建生产者发送消息
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.Date;

@RestController("KafkaProducer")
@RequestMapping("/kafka/producer")
public class Producer {

    @Autowired
    private KafkaTemplate<Object, Object> template;

    @GetMapping(value="/send/{msg}")
    public String send(HttpServletRequest request, @PathVariable String msg){
          String topic = "topic"; // 定义topic
          String key = "msgKey"; // 按照 Key 进行哈希，相同 Key 去一个 Partition
          long timestamp = new Date().getTime(); // 时间戳
          Message<String> message = MessageBuilder
                .withPayload(msg)
                .setHeader(KafkaHeaders.TOPIC, topic)
                .setHeader(KafkaHeaders.MESSAGE_KEY, key)
                .setHeader("ip", "127.0.0.1")
                .build(); // 创建消息体
          // ==========================  消息发送示例 ===================================================
        this.template.send(topic, msg); // 简单发送，只填写topic和message

        this.template.send(topic,0,timestamp,key, msg); // 填写topic, partition, key, message发送消息

        this.template.send(message);  // 自定义请求头发送消息

        this.template.send(topic,message).get(); // 同步发送消息

        // 异步发送消息回调
        ListenableFuture<SendResult<Object, Object>> future = this.template.send(topic, msg); 
        future.addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {
            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                System.out.println("消息发送成功." + result.toString());
            }
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("消息发送失败: " + ex.getMessage());
            }
        });

        return msg;
    }
}

上面代码生产消息分别为简单发送方式、指定partition发送方式、自定义请求头发送方式、同步发送方式（kafka默认为异步发送）以及异步发送消息添加回调的发送方式。

  • 创建消费者消费消息
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

import com.necor.mq.common.ResponseData;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.Date;
import java.util.List;

@Slf4j
public class KafkaConsumer {

    /**
     * 简单的消费
     * @param msg
     */
    @KafkaListener(id = "simpleConsumer", topics = {"topic"})
    public void consumer(String msg) {
        log.info("消费消息: {}" , msg);
    }

    /**
     * 获取请求头并消费
     */
    @KafkaListener(id = "requestHeaderConsumer", topics = {"topic"})
    public void consumer(ConsumerRecord<String, String> record, @Header("ip") String ip, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer<String, String> consumer) {
        log.info("请求头信息，topic:{}, msg:{}, ip:{}, key:{}, partition:{}, timestamp:{}",topic, record.value(), ip, record.key(), record.partition(), record.timestamp());
    }

    /**
     * 批量消费
     * 批量消费需要修改kafka消费者配置
     * spring
     *   kafka:
     *     listener:
     *       type: batch #设置是否批量消费，默认 single（单条），batch（批量）
     */
    @KafkaListener(id = "batchConsumer", topics = {"topic"})
    public void consumer(List<ConsumerRecord> records) {
        records.forEach(record->{
            log.info("消费者接收到消息: {}" , record.value());
        });
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 手动确认消费
     * 需要修改消费者配置
     * spring
     *   kafka:
     *     listener:
     *       ack-mode: manual # 配合 spring.kafka.consumer.enable-auto-commit 设置手动提交使用
     *     consumer:
     *       enable-auto-commit: false
     */
    @KafkaListener(id = "manualSyncConsumer", topics = {"topic"})
    public void consumer(String msg,Consumer consumer) {
        log.info("消费者1接收到消息: {}" , msg);
        // kafka客户端提供同步提交偏移量
        consumer.commitSync();
    }
    @KafkaListener(id = "manualAsyncConsumer", topics = {"topic"})
    public void consumer(ConsumerRecord record, Consumer consumer) {
        log.info("消费者2接收到消息: {}" , record.value());
        // kafka客户端提供异步提交偏移量
        consumer.commitAsync();
    }
    @KafkaListener(id = "manualConsumer", topics = {"topic"})
    public void consumer(ConsumerRecord record, Acknowledgment ack) {
        log.info("消费者3接收到消息: {}" , record.value());
        // kafka提供手动提交偏移量
        ack.acknowledge();
    }

    /**
     * 广播消费
     * 需要修改配置
     */
    @KafkaListener(id = "broadcastConsumer",groupId = "broadcastGroup0",topics = {"topic"})
    public void broadcast0(String msg) {
        log.info("消费消息: {}" , msg);
    }
    @KafkaListener(id = "broadcastConsumer",groupId = "broadcastGroup1",topics = {"topic"})
    public void broadcast1(String msg) {
        log.info("消费消息: {}" , msg);
    }
}

上面是不同消费模式的示例代码，注意有的需要修改消费者配置。

手动提交消费偏移量分为使用Consumer提交和Acknowledgment提交，它们的区别在于Acknowledgment是消费者监听方法内部调用的，它将确认当前消费的消息并提交相应的偏移量， Consumer是 Kafka 客户端提供的提交偏移量的方法，Consumer提交偏移量的方式又分为同步提交和异步提交。从颗粒度上讲Acknowledgment是每条消息被成功处理后手动调用，因此可以实现每条消息的精确提交。Consumer是提交一批消息的偏移量。你需要在适当的时机选择调用这些方法，例如在处理一批消息后、定期提交或基于特定的业务逻辑来确定提交时机。

消息传递模式分为点对点模式和发布-订阅消息模式，

在点对点消息系统中，消息持久化到一个队列中。此时有一个或多个消费者消费队列中的消息。但是一条消息只能被一个消费者消费。当一个消费者消费了队列中的这条消息后，该条数据则从消息队列中删除。

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。

kafka使用同一个消费者组或不用消费者组消费同一个topic实现了点对点和发布-订阅模式。

1

五、使用中需要注意的问题

kafka如何防止消息丢失

kafka消息可能会因为网络波动、服务异常等原因造成消息丢失，防止消息丢失可以分为三点

  • 生产防止数据丢失：生产者发送消息到kafka服务端可能由于网络问题或kafka服务宕机导致消息发送失败。可以将发送方式改为同步发送（默认为异步发送）获取实时发送结果，也可以使用异步发送添加回调函数监听发送结果来对发送失败的消息进行重试，并且kafka producer本身提供了重试机制，只需要配置retries即可对失败的消息自动进行重试。

  • kafka集群中的broker防止数据丢失：borker需要把producer发送的消息持久化到磁盘，由于kafka为了高性能，采用异步批量刷盘的实现机制，也就是说按照一定的消息量和时间间隔进行刷盘，而最终刷盘的动作是交给系统调度的，所以在刷盘之前系统崩溃会导致消息丢失。然而kafka并未提供同步刷盘的机制，所以我们需要借助kafka的副本机制和acks机制来确保broker写盘成功。例如将acks设置为-1（把消息发送到kafka leader分区，并且leader分区的副本follower对消息进行了同步就任务发送成功）。

  • 消费防止数据丢失：消费者手动提交偏移量（默认自动提交），这样即使消费者在运行过程中挂掉，也可以再此启动后重新找到该topic的offset接着消费。但是offset的信息写入的时候并不是每条消息消费完成后都写入的，所以这种情况有可能会造成重复消费，但是不会丢失消息。

kafka如何保证顺序消费

kafka只能保证partition内是有序的，但是partition间的有序是没办法的。让需要保证顺序消费的topic放到同一个分区下。