芝法酱学习笔记（2.6）——flink-cdc监听mysql binlog并同步数据至elastic- ...

一、需求背景

在有的项目中，尤其是进销存类的saas软件，一开始为了快速把产物做出来，并没有考虑缓存问题。而这类软件，有着复杂的业务逻辑。假如想在原先的代码中，添加redis缓存，改动面将非常大，还需要大量的测试工作。有些时间会有更离谱的环境，比如一些一些项目可能用JDK1.6写的，想要在这个框架下接入redis缓存，也会变得非常困难。
这时我们就会想到，能否像mysql的主从复制一样，监听mysql的binlog，对数据进行更新呢？Flink CDC就呼之欲出。
二、mysql环境搭建

需要留意的是，当前的flink-cdc，仅仅支持mysql8.0，8.4是完全不支持的。
由于我的mysql装的是8.4，为了方便起见，我们使用docker安装mysql8.0
2.1 docker-compose.yml

1. services:
2.   master:
3.     image: mysql:8.0.41
4.     container_name: mysql-8
5.     restart: always
6.     #mem_limit: 512M
7.     environment:
8.       MYSQL_ROOT_PASSWORD: study@2025
9.       TZ: Asia/Shanghai
10.     ports:
11.       - "3307:3306"
12.     volumes:
13.       - ./cfg/my.cnf:/etc/my.cnf
14.       - ./data:/var/lib/mysql
15.       - ./initdb:/docker-entrypoint-initdb.d
16.       - ./dump:/var/dump
17.       - ./log:/var/log
18.     networks:
19.       - mysql-cluster
20. networks:
21.   mysql-cluster:

复制代码

2.2 初始化sql

1. -- 创建复制用户
3. create role role_app;
4. GRANT SELECT,UPDATE,INSERT,DELETE ON *.* to role_app;
5. GRANT REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO role_app;
7. CREATE USER 'app'@'%' IDENTIFIED WITH caching_sha2_password by 'study@2025' DEFAULT ROLE role_app COMMENT 'app user';
10. FLUSH PRIVILEGES;
12. -- 创建两个数据库，用于测试
13. CREATE SCHEMA `shop-center`;
14. FLUSH TABLES WITH READ LOCK;

复制代码

2.3 留意点

起首把容器卷 - ./cfg/my.cnf:/etc/my.cnf的这一句注释掉，启动服务
而后使用下面语句，把配置文件粘出来

1. docker exec <id> cp /etc/my.cnf ./cfg/my.cnf

复制代码

之后把注释打开，再重新启动
三、工程搭建与pom引用

3.1 主模块pom引用

flink步伐不需要接入Spring框架，直接一个main就可运行。
但我们还想使用一些我们熟悉的接口，来操作redis和el。

1.                 <dependency>
2.             <groupId>org.apache.flink</groupId>
3.             <artifactId>flink-core</artifactId>
4.             <version>1.20.0</version>
5.         </dependency>
7.         <dependency>
8.             <groupId>org.apache.flink</groupId>
9.             <artifactId>flink-streaming-java</artifactId>
10.             <version>1.20.0</version>
11.         </dependency>
13.         <dependency>
14.             <groupId>org.apache.flink</groupId>
15.             <artifactId>flink-clients</artifactId>
16.             <version>1.20.0</version>
17.         </dependency>
19.         <dependency>
20.             <groupId>org.apache.flink</groupId>
21.             <artifactId>flink-runtime</artifactId>
22.             <version>1.20.0</version>
23.         </dependency>
24.         
25.                 <dependency>
26.             <groupId>org.apache.flink</groupId>
27.             <artifactId>flink-connector-mysql-cdc</artifactId>
28.             <version>3.3.0</version>
29.         </dependency>       

复制代码

3.2 common-data模块

一些entity数据，为了保持各模块共通，最好独立到一个common模块。
同时，我还会把redis和el-search的操作，在这个模块接入并封装
3.2.1 pom引用

1. <dependencies>
3.         <dependency>
4.             <groupId>org.yaml</groupId>
5.             <artifactId>snakeyaml</artifactId>
6.             <version>2.3</version>
7.         </dependency>
9.         <dependency>
10.             <groupId>co.elastic.clients</groupId>
11.             <artifactId>elasticsearch-java</artifactId>
12.             <version>8.17.0</version>
13.         </dependency>
15.         <dependency>
16.             <groupId>org.elasticsearch</groupId>
17.             <artifactId>elasticsearch-x-content</artifactId>
18.             <version>8.17.0</version>
19.         </dependency>
21.         <dependency>
22.             <groupId>cn.hutool</groupId>
23.             <artifactId>hutool-core</artifactId>
24.             <version>5.8.32</version>
25.         </dependency>
28.         <dependency>
29.             <groupId>org.projectlombok</groupId>
30.             <artifactId>lombok</artifactId>
31.             <optional>true</optional>
32.             <scope>provided</scope>
33.         </dependency>
36.         <dependency>
37.             <groupId>org.springframework.data</groupId>
38.             <artifactId>spring-data-redis</artifactId>
39.             <version>3.4.2</version>
40.         </dependency>
42.         <dependency>
43.             <groupId>com.alibaba.fastjson2</groupId>
44.             <artifactId>fastjson2-extension-spring6</artifactId>
45.             <version>2.0.54</version>
46.         </dependency>
48.         <dependency>
49.             <groupId>org.apache.commons</groupId>
50.             <artifactId>commons-pool2</artifactId>
51.             <version>2.12.1</version>
52.         </dependency>
54.         <dependency>
55.             <groupId>com.alibaba.fastjson2</groupId>
56.             <artifactId>fastjson2</artifactId>
57.             <version>2.0.54</version>
58.         </dependency>
60.         <dependency>
61.             <groupId>io.lettuce</groupId>
62.             <artifactId>lettuce-core</artifactId>
63.             <version>6.4.2.RELEASE</version>
64.         </dependency>
66.         <!-- Flink Redis Connector -->
67.         <!--        <dependency>-->
68.         <!--            <groupId>org.apache.bahir</groupId>-->
69.         <!--            <artifactId>flink-connector-redis_2.12</artifactId>-->
70.         <!--            <version>1.1.0</version>-->
71.         <!--        </dependency>-->
73.     </dependencies>

复制代码

3.2.2 一些根本的entity类

1. @Data
2. public class GenItemEntity{
3.     Long id;
4.     String name;
5.     Long price;
6.     String brand;
7.     String specification;
8.     Integer version;
9. }

复制代码

四、 redis操作和elsearch操作的封装

4.1 redis操作的封装

在pom上，接入spring-data-redis
而后，我们可以使用我们熟悉的RedisTemplate来操作redis

1. public class RedisConfig {
3.     public RedisConfig(){
4.         init();
5.     }
7.     protected FastJsonConfig redisFastJson(){
8.         FastJsonConfig config = new FastJsonConfig();
9.         config.setWriterFeatures(
10.                 JSONWriter.Feature.WriteNullListAsEmpty,
11.                 // 写入类名
12.                 JSONWriter.Feature.WriteClassName,
13.                 // 将 Boolean 类型的 null 转成 false
14.                 JSONWriter.Feature.WriteNullBooleanAsFalse,
15.                 JSONWriter.Feature.WriteEnumsUsingName);
16.         config.setReaderFeatures(
17.                 JSONReader.Feature.SupportClassForName,
18.                 // 支持autoType
19.                 JSONReader.Feature.SupportAutoType);
20.         return config;
21.     }
23.     protected FastJsonRedisSerializer fastJsonRedisSerializer(FastJsonConfig pFastJsonConfig) {
24.         FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer(Object.class);
25.         fastJsonRedisSerializer.setFastJsonConfig(pFastJsonConfig);
26.         return fastJsonRedisSerializer;
27.     }
29.     protected RedisConnectionFactory redisConnectionFactory(){
30.             // 这里最好读配置，我懒得搞了
31.         RedisStandaloneConfiguration redisConfiguration = new RedisStandaloneConfiguration("192.168.0.64",6379);
32.         redisConfiguration.setPassword("study@2025");
34.         GenericObjectPoolConfig<?> poolConfig = new GenericObjectPoolConfig<>();
35.         poolConfig.setMaxTotal(2);  // 最大连接数
36.         poolConfig.setMaxIdle(2);    // 最大空闲连接数
37.         poolConfig.setMinIdle(2);    // 最小空闲连接数
38.         poolConfig.setMaxWait(Duration.ofMillis(3000)); // 连接等待时间
40.         ClientResources clientResources = DefaultClientResources.create();
42.         LettucePoolingClientConfiguration lettucePoolingClientConfiguration = LettucePoolingClientConfiguration.builder()
43.                 .poolConfig(poolConfig)
44.                 .build();
46.         LettucePoolingClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder()
47.                 .clientResources(clientResources)
48.                 .commandTimeout(Duration.ofSeconds(5))
49.                 .poolConfig(poolConfig)
50.                 .build();
52.         LettuceConnectionFactory redisConnectionFactory = new LettuceConnectionFactory(redisConfiguration,lettucePoolingClientConfiguration);
53.         redisConnectionFactory.afterPropertiesSet(); // 初始化连接工厂
55.         return redisConnectionFactory;
56.     }
59.     protected RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory, FastJsonRedisSerializer pFastJsonRedisSerializer) {
60.         RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
61.         redisTemplate.setConnectionFactory(factory);
62.         redisTemplate.setEnableTransactionSupport(true);
63.         redisTemplate.setKeySerializer(new StringRedisSerializer());
64.         redisTemplate.setValueSerializer(pFastJsonRedisSerializer);
65.         redisTemplate.setHashKeySerializer(new StringRedisSerializer());
66.         redisTemplate.setHashValueSerializer(pFastJsonRedisSerializer);
67.         return redisTemplate;
68.     }
70.     protected void init(){
71.         mFastJsonConfig = redisFastJson();
72.         mFastJsonRedisSerializer = fastJsonRedisSerializer(mFastJsonConfig);
73.         mRedisConnectionFactory = redisConnectionFactory();
74.         mRedisTemplate = redisTemplate(mRedisConnectionFactory,mFastJsonRedisSerializer);
75.         mRedisTemplate.afterPropertiesSet();
76.     }
78.     private FastJsonConfig mFastJsonConfig;
79.     private FastJsonRedisSerializer mFastJsonRedisSerializer;
80.     private RedisConnectionFactory mRedisConnectionFactory;
81.     private RedisTemplate<String, Object> mRedisTemplate;
83.     public static RedisTemplate<String, Object> redisTemplate(){
84.         return Holder.INSTANCE.mRedisTemplate;
85.     }
87.     public static <T> String serialize(T entity){
88.         return JSON.toJSONString(entity,Holder.INSTANCE.mFastJsonConfig.getWriterFeatures());
89.     }
91.     private static class Holder {
92.         private static final RedisConfig INSTANCE = new RedisConfig();
93.     }
95. }

复制代码

4.2 elasticsearch操作的封装

由于el-search的毗连器，需要配置apikey，以及https，我们最好使用yml配置，并且把http_ca.crt放进该模块的resouce中。
在IDEA环境下，有可能找不到子模块的资源，这时在主模块引入子模块时，只需要如许配置即可：

1.         <dependency>
2.             <groupId>indi.zhifa.study2025</groupId>
3.             <artifactId>common-data</artifactId>
4.             <version>${project.version}</version>
5.             <scope>compile</scope>
6.         </dependency>

复制代码

留意，重点是<scope>compile</scope>

1. public class EsClientConfig {
3.     @Setter
4.     @Getter
5.     private String host;
7.     @Setter
8.     @Getter
9.     private Integer port;
11.     @Setter
12.     @Getter
13.     private String apiKey;
15. }

复制代码

1. public class ElasticSearchClientProvider {
3.     private EsClientConfig esClientConfig;
5.     private RestClientBuilder builder;
7.     public ElasticSearchClientProvider() {
8.         try{
9.             init();
10.         }catch (Exception e){
11.             e.printStackTrace();
12.         }
14.     }
16.     public void init() throws IOException {
18.         Yaml yaml = new Yaml();
19.         try (InputStream inputStream = FileUtil.class.getClassLoader().getResourceAsStream("el-config.yml")) {
20.             if (inputStream == null) {
21.                 throw new IllegalArgumentException("File not found: el-config.yml");
22.             }
23.             esClientConfig = yaml.loadAs(inputStream, EsClientConfig.class);
24.         } catch (Exception e) {
25.             throw new RuntimeException("Failed to load YAML file", e);
26.         }
28.         SSLContext sslContext;
29.         try (InputStream inputStream = FileUtil.class.getClassLoader().getResourceAsStream("http_ca.crt")){
30.             sslContext = TransportUtils.sslContextFromHttpCaCrt(inputStream);
31.         }catch (Exception e) {
32.             throw new RuntimeException("Failed to load http_ca.crt", e);
33.         }
35.         builder = RestClient.builder(
36.                         new HttpHost(esClientConfig.getHost(), esClientConfig.getPort(), "https") // 替换为你的Elasticsearch地址
37.                 ).setDefaultHeaders(new Header[]{
38.                         new BasicHeader("Authorization", "ApiKey " + esClientConfig.getApiKey())
39.                 })
40.                 .setFailureListener(new RestClient.FailureListener(){
41.                     @Override
42.                     public void onFailure(Node node) {
43.                         super.onFailure(node);
44.                     }
45.                 }).setHttpClientConfigCallback(hc->
46.                    hc.setSSLContext(sslContext)
47.                 );
48.     }
51.     public ElasticsearchClient get(){
52.         RestClient restClient = builder.build();
53.         ElasticsearchTransport transport = new RestClientTransport(
54.                 restClient, new JacksonJsonpMapper());
55.         ElasticsearchClient esClient = new ElasticsearchClient(transport);
56.         return esClient;
57.     }
59.     public static ElasticSearchClientProvider getInstance(){
60.         return Holder.INSTANCE;
61.     }
63.     private static class Holder {
64.         private static final ElasticSearchClientProvider INSTANCE = new ElasticSearchClientProvider();
65.     }
67. }

复制代码

五、 redis和elsearch的自定义sink编写

5.1 redis的sink编写

我们希望传入redis时，数据是被处置惩罚好的，redis的sink不需要处置惩罚任何逻辑，只管更新缓存和删除缓存。
5.1.1 RedisSinkCommand

1. public class RedisSinkCommand<T> {
2.     @Setter
3.     @Getter
4.     protected ERedisCommand command;
6.     @Setter
7.     @Getter
8.     protected long dua;
10.     @Setter
11.     @Getter
12.     protected  String key;
13.     @Setter
14.     @Getter
15.     protected  T value;
18.     public void initSet(String pKey, T pValue) {
19.         command = ERedisCommand.SET;
20.         dua = 300;
21.         key = pKey;
22.         value = pValue;
23.     }
25.     public void initDel(String pKey) {
26.         command = ERedisCommand.DEL;
27.         key = pKey;
28.     }
30. }

复制代码

1. public enum ERedisCommand {
2.     SET,
3.     DEL
4. }

复制代码

5.1.2 SpringDataRedisSink

1. @Slf4j
2. public class SpringDataRedisSink<T> implements Sink<RedisSinkCommand<T>> {
4.     @Override
5.     public SinkWriter<RedisSinkCommand<T>> createWriter(InitContext context) throws IOException {
6.         return null;
7.     }
9.     @Override
10.     public SinkWriter<RedisSinkCommand<T>> createWriter(WriterInitContext context){
11.         return new LettuceRedisSinkWriter();
12.     }
14.     class LettuceRedisSinkWriter implements SinkWriter<RedisSinkCommand<T>> {
16.         @Override
17.         public void write(RedisSinkCommand<T> pCmd, Context context) throws IOException, InterruptedException {
18.             RedisTemplate<String, Object> redisTemplate = RedisConfig.redisTemplate();
19.             switch (pCmd.getCommand()){
20.                 case SET-> {
21.                     redisTemplate.opsForValue().set(pCmd.getKey(),pCmd.getValue(),pCmd.getDua());
22.                 }
23.                 case DEL -> {
24.                     redisTemplate.delete(pCmd.getKey());
25.                 }
26.             }
27.         }
29.         @Override
30.         public void flush(boolean endOfInput) throws IOException, InterruptedException {
32.         }
34.         @Override
35.         public void close() throws Exception {
37.         }
38.     }
40. }

复制代码

5.2 elasticsearch的sink编写

elasticsearch的sink与redis的要求一致，在sink中不关心业务逻辑
5.2.1 ElCommand

1. @Data
2. public class ElCommand<T> {
3.     protected EElCommand command;
4.     protected String index;
5.     protected T entity;
6.     protected String id;
7. }

复制代码

1. public enum EElCommand {
2.     CREATE,UPDATE,DELETE
3. }

复制代码

5.2.2 ElSearchSink

1. public class ElSearchSink<T> implements Sink<ElCommand<T>> {
2.     @Override
3.     public SinkWriter<ElCommand<T>> createWriter(InitContext context) throws IOException {
4.         return null;
5.     }
7.     @Override
8.     public SinkWriter<ElCommand<T>> createWriter(WriterInitContext context){
9.         return new ElSearchSink.ElSearchSinkWriter();
10.     }
12.     class ElSearchSinkWriter implements SinkWriter<ElCommand<T>> {
14.         @Override
15.         public void write(ElCommand<T> pCmd, Context context) throws IOException, InterruptedException {
16.             ElasticSearchClientProvider elasticSearchClientProvider = ElasticSearchClientProvider.getInstance();
17.             ElasticsearchClient elClient =  elasticSearchClientProvider.get();
18.             String index = pCmd.getIndex();
19.             String id = pCmd.getId();
20.             T entity = pCmd.getEntity();
22.             switch (pCmd.getCommand()){
23.                 case CREATE,UPDATE -> {
24.                     elClient.index(i->i.index(index).id(id).document(entity));
25.                 }
26.                 case DELETE -> {
27.                     elClient.delete(d->d.index(index).id(id));
28.                 }
29.             }
30.         }
32.         @Override
33.         public void flush(boolean endOfInput) throws IOException, InterruptedException {
35.         }
37.         @Override
38.         public void close() throws Exception {
40.         }
41.     }
42. }

复制代码

六、主函数编写

1. public class FlinkMain {
3.     public static void main(String[] args) throws Exception {
4.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
5.                 .hostname("192.168.0.64")
6.                 .port(3307)
7.                 .databaseList("shop-center") // set captured database
8.                 .tableList("shop-center.item") // set captured table
9.                 .username("app")
10.                 .password("study@2025")
11.                 .serverTimeZone("Asia/Shanghai")
12.                 .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
13.                 .startupOptions(StartupOptions.latest())
14.                 .includeSchemaChanges(true)
15.                 .build();
17. //        FlinkJedisPoolConfig jedisConfig = new FlinkJedisPoolConfig.Builder()
18. //                .setHost("192.168.0.64") // 替换为 Redis 主机
19. //                .setPort(6379) // Redis 端口
20. //                .setPassword("ilv0404@1314") // 如果有密码，设置密码
21. //                .build();
23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
24.         env.setParallelism(1);
26. //        DataStream<BinlogInfo> mysqlStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source")
27. //                .map(str->{
28. //                    BinlogInfo res =JSONObject.parseObject(str, BinlogInfo.class);
29. //                    return res;
30. //                    }
31. //                 ).filter(bi->bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d"));
32. //
33. //        mysqlStream.addSink(new RedisSink(jedisConfig,new RedisItemMapper()));
35.         DataStream<RedisSinkCommand<GenItemEntity>> newMysqlStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source to redis")
36.                         .map(str->JSONObject.parseObject(str, new TypeReference<BinlogInfo<GenItemEntity>>() {
37.                         }), TypeInformation.of(new TypeHint<BinlogInfo<GenItemEntity>>() {}))
38.                 .filter(bi->bi.getSource().getTable().equals("item") &&  (bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d")))
39.                 .map(bi->{
40.                     String op = bi.getOp();
41.                     GenItemEntity itemEntity = bi.getAfter();
42.                     String key = "item:"+itemEntity.getId();
43.                     switch (op){
44.                         case "c","u"->{
45.                             RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
46.                             redisSinkCommand.initSet(key,itemEntity);
47.                             return redisSinkCommand;
48.                         }
49.                         case "d" ->{
50.                             RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
51.                             redisSinkCommand.initDel(key);
52.                             return redisSinkCommand;
53.                         }
54.                         default -> {
55.                             RedisSinkCommand<GenItemEntity> redisSinkCommand = new RedisSinkCommand();
56.                             redisSinkCommand.initDel(key);
57.                             return redisSinkCommand;
58.                         }
59.                     }
60.                 },TypeInformation.of(new TypeHint<RedisSinkCommand<GenItemEntity>>() {}));
62.         newMysqlStream.sinkTo(new SpringDataRedisSink<GenItemEntity>());
64.         DataStream<ElCommand<GenItemEntity>> mySqlToElStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"Mysql source to el")
65.                 .map(str->JSONObject.parseObject(str, new TypeReference<BinlogInfo<GenItemEntity>>() {})
66.                         , TypeInformation.of(new TypeHint<BinlogInfo<GenItemEntity>>() {}))
67.         .filter(bi->bi.getSource().getTable().equals("item") &&  (bi.getOp().equals("c")||bi.getOp().equals("u")||bi.getOp().equals("d")))
68.                 .map(bi->{
69.                     ElCommand elCommand = new ElCommand();
70.                     GenItemEntity itemEntity = bi.getAfter();
71.                     elCommand.setId(itemEntity.getId().toString());
72.                     elCommand.setEntity(itemEntity);
73.                     elCommand.setIndex("item_npc");
74.                     String op = bi.getOp();
75.                     switch (op){
76.                         case "c"->elCommand.setCommand(EElCommand.CREATE);
77.                         case "u"->elCommand.setCommand(EElCommand.UPDATE);
78.                         case "d"->elCommand.setCommand(EElCommand.DELETE);
79.                     }
80.                     return elCommand;
81.                 },TypeInformation.of(new TypeHint<ElCommand<GenItemEntity>>() {}));
82.         mySqlToElStream.sinkTo(new ElSearchSink());
83.         env.execute();
84.     }
85. }

复制代码

七、代码展示

请道友移步码云
八、相关实践的思考

8.1 redis相关

我这里的代码，仅仅是学习用的。在真实项目中，redis缓存的更新，通常源于查询时，假如发现缓存中没有数据，则查mysql，并把缓存数据加入redis。假如监听到表数据的更改或删除，则直接删除相应缓存，等待查询时重新加入缓存。固然，如许做在同一数据并发访问时，会有重复设置缓存的可能性，我们把这种征象叫缓存穿透。可以在更新缓存前，用redisson加个锁，防止重复读取mysql并更新redis。

1. public class CacheService {
2.     @Autowired
3.     private RedissonClient redissonClient;
4.     @Autowired
5.     private RedisTemplate<String, Object> redisTemplate;
6.     @Autowired
7.     private DataRepository dataRepository;
9.     public Object getData(String key) {
10.         // 第一次检查缓存
11.         Object value = redisTemplate.opsForValue().get(key);
12.         if (value != null) {
13.             return value;
14.         }
16.         RLock lock = redissonClient.getLock(key + ":LOCK");
17.         try {
18.             // 尝试加锁，设置锁超时时间防止死锁
19.             if (lock.tryLock(5, 30, TimeUnit.SECONDS)) {
20.                 try {
21.                     // 双重检查缓存
22.                     value = redisTemplate.opsForValue().get(key);
23.                     if (value != null) {
24.                         return value;
25.                     }
26.                     // 查询数据库
27.                     Object dbData = dataRepository.findById(key);
28.                     // 更新缓存，设置合理过期时间
29.                     redisTemplate.opsForValue().set(key, dbData, 1, TimeUnit.HOURS);
30.                     return dbData;
31.                 } finally {
32.                     lock.unlock();
33.                 }
34.             } else {
35.                 // 未获取到锁，短暂等待后重试
36.                 Thread.sleep(100);
37.                 return redisTemplate.opsForValue().get(key);
38.             }
39.         } catch (InterruptedException e) {
40.             Thread.currentThread().interrupt();
41.             throw new RuntimeException("获取锁失败", e);
42.         }
43.     }
44. }

复制代码

8.2 es相关

对于es，着实更新数据不建议接纳这种方式。由于es中需要反范式设计，不可能用1张表的数据做es查询数据的。
对于电商体系的商品查询，我们可以在商品上架的时间更新es。并且商品商家状态下，不允许修改商品。商品下架时，删除es的数据。想要修改商品数据，可以先下架，再修改，而后上架。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。