1
public void onSale(Long skuId) {
2
    SkuInfo skuInfo = new SkuInfo();
3
    skuInfo.setIsSale(1);
4
    skuInfo.setId(skuId);
5
    skuInfoMapper.updateById(skuInfo);
6

7
    RBloomFilter<Long> bloomFilter =
8
            redissonClient.getBloomFilter(RedisConst.SKU_BLOOM_FILTER);
9
    bloomFilter.add(skuId);
10

11
    searchApiClient.upperGoods(skuId);
12

13
    baseProducer.sendMessage(
14
            MqTopicConst.PRODUCT_ONSALE_TOPIC, skuId);
15
}

1
@FeignClient("service-product")
2
public interface ProductApiClient {
3
    @GetMapping("/api/product/inner/getSkuInfo/{skuId}")
4
    SkuInfoDTO getSkuInfo(@PathVariable Long skuId);
5

6
    @GetMapping("/api/product/inner/getCategoryView/{categoryId}")
7
    CategoryHierarchyDTO getCategoryView(
8
            @PathVariable("categoryId") Long categoryId);
9

10
    @GetMapping("/api/product/inner/getAttrList/{skuId}")
11
    List<PlatformAttributeInfoDTO> getAttrList(
12
            @PathVariable Long skuId);
13

14
    @GetMapping("/api/product/inner/getTrademark/{tmId}")
15
    TrademarkDTO getTrademark(@PathVariable Long tmId);
16
}

1
Goods goods = new Goods();
2
ExecutorService executorService = Executors.newFixedThreadPool(8);
3

4
CompletableFuture<SkuInfoDTO> cf1 =
5
        CompletableFuture.supplyAsync(() -> {
6
            SkuInfoDTO skuInfo = productApiClient.getSkuInfo(skuId);
7
            if (skuInfo == null) {
8
                throw new RuntimeException(
9
                        "获取 SKU 基本信息失败，skuId=" + skuId);
10
            }
11

12
            goods.setId(skuInfo.getId());
13
            goods.setTitle(skuInfo.getSkuName());
14
            goods.setDefaultImg(skuInfo.getSkuDefaultImg());
15
            goods.setPrice(skuInfo.getPrice().doubleValue());
16
            return skuInfo;
17
        }, executorService);

1
List<SearchAttr> searchAttrs = attrList.stream()
2
        .filter(Objects::nonNull)
3
        .map(attr -> {
4
            List<PlatformAttributeValueDTO> values =
5
                    attr.getAttrValueList();
6
            if (values == null || values.isEmpty()) {
7
                return null;
8
            }
9

10
            PlatformAttributeValueDTO value = values.get(0);
11
            if (value == null || value.getValueName() == null) {
12
                return null;
13
            }
14

15
            SearchAttr searchAttr = new SearchAttr();
16
            searchAttr.setAttrId(attr.getId());
17
            searchAttr.setAttrName(attr.getAttrName());
18
            searchAttr.setAttrValue(value.getValueName());
19
            return searchAttr;
20
        })
21
        .filter(Objects::nonNull)
22
        .collect(Collectors.toList());

1
defaultMQPushConsumer.subscribe(
2
        MqTopicConst.PRODUCT_ONSALE_TOPIC, "*");
3

4
defaultMQPushConsumer.setMessageListener(
5
        new MessageListenerConcurrently() {
6
            @Override
7
            public ConsumeConcurrentlyStatus consumeMessage(
8
                    List<MessageExt> msgs,
9
                    ConsumeConcurrentlyContext context) {
10
                Long skuId = JSON.parseObject(
11
                        new String(msgs.get(0).getBody()), Long.class);
12
                searchService.upperGoods(skuId);
13
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
14
            }
15
        });
16

17
// 文件末尾没有 defaultMQPushConsumer.start()

1
defaultMQPushConsumer.subscribe(
2
        MqTopicConst.PRODUCT_OFFSALE_TOPIC, "*");
3

4
defaultMQPushConsumer.setMessageListener(
5
        new MessageListenerConcurrently() {
6
            @Override
7
            public ConsumeConcurrentlyStatus consumeMessage(
8
                    List<MessageExt> msgs,
9
                    ConsumeConcurrentlyContext context) {
10
                Long skuId = JSON.parseObject(
11
                        new String(msgs.get(0).getBody()), Long.class);
12
                searchService.lowerGoods(skuId);
13
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
14
            }
15
        });
16

17
defaultMQPushConsumer.start();