day11-物流信息微服务

shuhongfan
1 parent 97b0ec46
Showing 12 changed files with 457 additions and 2 deletions
sl-express-ms-transport-info-service/pom.xml
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/config/BloomFilterConfig.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/config/CaffeineConfig.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/config/MyRedisCacheManager.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/config/RedisConfig.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/config/RedissonConfiguration.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/controller/TransportInfoController.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/mq/RedisMessageListener.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/mq/TransportInfoMQListener.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/service/BloomFilterService.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/service/impl/BloomFilterServiceImpl.java
sl-express-ms-transport-info-service/src/main/java/com/sl/transport/info/service/impl/TransportInfoServiceImpl.java
@@ -49,6 +49,10 @@
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
         <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson</artifactId>
+        </dependency>
+        <dependency>
             <groupId>com.alibaba.cloud</groupId>
             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
         </dependency>
+package com.sl.transport.info.config;
+
+import lombok.Getter;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * 布隆过滤器相关配置
+ */
+@Getter
+@Configuration
+public class BloomFilterConfig {
+
+    /**
+     * 名称，默认：sl-bloom-filter
+     */
+    @Value("${bloom.name:sl-bloom-filter}")
+    private String name;
+
+    /**
+     * 布隆过滤器长度，最大支持Integer.MAX_VALUE*2，即：4294967294，默认：1千万
+     */
+    @Value("${bloom.expectedInsertions:10000000}")
+    private long expectedInsertions;
+
+    /**
+     * 误判率，默认：0.05
+     */
+    @Value("${bloom.falseProbability:0.05d}")
+    private double falseProbability;
+
+}
+package com.sl.transport.info.config;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.sl.transport.info.domain.TransportInfoDTO;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+
+/**
+ * Caffeine缓存配置
+ */
+@Configuration
+public class CaffeineConfig {
+
+    @Value("${caffeine.init}")
+    private Integer init;
+
+    @Value("${caffeine.max}")
+    private Integer max;
+
+    @Bean
+    public Cache<String, TransportInfoDTO> transportInfoCache() {
+        return Caffeine.newBuilder()
+                .initialCapacity(init)
+                .maximumSize(max)
+                .build();
+    }
+
+}
+package com.sl.transport.info.config;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.RandomUtil;
+import org.springframework.data.redis.cache.RedisCache;
+import org.springframework.data.redis.cache.RedisCacheConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.cache.RedisCacheWriter;
+
+import java.time.Duration;
+
+/**
+ * 自定义CacheManager，用于设置不同的过期时间，防止雪崩问题的发生
+ */
+public class MyRedisCacheManager extends RedisCacheManager {
+
+    public MyRedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration defaultCacheConfiguration) {
+        super(cacheWriter, defaultCacheConfiguration);
+    }
+
+    @Override
+    protected RedisCache createRedisCache(String name, RedisCacheConfiguration cacheConfig) {
+        //获取到原有过期时间
+        Duration duration = cacheConfig.getTtl();
+        if (ObjectUtil.isNotEmpty(duration)) {
+            //在原有时间上随机增加1~10分钟
+            Duration newDuration = duration.plusMinutes(RandomUtil.randomInt(1, 11));
+            cacheConfig = cacheConfig.entryTtl(newDuration);
+        }
+        return super.createRedisCache(name, cacheConfig);
+    }
+}
+package com.sl.transport.info.config;
+
+import io.lettuce.core.dynamic.RedisCommandFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.cache.RedisCacheConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.cache.RedisCacheWriter;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+import java.time.Duration;
+
+@Configuration
+public class RedisConfig {
+
+    public static final String CHANNEL_TOPIC = "sl-express-ms-transport-info-caffeine";
+
+    /**
+     * 存储的默认有效期时间，单位：小时
+     */
+    @Value("${redis.ttl:1}")
+    private Integer redisTtl;
+
+    @Bean
+    public RedisCacheManager redisCacheManager(RedisTemplate redisTemplate) {
+        // 默认配置
+        RedisCacheConfiguration defaultCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig()
+                // 设置key的序列化方式为字符串
+                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
+                // 设置value的序列化方式为json格式
+                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))
+                .disableCachingNullValues() // 不缓存null
+                .entryTtl(Duration.ofHours(redisTtl));  // 默认缓存数据保存1小时
+
+        // 构redis缓存管理器
+//        RedisCacheManager redisCacheManager = RedisCacheManager.RedisCacheManagerBuilder
+//                .fromConnectionFactory(redisTemplate.getConnectionFactory())
+//                .cacheDefaults(defaultCacheConfiguration)
+//                .transactionAware() // 只在事务成功提交后才会进行缓存的put/evict操作
+//                .build();
+
+        RedisCacheWriter redisCacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(redisTemplate.getConnectionFactory());
+        MyRedisCacheManager myRedisCacheManager = new MyRedisCacheManager(redisCacheWriter, defaultCacheConfiguration);
+        myRedisCacheManager.setTransactionAware(true);
+        return myRedisCacheManager;
+    }
+
+    /**
+     * 配置订阅，用于解决Caffeine一致性的问题
+     *
+     * @param redisCommandFactory 链接工厂
+     * @param messageListenerAdapter 消息监听器
+     * @return 消息监听容器
+     */
+    @Bean
+    public RedisMessageListenerContainer container(
+            RedisConnectionFactory redisCommandFactory,
+            MessageListenerAdapter messageListenerAdapter
+    ) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(redisCommandFactory);
+        container.addMessageListener(messageListenerAdapter, new ChannelTopic(CHANNEL_TOPIC));
+        return container;
+    }
+
+
+}
+package com.sl.transport.info.config;
+
+
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.util.StrUtil;
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+import org.redisson.config.SingleServerConfig;
+import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.Resource;
+
+@Configuration
+public class RedissonConfiguration {
+
+    @Resource
+    private RedisProperties redisProperties;
+
+    @Bean
+    public RedissonClient redissonSingle() {
+        Config config = new Config();
+        SingleServerConfig serverConfig = config.useSingleServer()
+                .setAddress("redis://" + redisProperties.getHost() + ":" + redisProperties.getPort());
+        if (null != (redisProperties.getTimeout())) {
+            serverConfig.setTimeout(1000 * Convert.toInt(redisProperties.getTimeout().getSeconds()));
+        }
+        if (StrUtil.isNotEmpty(redisProperties.getPassword())) {
+            serverConfig.setPassword(redisProperties.getPassword());
+        }
+        return Redisson.create(config);
+    }
+
+}
 package com.sl.transport.info.controller;
  
+import com.github.benmanes.caffeine.cache.Cache;
 import com.sl.transport.common.exception.SLException;
 import com.sl.transport.common.util.BeanUtil;
 import com.sl.transport.common.util.ObjectUtil;
+import com.sl.transport.info.config.CaffeineConfig;
 import com.sl.transport.info.domain.TransportInfoDTO;
 import com.sl.transport.info.entity.TransportInfoEntity;
 import com.sl.transport.info.enums.ExceptionEnum;
+import com.sl.transport.info.service.BloomFilterService;
 import com.sl.transport.info.service.TransportInfoService;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiImplicitParam;
 import io.swagger.annotations.ApiImplicitParams;
 import io.swagger.annotations.ApiOperation;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -26,6 +30,12 @@ public class TransportInfoController {
     @Resource
     private TransportInfoService transportInfoService;
  
+    @Resource
+    public Cache<String, TransportInfoDTO> transportInfoCache;
+
+    @Autowired
+    private BloomFilterService bloomFilterService;
+
     /**
      * 根据运单id查询运单信息
      *
@@ -38,10 +48,23 @@ public class TransportInfoController {
     @ApiOperation(value = "查询", notes = "根据运单id查询物流信息")
     @GetMapping("{transportOrderId}")
     public TransportInfoDTO queryByTransportOrderId(@PathVariable("transportOrderId") String transportOrderId) {
-        TransportInfoEntity transportInfoEntity = transportInfoService.queryByTransportOrderId(transportOrderId);
-        if (ObjectUtil.isNotEmpty(transportInfoEntity)) {
+//        如果布隆过滤器中不存在，无需缓存命中，直接返回即可
+        boolean contains = bloomFilterService.contains(transportOrderId);
+        if (!contains) {
+            throw new SLException(ExceptionEnum.NOT_FOUND);
+        }
+
+        TransportInfoDTO transportInfoDTO = transportInfoCache.get(transportOrderId, s -> {
+//            未命中，查询MongoMD
+            TransportInfoEntity transportInfoEntity = this.transportInfoService.queryByTransportOrderId(transportOrderId);
+//            转换成DTO
             return BeanUtil.toBean(transportInfoEntity, TransportInfoDTO.class);
+        });
+
+        if (ObjectUtil.isNotEmpty(transportInfoDTO)) {
+            return transportInfoDTO;
         }
+
         throw new SLException(ExceptionEnum.NOT_FOUND);
     }
  
+package com.sl.transport.info.mq;
+
+import cn.hutool.core.convert.Convert;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.sl.transport.info.domain.TransportInfoDTO;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * redis消息监听，解决Caffeine一致性的问题
+ */
+@Component
+public class RedisMessageListener extends MessageListenerAdapter {
+
+    @Resource
+    public Cache<String, TransportInfoDTO> transportInfoCache;
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+//        获取到消息中的运单ID
+        String transportOrderId = Convert.toStr(message);
+
+//        将本JVM中的缓存删除掉
+        transportInfoCache.invalidate(transportOrderId);
+    }
+}
 package com.sl.transport.info.mq;
  
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
+import com.sl.ms.transport.api.OrganFeign;
 import com.sl.transport.common.constant.Constants;
+import com.sl.transport.common.vo.TransportInfoMsg;
+import com.sl.transport.domain.OrganDTO;
+import com.sl.transport.info.entity.TransportInfoDetail;
+import com.sl.transport.info.service.TransportInfoService;
 import org.springframework.amqp.core.ExchangeTypes;
 import org.springframework.amqp.rabbit.annotation.Exchange;
 import org.springframework.amqp.rabbit.annotation.Queue;
 import org.springframework.amqp.rabbit.annotation.QueueBinding;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
  
@@ -15,6 +24,13 @@ import org.springframework.transaction.annotation.Transactional;
 @Component
 public class TransportInfoMQListener {
  
+    @Autowired
+    private OrganFeign organFeign;
+
+    @Autowired
+    private TransportInfoService transportInfoService;
+
+
     @RabbitListener(bindings = @QueueBinding(
             value = @Queue(name = Constants.MQ.Queues.TRANSPORT_INFO_APPEND),
             exchange = @Exchange(name = Constants.MQ.Exchanges.TRANSPORT_INFO, type = ExchangeTypes.TOPIC),
@@ -23,6 +39,29 @@ public class TransportInfoMQListener {
     @Transactional
     public void listenTransportInfoMsg(String msg) {
         //{"info":"您的快件已到达【$organId】", "status":"运输中", "organId":90001, "transportOrderId":"SL920733749248" , "created":1653133234913}
+        TransportInfoMsg transportInfoMsg = JSONUtil.toBean(msg, TransportInfoMsg.class);
+        Long organId = transportInfoMsg.getOrganId();
+        String transportOrderId = Convert.toStr(transportInfoMsg.getTransportOrderId());
+        String info = transportInfoMsg.getInfo();
+
+//        查询机构信息
+        if (StrUtil.contains(info, "$organId")) {
+            OrganDTO organDTO = organFeign.queryById(organId);
+            if (organDTO == null) {
+                return;
+            }
+            info = StrUtil.replace(info, "$organId", organDTO.getName());
+        }
+
+//        封装Detail对象
+        TransportInfoDetail infoDetail = TransportInfoDetail.builder()
+                .info(info)
+                .status(transportInfoMsg.getStatus())
+                .created(transportInfoMsg.getCreated())
+                .build();
+
+//        存储到MongoDB
+        transportInfoService.saveOrUpdate(transportOrderId, infoDetail);
  
     }
 }
+package com.sl.transport.info.service;
+
+/**
+ * 布隆过滤器服务
+ */
+public interface BloomFilterService {
+
+    /**
+     * 初始化布隆过滤器
+     */
+    void init();
+
+    /**
+     * 向布隆过滤器中添加数据
+     *
+     * @param obj 待添加的数据
+     * @return 是否成功
+     */
+    boolean add(Object obj);
+
+    /**
+     * 判断数据是否存在
+     *
+     * @param obj 数据
+     * @return 是否存在
+     */
+    boolean contains(Object obj);
+
+}
+package com.sl.transport.info.service.impl;
+
+import com.sl.transport.info.config.BloomFilterConfig;
+import com.sl.transport.info.service.BloomFilterService;
+import org.redisson.api.RBloomFilter;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+
+@Service
+public class BloomFilterServiceImpl implements BloomFilterService {
+
+    @Autowired
+    private RedissonClient redissonClient;
+
+    @Autowired
+    private BloomFilterConfig bloomFilterConfig;
+
+    private RBloomFilter<Object> getBloomFilter() {
+        return redissonClient.getBloomFilter(this.bloomFilterConfig.getName());
+    }
+
+    @Override
+    @PostConstruct
+    public void init() {
+        RBloomFilter<Object> bloomFilter = getBloomFilter();
+        bloomFilter.tryInit(
+                bloomFilterConfig.getExpectedInsertions(),
+                bloomFilterConfig.getFalseProbability());
+    }
+
+    @Override
+    public boolean add(Object obj) {
+        return getBloomFilter().add(obj);
+    }
+
+    @Override
+    public boolean contains(Object obj) {
+        return getBloomFilter().contains(obj);
+    }
+
+}
+package com.sl.transport.info.service.impl;
+
+import cn.hutool.core.collection.ListUtil;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.sl.transport.common.util.ObjectUtil;
+import com.sl.transport.info.config.RedisConfig;
+import com.sl.transport.info.entity.TransportInfoDetail;
+import com.sl.transport.info.entity.TransportInfoEntity;
+import com.sl.transport.info.service.TransportInfoService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cache.annotation.CachePut;
+import org.springframework.cache.annotation.Cacheable;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class TransportInfoServiceImpl implements TransportInfoService {
+
+    @Autowired
+    private MongoTemplate mongoTemplate;
+    @Autowired
+    private Cache transportInfoCache;
+    private StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    private BloomFilterServiceImpl bloomFilterService;
+
+    /**
+     * 如果运单数据不存在，就新增，否则更新数据
+     *
+     * @param transportOrderId 运单id
+     * @param infoDetail       信息详情
+     * @return 运输信息数据
+     */
+    @Override
+    @CachePut(value = "transport-info",key = "#p0")
+    public TransportInfoEntity saveOrUpdate(String transportOrderId, TransportInfoDetail infoDetail) {
+//        根据运单ID查询
+        Query query = Query.query(Criteria.where("transportOrderId").is(transportOrderId));
+        TransportInfoEntity transportInfoEntity = mongoTemplate.findOne(query, TransportInfoEntity.class);
+        if (ObjectUtil.isEmpty(transportInfoEntity)) {
+//            运单信息不存在，新增数据
+            transportInfoEntity = new TransportInfoEntity();
+            transportInfoEntity.setTransportOrderId(transportOrderId);
+            transportInfoEntity.setInfoList(ListUtil.toList(infoDetail));
+            transportInfoEntity.setCreated(System.currentTimeMillis());
+
+//            写到布隆过滤器中
+            bloomFilterService.add(transportOrderId);
+        } else {
+//            运单信息存在，只需要追加物流详情数据
+            transportInfoEntity.getInfoList().add(infoDetail);
+        }
+
+//        无论新增还是更新都要设置更新时间
+        transportInfoEntity.setUpdated(System.currentTimeMillis());
+
+//        清除缓存中的数据
+//        transportInfoCache.invalidate(transportOrderId);
+        stringRedisTemplate.convertAndSend(RedisConfig.CHANNEL_TOPIC, transportOrderId);
+
+        return mongoTemplate.save(transportInfoEntity);
+    }
+
+    /**
+     * 根据运单id查询运输信息
+     *
+     * @param transportOrderId 运单id
+     * @return 运输信息数据
+     */
+    @Override
+    @Cacheable(value = "transport-info", key = "#p0")
+    public TransportInfoEntity queryByTransportOrderId(String transportOrderId) {
+        Query query = Query.query(Criteria.where("transportOrderId").is(transportOrderId));
+        return mongoTemplate.findOne(query, TransportInfoEntity.class);
+    }
+
+}