---
url: /notes/面试/真实面经/诺安科技/index.md
---
## 数据库和redis信息怎么保持一致

1.先更新缓存，在更新数据库：

2.先更新数据库，再更新缓存

3.先删除缓存，再更新数据库，再缓存新数据

4.先更新数据库，再删除缓存，再缓存新数据

5.缓存双删：先删除缓存，再更新数据库，再延迟删除缓存

6.binlog异步更新缓存

保证实时一致性：先更新数据库，再删除缓存，短期数据不一致

保证最终一致性：先更新数据库，在binlog+消息队列

**保证强一致性**：

分布式读写锁

![image-20250219224304563](https://note-1259190304.cos.ap-chengdu.myqcloud.com/noteimage-20250219224304563.png)

写操作：

* 获取写锁（检查是否有写锁或读锁）
* 更新数据库
* 删除缓存
* 释放写锁

读操作：

* 获取读锁（检查是否有写锁）.
* 查询缓存：如果命中缓存，释放读锁，返回结果。如果未命中，读取数据库，并将数据更新到缓存。
* 释放读锁

## kafak消息拥堵

遇到的场景：一个数据点位有两个数采地址，每次只有一个地址有数据；通过另一个参数来确认取哪一个的值，刚好这个参数的值是空，取不到数据，导致kafka消息无法手动确认，消息就堵死了；

监控图形上就只展示到前10分钟之前的图形数据，就不动了

groups命令查询消息偏移数量

## kafka保证消息不丢失，重复消费

生产者：

消息不丢失：

确保消息写入所有副本（acks=all）

发送失败可以重试

重复消费：

开启幂等性

消费者：

消息不丢失：

手动提交偏移量

重试

重复消费：

保证多次消费消息相同

## kafka自动确认机制

确认模式类型：

1. RECORD: 每处理一条消息提交一次
2. BATCH: 每处理一批消息提交一次
3. TIME: 按时间间隔提交
4. COUNT: 按消息数量提交
5. COUNT\_TIME: 按时间或数量提交，先到先提交
6. MANUAL: 手动提交
7. MANUAL\_IMMEDIATE: 手动立即提交

## es 100w索引设计

比如日志信息，怎么设计索引，肯定不能都在一个索引上

根据时间动态创建索引，每天或者每月

## es和数据库的数据同步

logstash配置文件

关键字

input，filter，output

```properties
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  jdbc {
    jdbc_driver_library => "C:\Users\Administrator\AppData\Roaming\JetBrains\DataGrip2021.1\jdbc-drivers\MySQL ConnectorJ\8.0.25\mysql\mysql-connector-java\8.0.25\mysql-connector-java-8.0.25.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/xiaofei_site_search"
    jdbc_user => "root"
    jdbc_password => "root"
    use_column_value => true
    tracking_column => "updateTime"
    tracking_column_type => "timestamp"
    parameters => { "isDelete" => 1 }
    schedule => "*/5 * * * * *"
    jdbc_default_timezone => "Asia/Shanghai"
    # 记录上次执行位置
    last_run_metadata_path => "D:/tools/Elastic_Search/logstash-7.17.12/data/plugins/last_run_metadata_file"

    statement => "
      SELECT 
        id,
        fileName,
        description,
        file_type as fileType,
        file_extension as fileExtension,
        file_content as textContent,
        description,
        upload_time as uploadTime,
        upload_user_id as uploadUserId,
        biz,
        is_public as isPublic,
        created_at as createdAt,
        updated_at as updatedAt,
        status as status
      FROM file where upload_time > :sql_last_value and upload_time < now() order by upload_time desc
    "
  }
}

filter{
  mutate {
    rename => {
      "file_name" => "fileName"
      "file_path" => "filePath"
      "file_type" => "filetype"
      "file_extension" => "fileExtension"
      "file_content" => "textContent"
      "upload_time" => "uploadTime"
      "upload_user_id" => "uploadUserId"
      "is_public" => "isPublic"
      "created_at" => "createdAt"
      "updated_at" => "updatedAt"
      "description" => "description"
      "status" => "status"
    }
  }
}

output {
  stdout { codec => rubydebug }
  elasticsearch {
        hosts => "127.0.0.1:9200"
        index => "file_v1"
        document_id => "%{id}"
    }
}
```

## hibernate 二三级缓存

一次缓存：session关联；查询实体时优先查询

二级缓存：sessionFactory关联，跨越多个session；一级缓存没有时，查询并将查询结果存入二级缓存，数据更新、删除、新增时二级缓存同步更新

三级缓存（查询缓存）：查询结果存储在查询缓存

查询数据的过程：

1. 先查一级缓存（Session）
   ↓ 未命中
2. 再查二级缓存（SessionFactory）
   ↓ 未命中
3. 最后查询缓存（Query Cache）
   ↓ 未命中
4. 查询数据库

```java
// Hibernate使用CacheKey类来标识缓存对象
public class CacheKey {
    private final String entityName;      // 实体类名
    private final Serializable identifier; // 主键值
    private final EntityMode entityMode;   // 实体模式
    private final String tenantId;        // 租户ID（如果有）
    
    // 示例
    @Entity
    public class User {
        @Id
        private Long id;   // identifier
        private String name;
        
        // entityName = "com.example.User"
        // identifier = id值
    }
}
```

一级缓存使用EntityKey作为key

EntityKey = entityName + id + entityMode

二级缓存key组成：

regionName + key

regionName默认是实体类全限定名

key是CacheKey对象

查询缓存key组成：

1. 查询SQL

2. 查询参数值

3. 首个返回行数

4. 当前使用的Filters