ToB企服应用市场:ToB评测及商务社交产业平台

标题: 芝法酱学习笔记（2.6）——flink-cdc监听mysql binlog并同步数据至elastic- [打印本页]
作者: 火影    时间: 昨天 20:51
标题: 芝法酱学习笔记（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企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4