当前位置：首页 > article >正文

springboot第75集：kafka，线程，进程，容器化服务，线程池

article 2024/10/11 11:04:23

消息中间件在异步通信中⽤的最多，很多业务流程中，如果所有步骤都同步进⾏可能会导致核⼼流程耗时⾮常⻓，更重

要的是所有步骤都同步进⾏⼀旦⾮核⼼步骤失败会导致核⼼流程整体失败，因此在很多业务流程中Kafka就充当了异步

通信⻆⾊。

⼤规模分布式系统中的机器⾮常多⽽且分散在不同机房中，分布式系统带来的⼀个明显问题就是业务⽇志的查看、追踪

和分析等⾏为变得⼗分困难，对于集群规模在百台以上的系统，查询线上⽇志很恐怖。

为了应对这种场景统⼀⽇志系统应运⽽⽣，⽇志数据都是海量数据，通常为了不给系统带来额外负担⼀般会采⽤异步上

报，这⾥Kafka以其⾼吞吐量在⽇志处理中得到了很好的应⽤。

随着据量的增加，离线的计算会越来越慢，难以满⾜⽤户在某些场景下的实时性要求，因此很多解决⽅案中引⼊了实时

计算。

很多时候，即使是海量数据，我们也希望即时去查看⼀些数据指标，实时流计算应运⽽⽣。

实时流计算有两个特点，⼀个是实时，随时可以看数据；另⼀个是流。实时流计算有两个特点，⼀个是实时，随时可以看数据；另⼀个是流。

push模式由broker决定消息推送的速率，对于不同消费速率的consumer就不太好处理了。

消息系统都致⼒于让consumer以最⼤的速率最快速的消费消息，push模式下，当broker推送的速率远⼤于consumer消息系统都致⼒于让consumer以最⼤的速率最快速的消费消息，push模式下，当broker推送的速率远⼤于consumer

消费的速率时，consumer恐怕就要崩溃了。

例如消息发送设置了重试机制，并且异步发送，消息A和B设置相同的key，业务上A先发，B后发，由于⽹络或者其他

原因A发送失败，B发送成功；A由于发送失败就会重试且重试成功，这时候消息顺序B在前A在后，与业务发送顺序不

⼀致，如果需要解决这个问题，需要设置参数 max.in.flight.requests.per.connection=1 ，其含义是限制客户

端在单个连接上能够发送的未响应请求的个数，设置此值是1表示kafka broker在响应请求之前client不能再向同⼀个

broker发送请求，这个参数默认值是5

kafka的消息是不断追加到⽂件中的，这个特性使 kafka 可以充分利⽤磁盘的顺序读写性能

顺序读写不需要硬盘磁头的寻道时间，只需很少的扇区旋转时间，所以速度远快于随机读写

Kafka 可以配置异步刷盘，不开启同步刷盘，异步刷盘不需要等写⼊磁盘后返回消息投递的 ACK，所以它提⾼了消息发

送的吞吐量，降低了请求的延时

传统的 IO 流程，需要先把数据拷⻉到内核缓冲区，再从内核缓冲拷⻉到⽤户空间，应⽤程序处理完成以后，再拷⻉回

内核缓冲区内核缓冲区

这个过程中发⽣了多次数据拷⻉

将数据保存到内存中的 Map（或其他数据结构）后，服务重启时这些数据不会自动保留，因为内存中的数据在进程结束时会被清除。要保留数据，可以考虑以下几种方法：

1. 持久化存储

将数据从 Map 持久化到文件或数据库中，以便在服务重启时能够恢复。

a. 使用文件存储

可以将 Map 的数据序列化并写入文件，例如使用 JSON 格式：

import java.io.*;
import java.util.HashMap;
import java.util.Map;

public class DataStorage {
    private Map<String, String> dataMap = new HashMap<>();

    // 保存数据到文件
    public void saveToFile() throws IOException {
        try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("data.ser"))) {
            oos.writeObject(dataMap);
        }
    }

    // 从文件加载数据
    public void loadFromFile() throws IOException, ClassNotFoundException {
        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream("data.ser"))) {
            dataMap = (Map<String, String>) ois.readObject();
        }
    }
}

// 使用 Redis 的 Java 客户端（如 Jedis）存储 Map 数据
import redis.clients.jedis.Jedis;

// 保存数据到 Redis
public void saveToRedis() {
    try (Jedis jedis = new Jedis("localhost")) {
        for (Map.Entry<String, String> entry : dataMap.entrySet()) {
            jedis.set(entry.getKey(), entry.getValue());
        }
    }
}

// 从 Redis 加载数据
public void loadFromRedis() {
    try (Jedis jedis = new Jedis("localhost")) {
        for (String key : jedis.keys("*")) {
            String value = jedis.get(key);
            dataMap.put(key, value);
        }
    }
}

定期将 Map 中的数据快照保存到文件或数据库中，以便在服务重启时快速恢复。

在服务启动时，添加加载数据的逻辑，从持久化存储中读取数据并填充 Map。