一篇文章速通kafka——day02

国内服务器12小时前发布 beixibaobao

3 0 0

经过昨天学习了kafaka的基本知识和概念后，今天将使用springboot集成kafka，进行具体的操作，全程高能，步骤十分清晰，让你快速搞懂kafka的基本使用api

六、kafka的java客户端-⽣产者的实现

1.⽣产者的基本实现

2.⽣产者的同步发送消息

3.⽣产者的异步发送消息

4.⽣产者中的ack的配置

5.关于消息发送的缓冲区

七、Java客户端消费者的实现细节

1.消费者的基本实现

2.关于消费者⾃动提交和⼿动提交offset

3.⻓轮询poll消息

4.消费者的健康状态检查

5.指定分区和偏移量、时间消费

6.新消费组的消费offset规则

⼋、Springboot中使⽤Kafka

1.引⼊依赖

2.编写配置⽂件

3.编写消息⽣产者

4.编写消费者

5.消费者中配置消费主题、分区和偏移量

九、kafka集群中的controller、

rebalance、HW

1.controller

2.rebalance机制

3.HW和LEO

⼗、Kafka中的优化问题

1.如何防⽌消息丢失

2.如何防⽌重复消费

3.如何做到消息的顺序消费

4.如何解决消息积压问题

5.实现延时队列的效果

⼗⼀、Kafka-eagle监控平台

1.搭建

2.平台的使⽤

六、kafka的java客户端-⽣产者的实现

1.⽣产者的基本实现

引⼊依赖

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.1</version>
</dependency>

具体实现

package com.qf.kafka;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class MySimpleProducer {
private final static String TOPIC_NAME = "my-replicated-topic";
public static void main(String[] args) throws ExecutionException,
InterruptedException {
//1.设置参数
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094");
//把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
//2.创建⽣产消息的客户端，传⼊参数
Producer<String,String> producer = new KafkaProducer<String,
String>(props);
//3.创建消息
//key：作⽤是决定了往哪个分区上发，value：具体要发送的消息内容
ProducerRecord<String,String> producerRecord = new ProducerRecord<>
(TOPIC_NAME,"mykeyvalue","hellokafka");
//4.发送消息,得到消息发送的元数据并输出
RecordMetadata metadata = producer.send(producerRecord).get();
System.out.println("同步⽅式发送消息结果：" + "topic-" +
metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());
}
}

2.⽣产者的同步发送消息

如果⽣产者发送消息没有收到ack，⽣产者会阻塞，阻塞到3s的时间，如果还没有收到消息，

会进⾏重试。重试的次数3次

RecordMetadata metadata = producer.send(producerRecord).get();
System.out.println("同步⽅式发送消息结果：" + "topic-" +
metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());

3.⽣产者的异步发送消息

异步发送，⽣产者发送完消息后就可以执⾏之后的业务，broker在收到消息后异步调⽤⽣产者提供的callback回调⽅法

//5.异步发送消息
producer.send(producerRecord, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
System.err.println("发送消息失败：" +
exception.getStackTrace());
}
if (metadata != null) {
System.out.println("异步⽅式发送消息结果：" + "topic-" +
metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
}
}
});

4.⽣产者中的ack的配置

在同步发送的前提下，⽣产者在获得集群返回的ack之前会⼀直阻塞。那么集群什么时候返回

ack呢？此时ack有3个配置：

ack = 0 :kafka-cluster不需要任何的broker收到消息，就⽴即返回ack给⽣产者，最容易丢消息的，效率是最⾼的
ack=1（默认）：多副本之间的leader已经收到消息，并把消息写⼊到本地的log中，才会返回ack给⽣产者，性能和安全性是最均衡的
ack=-1/all :⾥⾯有默认的配置min.insync.replicas=2(默认为1，推荐配置⼤于等于2)，此时就需要leader和⼀个follower同步完后，才会返回ack给⽣产者（此时集群中有2个broker已完成数据的接收），这种⽅式最安全，但性能最差。

下⾯是关于ack和重试（如果没有收到ack，就开启重试）的配置

props.put(ProducerConfig.ACKS_CONFIG, "1");
/*
发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造
成消息重复发送，⽐如⽹络抖动，所以需要在
接收者那边做好消息接收的幂等性处理
*/
props.put(ProducerConfig.RETRIES_CONFIG, 3);
//重试间隔设置
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);

5.关于消息发送的缓冲区

kafka默认会创建⼀个消息缓冲区，⽤来存放要发送的消息，缓冲区是32m

props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

kafka本地线程会去缓冲区中⼀次拉16k的数据，发送到broker

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

如果线程拉不到16k的数据，间隔10ms也会将已拉到的数据发到broker

props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

七、Java客户端消费者的实现细节

1.消费者的基本实现

package com.qf.kafka;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class MySimpleConsumer {
private final static String TOPIC_NAME = "my-replicated-topic";
private final static String CONSUMER_GROUP_NAME = "testGroup";
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094");
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
//1.创建⼀个消费者的客户端
KafkaConsumer<String, String> consumer = new KafkaConsumer<String,
String>(props);
//2. 消费者订阅主题列表
consumer.subscribe(Arrays.asList(TOPIC_NAME));
while (true) {
/*
* 3.poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
//4.打印消息
System.out.printf("收到消息：partition = %d,offset = %d, key =
%s, value = %s%n", record.partition(),
record.offset(), record.key(), record.value());
}
}
}
}

2.关于消费者⾃动提交和⼿动提交offset

1）提交的内容

消费者⽆论是⾃动提交还是⼿动提交，都需要把所属的消费组+消费的某个主题+消费的某个

分区及消费的偏移量，这样的信息提交到集群的_consumer_offsets主题⾥⾯。

2）⾃动提交

消费者poll消息下来以后就会⾃动提交offset

// 是否⾃动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// ⾃动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

注意：⾃动提交会丢消息。因为消费者在消费前提交offset，有可能提交完后还没消费时消费者挂了。

3）⼿动提交

需要把⾃动提交的配置改成false

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

⼿动提交⼜分成了两种：

⼿动同步提交

在消费完消息后调⽤同步提交的⽅法，当集群返回ack前⼀直阻塞，返回ack后表示提交成功，执⾏之后的逻辑

while (true) {
/*
* poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key
= %s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
}
//所有的消息已消费完
if (records.count() > 0) {//有消息
// ⼿动同步提交offset，当前线程会阻塞直到offset提交成功
// ⼀般使⽤同步提交，因为提交之后⼀般也没有什么逻辑代码了
consumer.commitSync();//=======阻塞=== 提交成功
}
}
}

⼿动异步提交

在消息消费完后提交，不需要等到集群ack，直接执⾏之后的逻辑，可以设置⼀个回调⽅法，供集群调⽤

while (true) {
/*
* poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
}
//所有的消息已消费完
if (records.count() > 0) {
// ⼿动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后⾯的程序逻辑
consumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition,OffsetAndMetadata> offsets, Exception exception) {
if (exception != null) {
System.err.println("Commit failed for " + offsets);
System.err.println("Commit failed exception: " +
exception.getStackTrace());
}
}
});
}
}
}

3.⻓轮询poll消息

默认情况下，消费者⼀次会poll500条消息。

//⼀次poll最⼤拉取消息的条数，可以根据消费速度的快慢来设置
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

代码中设置了⻓轮询的时间是1000毫秒

while (true) {
/*
* poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key = %s,
value = %s%n", record.partition(),
record.offset(), record.key(), record.value());
}

意味着：

如果⼀次poll到500条，就直接执⾏for循环
如果这⼀次没有poll到500条。且时间在1秒内，那么⻓轮询继续poll，要么到500条，要么到1s

- 如果多次poll都没达到500条，且1秒时间到了，那么直接执⾏for循环

如果两次poll的间隔超过30s，集群会认为该消费者的消费能⼒过弱，该消费者被踢出消费组，触发rebalance机制，rebalance机制会造成性能开销。可以通过设置这个参数，让⼀次poll的消息条数少⼀点

//⼀次poll最⼤拉取消息的条数，可以根据消费速度的快慢来设置
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
//如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能⼒过弱，将其踢出消费组。将分区分配给其他消费者。-rebalance
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

4.消费者的健康状态检查

消费者每隔1s向kafka集群发送⼼跳，集群发现如果有超过10s没有续约的消费者，将被踢出

消费组，触发该消费组的rebalance机制，将该分区交给消费组⾥的其他消费者进⾏消费。

//consumer给broker发送⼼跳的间隔时间
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
//kafka如果超过10秒没有收到消费者的⼼跳，则会把消费者踢出消费组，进⾏ rebalance，把分区分配给其他消费者。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

5.指定分区和偏移量、时间消费

指定分区消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

从头消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,
0)));

指定offset消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);

指定时间消费

根据时间，去所有的partition中确定该时间对应的offset，然后去所有的partition中找到该

offset之后的消息开始消费。

//根据时间消费偏移量
List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC_NAME);
long date=new Date().getTime()-1000*60*60;
Map<TopicPartition,Long> map=new HashMap<>();
for (PartitionInfo par : partitionInfos) {
map.put(new TopicPartition(TOPIC_NAME, par.partition()), date);
}
Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
TopicPartition key = entry.getKey();
OffsetAndTimestamp value = entry.getValue();
if (key==null || value==null) continue;
Long offset=value.offset();
System.out.println("topicPartition = " + key + ", offsetAndTimestamp = " + offset);
//根据指定时间进行消费
if (value!=null){
consumer.assign(Arrays.asList(key));
consumer.seek(key,offset);
}
}

6.新消费组的消费offset规则

新消费组中的消费者在启动以后，默认会从当前分区的最后⼀条消息的offset+1开始消费（消费新消息）。可以通过以下的设置，让新的消费者第⼀次从头开始消费。之后开始消费新消息（最后消费的位置的偏移量+1）

Latest:默认的，消费新消息

earliest：第⼀次从头开始消费。之后开始消费新消息（最后消费的位置的偏移量+1）

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

⼋、Springboot中使⽤Kafka

1.引⼊依赖

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

2.编写配置⽂件

server:
port: 8080
spring:
kafka:
bootstrap-servers:
172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094
producer: # ⽣产者
retries: 3 # 设置⼤于0的值，则客户端会将发送失败的记录重新发送
batch-size: 16384
buffer-memory: 33554432
acks: 1
# 指定消息key和消息体的编解码⽅式
key-serializer:
org.apache.kafka.common.serialization.StringSerializer
value-serializer:
org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: default-group
enable-auto-commit: false
auto-offset-reset: earliest
key-deserializer:
org.apache.kafka.common.serialization.StringDeserializer
value-deserializer:
org.apache.kafka.common.serialization.StringDeserializer
max-poll-records: 500
listener:

3.编写消息⽣产者

package com.qf.kafka.spring.boot.demo.controller;
import org.apache.kafka.common.protocol.types.Field;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/msg")
public class MyKafkaController {
private final static String TOPIC_NAME = "my-replicated-topic";
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@RequestMapping("/send")
public String sendMessage(){
kafkaTemplate.send(TOPIC_NAME,0,"key","this is a message!");
return "send success!";
}
}

4.编写消费者

package com.qf.kafka.spring.boot.demo.consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class MyConsumer {
@KafkaListener(topics = "my-replicated-topic",groupId = "MyGroup1")
public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
String value = record.value();
System.out.println(value);
System.out.println(record);
//⼿动提交offset
ack.acknowledge();
}
}

5.消费者中配置消费主题、分区和偏移量

@KafkaListener(groupId = "testGroup", topicPartitions = {
@TopicPartition(topic = "topic1", partitions = {"0", "1"}),
@TopicPartition(topic = "topic2", partitions = "0",
partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
},concurrency = "3")//concurrency就是同组下的消费者个数，就是并发消费数，建议小于等于分区总数
public void listenGroupPro(ConsumerRecord<String, String> record, Acknowledgment ack) {
String value = record.value();
System.out.println(value);
System.out.println(record);
//手动提交offset
ack.acknowledge();
}

九、kafka集群中的controller、

rebalance、HW

1.controller

集群中谁来充当controller

每个broker启动时会向zk创建⼀个临时序号节点，获得的序号最⼩的那个broker将会作为集

群中的controller，负责这么⼏件事：

当集群中有⼀个副本的leader挂掉，需要在集群中选举出⼀个新的leader，选举的规则是从isr集合中最左边获得。
当集群中有broker新增或减少，controller会同步信息给其他broker
当集群中有分区新增或减少，controller会同步信息给其他broker

2.rebalance机制

前提：消费组中的消费者没有指明分区来消费
触发的条件：当消费组中的消费者和分区的关系发⽣变化的时候
分区分配的策略：在rebalance之前，分区怎么分配会有这么三种策略

- range：根据公示计算得到每个消费消费哪⼏个分区：前⾯的消费者是分区总数/消费者数量 +1,之后的消费者是分区总数/消费者数量
- 轮询：⼤家轮着来
- sticky：粘合策略，如果需要rebalance，会在之前已分配的基础上调整，不会改变之前的分配情况。如果这个策略没有开，那么就要进⾏全部的重新分配。建议开启。（如果之前使用了rebalance，在之前已分配的策略上进行再次分配）

3.HW和LEO

LEO是某个副本最后消息的消息位置（log-end-offset）

HW是已完成同步的位置。消息在写⼊broker时，且每个broker完成这条消息的同步后，hw

才会变化。在这之前消费者是消费不到这条消息的。在同步完成之后，HW更新之后，消费者

才能消费到这条消息，这样的⽬的是防⽌消息的丢失。

⼗、Kafka中的优化问题

1.如何防⽌消息丢失

⽣产者：1）使⽤同步发送 2）把ack设成1或者all，并且设置同步的分区数>=2
消费者：把⾃动提交改成⼿动提交

2.如何防⽌重复消费

在防⽌消息丢失的⽅案中，如果⽣产者发送完消息后，因为⽹络抖动，没有收到ack，但实际

上broker已经收到了。

此时⽣产者会进⾏重试，于是broker就会收到多条相同的消息，⽽造成消费者的重复消费。

怎么解决：

⽣产者关闭重试：会造成丢消息（不建议）
消费者解决⾮幂等性消费问题：

- 所谓的幂等性：多次访问的结果是⼀样的。对于rest的请求（get（幂等）、post（⾮幂等）、put（幂等）、delete（幂等））
- 解决⽅案：

- - 在数据库中创建联合主键，防⽌相同的主键创建出多条记录
  - 使⽤分布式锁，以业务id为锁。保证只有⼀条记录能够创建成功

3.如何做到消息的顺序消费

⽣产者：保证消息按顺序消费，且消息不丢失——使⽤同步的发送，ack设置成⾮0的值。

- （当上一条成功发送，下一条才能继续发送，保证消息的顺序性）

消费者：主题只能设置⼀个分区，消费组中只能有⼀个消费者

kafka的顺序消费使⽤场景不多，因为牺牲掉了性能，但是⽐如rocketmq在这⼀块有专⻔的功能已设计好。

4.如何解决消息积压问题

1）消息积压问题的出现

消息的消费者的消费速度远赶不上⽣产者的⽣产消息的速度，导致kafka中有⼤量的数据没有被消费。随着没有被消费的数据堆积越多，消费者寻址的性能会越来越差，最后导致整个kafka对外提供的服务的性能很差，从⽽造成其他服务也访问速度变慢，造成服务雪崩。

2）消息积压的解决⽅案

在这个消费者中，使⽤多线程，充分利⽤机器的性能进⾏消费消息。
通过业务的架构设计，提升业务层⾯消费的性能。
创建多个消费组，多个消费者，部署到其他机器上，⼀起消费，提⾼消费者的消费速度
创建⼀个消费者，该消费者在kafka另建⼀个主题，配上多个分区，多个分区再配上多个

消费者。该消费者将poll下来的消息，不进⾏消费，直接转发到新建的主题上。此时，新

的主题的多个分区的多个消费者就开始⼀起消费了。——不常⽤

5.实现延时队列的效果

1）应⽤场景

订单创建后，超过30分钟没有⽀付，则需要取消订单，这种场景可以通过延时队列来实现

2）具体⽅案

kafka中创建创建相应的主题
消费者消费该主题的消息（轮询）
消费者消费消息时判断消息的创建时间和当前时间是否超过30分钟（前提是订单没⽀付）

- 如果是：去数据库中修改订单状态为已取消
- 如果否：记录当前消息的offset，并不再继续消费之后的消息。等待1分钟后，再次向kafka拉取该offset及之后的消息，继续进⾏判断，以此反复。

⼗⼀、Kafka-eagle监控平台

1.搭建

去kafka-eagle官⽹下载压缩包
http://download.kafka-eagle.org/
分配⼀台虚拟机
虚拟机中安装jdk
解压缩kafka-eagle的压缩包
给kafka-eagle配置环境变量

export KE_HOME=/usr/local/kafka-eagle
export PATH=$PATH:$KE_HOME/bin

需要修改kafka-eagle内部的配置⽂件： vim system-config.properties
修改⾥⾯的zk的地址和mysql的地址
进⼊到bin中，通过命令来启动

./ke.sh start

2.平台的使⽤

国内服务器

文章版权归作者所有，未经允许请勿转载。

Spark+Flask新能源车数据分析与推荐系统实战：从0到1搭建完整项目

国内服务器

2个月前

290

【AI测试全栈：Vue核心】23、Vue3+ECharts AI测试可视化性能优化大全：从大数据渲染到错误监控

国内服务器

3个月前

340

DataInLong任务切换实践：从Kafka-A到Kafka-B的数据迁移指南

国内服务器

1个月前

230

【JavaWeb学习 | 第23篇】监听器、RBAC权限模型

国内服务器

3个月前

330

一篇文章速通kafka——day02

六、kafka的java客户端-⽣产者的实现

1.⽣产者的基本实现

2.⽣产者的同步发送消息

3.⽣产者的异步发送消息

4.⽣产者中的ack的配置

5.关于消息发送的缓冲区

七、Java客户端消费者的实现细节

1.消费者的基本实现

2.关于消费者⾃动提交和⼿动提交offset

3.⻓轮询poll消息

4.消费者的健康状态检查

5.指定分区和偏移量、时间消费

6.新消费组的消费offset规则

⼋、Springboot中使⽤Kafka

1.引⼊依赖

2.编写配置⽂件

3.编写消息⽣产者

4.编写消费者

5.消费者中配置消费主题、分区和偏移量

九、kafka集群中的controller、

rebalance、HW

1.controller

2.rebalance机制

3.HW和LEO

⼗、Kafka中的优化问题

1.如何防⽌消息丢失

2.如何防⽌重复消费

3.如何做到消息的顺序消费

4.如何解决消息积压问题

5.实现延时队列的效果

⼗⼀、Kafka-eagle监控平台

1.搭建

2.平台的使⽤

基于Doris的实时数据仓库建设：从理论到实践的完整指南

【理财类-01-04】20260321“微信”定期60天、90天、120天，倒退计算最适合买入的日期，在每月8日和每月20日准时到账还款（与理财系统一致）

相关文章

Spark+Flask新能源车数据分析与推荐系统实战：从0到1搭建完整项目

【AI测试全栈：Vue核心】23、Vue3+ECharts AI测试可视化性能优化大全：从大数据渲染到错误监控

DataInLong任务切换实践：从Kafka-A到Kafka-B的数据迁移指南

【JavaWeb学习 | 第23篇】监听器、RBAC权限模型

标签云