Spring boot集成kafka(windows)

1.环境搭建

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

解压缩即可完成安装

2.服务启动

//我演示使用的windows版，命令如下
//进入命令执行目录
cd D:\queue\kafka_2.13-2.8.1\bin\windows
//启动zookeeper
zookeeper-server-start.bat ..\..\config\zookeeper.properties
//启动kafka
kafka-server-start.bat ..\..\config\server.properties

我使用的是2.8.1这个版本，启动时可以直接使用kafka包自带的zookeeper进行启动，如果启动完zookeeper后，启动kafka发生错误，可能需要下载一个和kafka相应版本的zookeeper。

3.名词说明

• Broker。负责接收和处理客户端发送过来的请求，以及对消息进行持久化。虽然多个 Broker 进程能够运行在同一台机器上，但更常见的做法是将不同的 Broker 分散运行在不同的机器上
• 主题：Topic。主题是承载消息的逻辑容器，在实际使用中多用来区分具体的业务。
• 分区：Partition。一个有序不变的消息序列。每个主题下可以有多个分区。
• 消息：这里的消息就是指 Kafka 处理的主要对象。
• 消息位移：Offset。表示分区中每条消息的位置信息，是一个单调递增且不变的值。
• 副本：Replica。Kafka 中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者副本和追随者副本，各自有不同的角色划分。每个分区可配置多个副本实现高可用。一个分区的N个副本一定在N个不同的Broker上。
• Leader：每个分区多个副本的“主”副本，生产者发送数据的对象，以及消费者消费数据的对象，都是 Leader。
• Follower：每个分区多个副本的“从”副本，实时从 Leader 中同步数据，保持和 Leader 数据的同步。Leader 发生故障时，某个 Follower 还会成为新的 Leader。
• 生产者：Producer。向主题发布新消息的应用程序。
• 消费者：Consumer。从主题订阅新消息的应用程序。
• 消费者位移：Consumer Offset。表示消费者消费进度，每个消费者都有自己的消费者位移。offset保存在broker端的内部topic中，不是在clients中保存
• 消费者组：Consumer Group。多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐。
• 重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance 是 Kafka 消费者端实现高可用的重要手段。

4.执行步骤

(1) Producer 生产消息，发送到Broker中

(2) Leader状态的Broker接收消息，写入到相应topic中。在一个分区内，这些消息被索引并连同时间戳存储在一起

(3) Leader状态的Broker接收完毕以后，传给Follow状态的Broker作为副本备份

(4) Consumer 消费者的进程可以从分区订阅，并消费消息

5.示例

(1)引入依赖

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

这边默认使用2.8.10版本

(2)配置

  kafka:
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: kafak_order

(2)服务层

package com.example.demo.kafka.impl;

import cn.hutool.core.util.StrUtil;
import com.example.demo.kafka.KafkaMessageService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author linaibo
 * @version 1.0
 * Create by 2022/12/25 19:59
 */
@Service
public class KafkaMessageServiceImpl implements KafkaMessageService {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Override
    public void sendMessage(String msg) {
        System.out.println("Kafka消息开始发送");
        ListenableFuture<SendResult<String,String>> future = kafkaTemplate.send("order",msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败" + ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                //主题
                String topic = result.getRecordMetadata().topic();
                //分区
                int partition = result.getRecordMetadata().partition();
                //offset
                long offset = result.getRecordMetadata().offset();
                System.out.println(StrUtil.format("消息发送成功,topic={},partition={},offset={}",topic,partition,offset));
            }
        });
    }}

将消息发送到特定的主题中，并对结果进行判断，成功的场合可以获取消息的信息，失败的场合可以获取失败的原因。

(3)监听器

package com.example.demo.kafka.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author linaibo
 * @version 1.0
 * Create by 2022/12/25 20:02
 */
@Component
public class KafkaMessageListener {

    @KafkaListener(topics = {"order"})
    public void onMessage(ConsumerRecord<String,String> record){
        System.out.println("接收到发送的kafka消息" + record.value());
        System.out.println("接收到发送的kafka消息" + record);

    }
}

运行结果

Kafka消息开始发送
2022-12-31 13:10:14.366  INFO 78104 --- [nio-8888-exec-1] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
    acks = -1
    batch.size = 16384
    bootstrap.servers = [127.0.0.1:9092]
    buffer.memory = 33554432
    client.dns.lookup = use_all_dns_ips
    client.id = producer-1
    compression.type = none
    connections.max.idle.ms = 540000
    delivery.timeout.ms = 120000
    enable.idempotence = true
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 0
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metadata.max.idle.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.connect.timeout.ms = null
    sasl.login.read.timeout.ms = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.login.retry.backoff.max.ms = 10000
    sasl.login.retry.backoff.ms = 100
    sasl.mechanism = GSSAPI
    sasl.oauthbearer.clock.skew.seconds = 30
    sasl.oauthbearer.expected.audience = null
    sasl.oauthbearer.expected.issuer = null
    sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
    sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
    sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
    sasl.oauthbearer.jwks.endpoint.url = null
    sasl.oauthbearer.scope.claim.name = scope
    sasl.oauthbearer.sub.claim.name = sub
    sasl.oauthbearer.token.endpoint.url = null
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.2
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.StringSerializer

2022-12-31 13:10:14.376  INFO 78104 --- [nio-8888-exec-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-1] Instantiated an idempotent producer.
2022-12-31 13:10:14.390  INFO 78104 --- [nio-8888-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.1.2
2022-12-31 13:10:14.390  INFO 78104 --- [nio-8888-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: f8c67dc3ae0a3265
2022-12-31 13:10:14.390  INFO 78104 --- [nio-8888-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1672463414390
2022-12-31 13:10:14.397  INFO 78104 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Resetting the last seen epoch of partition order-0 to 0 since the associated topicId changed from null to Cc4Zxy54SbC-ylUTv7wCKQ
2022-12-31 13:10:14.397  INFO 78104 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : [Producer clientId=producer-1] Cluster ID: 9kfe9DyPRpeGhkcflXFSTA
2022-12-31 13:10:14.398  INFO 78104 --- [ad | producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=producer-1] ProducerId set to 1002 with epoch 0
接收到发送的kafka消息258741369
接收到发送的kafka消息ConsumerRecord(topic = order, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1672463414397, serialized key size = -1, serialized value size = 9, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = 258741369)
消息发送成功,topic=order,partition=0,offset=5

6.集群搭建

kafka集群、listeners配置、broker&topic&partition(leader、follower)关系、消费问题(同步、异步、ack应答机制、缓冲区、消费者poll长轮询)_鸢尾の的博客-CSDN博客_kafka listeners配置

7.问题

报错{orderInfo=UNKNOWN_TOPIC_OR_PARTITION}

服务启动中修改了log文件的路径

解决方法：

将log文件路径删除，再次重新启动zookeeper和kafka ，两个log的文件目录要一致