整理 Spring Data Elasticsearch 接入、专辑上架与下架、RabbitMQ 异步处理及批量添加专辑的学习过程。

本篇要点

  • 在 Spring Boot 中操作索引
  • 梳理专辑上下架链路
  • 理解消息驱动的检索数据更新

**学习提示：**配置中的示例密码已替换为占位符；消息异步执行不等于搜索索引与数据库天然强一致。

内容回顾

1、登录其他接口

  • 登录校验
  • 登录接口

2、es

1、搜索引擎

2、索引库、类型、文档（json）

**3、DSL：**领域专用语言，专门使用在es里面对es里面json格式文档进行查询语言

  • 分页开始位置计算：（当前页-1）*每页记录数

3、Java API操作es

  • ElasticsearchClient操作
1
2
3
4
5
6
7
8
9
SearchResponse<Object> search =
        elasticsearchClient.search(
                s->s.index("my_index")
                 .query(
                        f->f.match(
                                f1->f1.field("title").query("华为手机")
                        )
                )
                , Object.class);

今天内容

1、Java操作es

介绍

  • Java操作es有两种方式：
  • 第一种方式 使用原生api操作-ElasticsearchClient操作

– 实现DSL高级查询操作

  • 第二种方式 使用SpringBoot整合es-ElasticsearchRepository操作

– 实现添加修改或者删除操作

SpringBoot整合ES

  • 在Spring框架里面有模块Spring Data
  • Spring Data是Spring框架封装专门用于操作数据库的模块
  • Spring Data可以操作mysql，可以操作redis，可以操作mongodb，也可以操作es
  • Spring Data封装接口操作es，接口名称 ElasticsearchRepository

第一步 引入Spring Data操作es依赖

1
2
3
4
5
6
7
8
9
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

第二步 创建实体类，设置索引库和映射

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Data
@Document(indexName = "userinfo")
public class UserIndex {

    @Id
    private Long id;

    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String userName;

    @Field(type = FieldType.Long, index = false)
    private Long age;

}

第三步 创建SpringBoot配置文件，配置es服务信息

1
2
3
spring.elasticsearch.uris=http://192.168.200.130:9200
spring.elasticsearch.username=elastic
spring.elasticsearch.password=<ELASTICSEARCH_PASSWORD>

第四步 创建interface继承ElasticsearchRepository

1
2
3
4
5
//ElasticsearchRepository<实体类,实体类主键类型>
public interface UserIndexRepository 
           extends ElasticsearchRepository<UserIndex,Long> {
    
}

第五步 调用ElasticsearchRepository封装方法实现es操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@SpringBootTest
public class ElasticsearchDemo2 {

    @Autowired
    private UserIndexRepository userIndexRepository;

    //添加/修改
    @Test
    public void demo1() throws Exception {
        UserIndex userIndex = new UserIndex();
        userIndex.setId(1L);
        userIndex.setUserName("lucy");
        userIndex.setAge(30L);
        userIndexRepository.save(userIndex);
    }

    //删除
    @Test
    public void demo2() throws Exception {
        userIndexRepository.deleteById(1L);
    }
}

2、专辑上架功能

2.0 概述

  • 如果实现专辑搜索功能，首先把mysql数据库里面专辑信息添加到es里面
  • 专辑添加es过程，在实际操作中有两种场景：

第一种 管理员一次性批量添加专辑数据到es里面

第二种 用户添加专辑时候，同时把专辑数据添加到es里面

  • 最终目的：把mysql里面专辑数据添加到es里面

2.1 使用普通方式实现

2.1.1 分析过程

  • 找到项目中有实体类，和es索引库对应关系

image-20251027093659658

  • 目前做事情：到mysql里面找到AlbumInfoIndex实体类需要的数据，把数据封装到AlbumInfoIndex里面，最终调用ElasticsearchRepository里面save方法添加到es里面

image-20251027101405473

2.1.2 编写五个远程调用接口

接口1：根据专辑id获取专辑信息（完成）
1
2
3
4
5
@GetMapping("getAlbumInfo/{albumId}")
public Result<AlbumInfo> getAlbumInfo(@PathVariable Long albumId) {
    AlbumInfo albumInfo = albumInfoService.getAlbumInfo(albumId);
    return Result.ok(albumInfo);
}
接口2：根据三级分类id获取一级二级分类
BaseCategoryApiController
1
2
3
4
5
6
7
8
9
//远程调用：根据三级分类id获取一级分类id
@Operation(summary = "通过三级分类id查询分类信息")
@GetMapping("getCategoryView/{category3Id}")
public Result<BaseCategoryView>
                  getCategoryView(@PathVariable Long category3Id){
    //因为视图，id就是三级分类id
    BaseCategoryView baseCategoryView = baseCategoryService.getCategoryView(category3Id);
    return Result.ok(baseCategoryView);
}
service
1
2
3
4
5
6
7
//根据三级分类id获取一级和二级id
@Override
public BaseCategoryView getCategoryView(Long category3Id) {
    BaseCategoryView baseCategoryView =
          baseCategoryViewMapper.selectById(category3Id);
    return baseCategoryView;
}
接口3：根据专辑id获取四个统计数据
1
2
3
4
5
6
7
select 
	max(if(stat.stat_type='0401',stat.stat_num,0)) as play1,
	max(if(stat.stat_type='0402',stat.stat_num,0)) as play2,
	max(if(stat.stat_type='0403',stat.stat_num,0)) as play3,
	max(if(stat.stat_type='0404',stat.stat_num,0)) as play4

from album_stat stat where stat.album_id=2
  • BaseCategoryApiController
1
2
3
4
5
6
//远程调用：根据专辑id获取四个统计数据
@GetMapping("getAlbumInfoStat/{albumId}")
public Result getAlbumInfoStat(@PathVariable Long albumId) {
    Map<String,Object> map = albumInfoService.getAlbumInfoStat(albumId);
    return Result.ok(map);
}
  • service
1
2
3
4
5
//远程调用：根据专辑id获取四个统计数据
@Override
public Map<String, Object> getAlbumInfoStat(Long albumId) {
    return albumStatMapper.getAlbumInfoStat(albumId);
}
  • mapper
1
2
3
4
5
6
7
8
9
10
11
12
13
<mapper namespace="com.atguigu.tingshu.album.mapper.AlbumStatMapper">

    <!--//查询四个统计数据-->
    <select id="getAlbumInfoStat" resultType="map">
        SELECT
            max(if(stat.stat_type = '0401', stat_num, 0)) playStatNum,
            max(if(stat.stat_type = '0402', stat_num, 0)) subscribeStatNum,
            max(if(stat.stat_type = '0403', stat_num, 0)) buyStatNum,
            max(if(stat.stat_type = '0404', stat_num, 0)) commentStatNum

        FROM album_stat stat WHERE stat.album_id=#{albumId}
    </select>
</mapper>
接口4：根据专辑id获取标签名称和标签值
1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
 * 根据专辑Id 获取到专辑属性列表
 * @param albumId
 * @return
 */
@Operation(summary = "获取专辑属性值列表")
@GetMapping("findAlbumAttributeValue/{albumId}")
public Result<List<AlbumAttributeValue>>
          findAlbumAttributeValue(@PathVariable Long albumId) {
    // 获取到专辑属性集合
    List<AlbumAttributeValue> albumAttributeValueList =
          albumInfoService.findAlbumAttributeValueByAlbumId(albumId);
    return Result.ok(albumAttributeValueList);
}
1
2
3
4
5
6
7
8
9
10
11
//根据专辑Id 获取到专辑属性列表
@Override
public List<AlbumAttributeValue> findAlbumAttributeValueByAlbumId(Long albumId) {
    LambdaQueryWrapper<AlbumAttributeValue> wrapper = new LambdaQueryWrapper<>();
    wrapper.eq(AlbumAttributeValue::getAlbumId,albumId);

    List<AlbumAttributeValue> albumAttributeValueList =
          albumAttributeValueMapper.selectList(wrapper);
    // 返回集合数据
    return albumAttributeValueList;
}
接口5：根据用户id获取用户信息
  • 在service-user模块编写
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
 * 根据用户Id获取用户信息
 * @param userId
 * @return
 */
@Operation(summary = "根据用户id获取用户信息")
@GetMapping("getUserInfoVo/{userId}")
public Result<UserInfoVo> getUserInfoVo(@PathVariable Long userId) {
    
    UserInfo userInfo = userInfoService.getById(userId);

    // 创建UserInfoVo 对象
    UserInfoVo userInfoVo = new UserInfoVo();
    // 属性拷贝
    BeanUtils.copyProperties(userInfo,userInfoVo);

    return Result.ok(userInfoVo);
}

2.1.3 定义远程调用接口

image-20251027134704503

1
2
3
4
5
6
7
8
9
10
11
12
@FeignClient(value = "service-album",fallback = AlbumInfoDegradeFeignClient.class)
public interface AlbumInfoFeignClient {

    //根据专辑id获取专辑信息
    @GetMapping("api/album/albumInfo/getAlbumInfo/{albumId}")
    public Result<AlbumInfo> getAlbumInfo(@PathVariable("albumId") Long albumId);


    //获取专辑标签名称和标签值列表
    @GetMapping("api/album/albumInfo/findAlbumAttributeValue/{albumId}")
    Result<List<AlbumAttributeValue>> findAlbumAttributeValue(@PathVariable("albumId") Long albumId);
}
1
2
3
4
5
6
7
8
9
10
11
12
@FeignClient(value = "service-album", fallback = CategoryDegradeFeignClient.class)
public interface CategoryFeignClient {

    /**
     * 根据三级分类Id 获取到分类数据
     * @param category3Id
     * @return
     */
    @GetMapping("api/album/category/getCategoryView/{category3Id}")
    Result<BaseCategoryView> getCategoryView(@PathVariable Long category3Id);

}
1
2
3
4
5
6
7
8
9
10
11
12
@FeignClient(value = "service-user", fallback = UserInfoDegradeFeignClient.class)
public interface UserInfoFeignClient {

    /**
     * 根据userId 获取到用户信息
     * @param userId
     * @return
     */
    @GetMapping("api/user/userInfo/getUserInfoVo/{userId}")
    Result<UserInfoVo> getUserInfoVo(@PathVariable Long userId);

}

2.1.4 service-search远程调用

创建Repository
1
2
3
public interface AlbumInfoIndexRepository 
        extends ElasticsearchRepository<AlbumInfoIndex,Long> {
}
SearchApiController编写
1
2
3
4
5
6
7
8
9
//根据专辑id实现上架
@Operation(summary = "上架专辑")
@GetMapping("/upperAlbum/{albumId}")
public Result upperAlbum(@PathVariable Long albumId){
    //  调用服务层方法.
    this.searchService.upperAlbum(albumId);
    //  默认返回
    return Result.ok();
}
service实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private CategoryFeignClient categoryFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;
    
    //根据专辑id实现上架
    @Override
    public void upperAlbum(Long albumId) {
        //1 调用5个远程调用接口，得到返回数据
        //根据专辑id获取专辑信息
        Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
        AlbumInfo albumInfo = albumInfoResult.getData();
        //        if(albumInfo == null) {
        //            throw new GuiguException(ResultCodeEnum.DATA_ERROR);
        //        }
        //断言
        Assert.notNull(albumInfo,"专辑为空");

        //从专辑信息获取三级分类id，根据三级分类id获取一级和二级分类数据
        Long category3Id = albumInfo.getCategory3Id();
        Result<BaseCategoryView> categoryViewResult = categoryFeignClient.getCategoryView(category3Id);
        BaseCategoryView baseCategoryView = categoryViewResult.getData();
        Assert.notNull(baseCategoryView,"分类为空");

        //根据专辑id获取四个统计数据
        Result<Map<String, Object>> albumInfoStatResult = albumInfoFeignClient.getAlbumInfoStat(albumId);
        Map<String, Object> map = albumInfoStatResult.getData();
        //Integer playStatNum = (Integer) map.get("playStatNum");

        //根据专辑id获取标签名称和标签值数据列表
        Result<List<AlbumAttributeValue>> albumAttributeValueResult =
                        albumInfoFeignClient.findAlbumAttributeValue(albumId);
        List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueResult.getData();

        //从专辑信息获取userId，根据userId获取用户信息
        Long userId = albumInfo.getUserId();
        Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(userId);
        UserInfoVo userInfoVo = userInfoVoResult.getData();
        Assert.notNull(userInfoVo,"用户信息为空");

        //2 把五个远程调用接口获取数据封装AlbumInfoIndex实体类
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        //封装albumInfo
        BeanUtils.copyProperties(albumInfo,albumInfoIndex);
        
        //封装baseCategoryView 一级二级三级分类
        Long category1Id = baseCategoryView.getCategory1Id();
        Long category2Id = baseCategoryView.getCategory2Id();
        albumInfoIndex.setCategory1Id(category1Id);
        albumInfoIndex.setCategory2Id(category2Id);
        albumInfoIndex.setCategory3Id(category3Id);
        
        //四个统计数据
        //Integer playStatNum = (Integer) map.get("playStatNum");
        //albumInfoIndex.setPlayStatNum(playStatNum);
        //TODO 为了测试方便，四个统计数据随机数
        //更新统计量与得分，默认随机，方便测试
        int num1 = new Random().nextInt(1000);
        int num2 = new Random().nextInt(100);
        int num3 = new Random().nextInt(50);
        int num4 = new Random().nextInt(300);
        albumInfoIndex.setPlayStatNum(num1);
        albumInfoIndex.setSubscribeStatNum(num2);
        albumInfoIndex.setBuyStatNum(num3);
        albumInfoIndex.setCommentStatNum(num4);
        double hotScore = num1*0.2 + num2*0.3 + num3*0.4 + num4*0.1;
        //  设置热度排名
        albumInfoIndex.setHotScore(hotScore);

        //封装标签名称和标签值 List<AlbumAttributeValue> albumAttributeValueList
        // List<AlbumAttributeValue> --  List<AttributeValueIndex>
        if (!CollectionUtils.isEmpty(albumAttributeValueList)){

            List<AttributeValueIndex> attributeValueIndexList =
                    albumAttributeValueList.stream().map(albumAttributeValue -> {
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                return attributeValueIndex;
            }).collect(Collectors.toList());

            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
        }
        //封装用户名称
        //  赋值主播名称
        albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        
        //3 调用AlbumInfoIndexRepository的save方法实现添加
        albumInfoIndexRepository.save(albumInfoIndex);
    }
}

2.2 CompletableFuture优化功能

分析

串行修改并行

image-20251027144119421

CompletableFuture复习
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public class Demo {

    public static void main(String[] args) {
        //supplyAsync:使用一个线程执行操作，返回结果
        CompletableFuture<Integer> completableFuture1 =
                                 CompletableFuture.supplyAsync(() -> {
            System.out.println("1");
            return 1024;
        });

        //runAsync:使用一个线程执行操作，没有返回结果
        CompletableFuture<Void> completableFuture2 =
                                CompletableFuture.runAsync(() -> {
            System.out.println("2");
        });

        //thenAcceptAsync:上一个线程执行完成之后，才执行当前操作，
        // 获取上一个线程执行返回结果
        CompletableFuture<Void> completableFuture3 =
                         completableFuture1.thenAcceptAsync(value -> {
             System.out.println("3 "+value);
        });

        //等待所有线程执行完成之后，进行汇总
        CompletableFuture.allOf(
                completableFuture1,
                completableFuture2,
                completableFuture3
        ).join();
    }
}

自定义线程池

1
2
3
4
5
6
7
8
9
10
11
12
13
@Configuration
public class ThreadPoolExecutorConfig {

    @Bean
    public ThreadPoolExecutor threadPoolExecutor(){
        return new ThreadPoolExecutor(3,
                5,
                10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    }
}

改造service代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
//改造使用并行实现专辑上架
@Autowired
private ThreadPoolExecutor threadPoolExecutor;

public void upperAlbum1(Long albumId) {
    AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

    //根据专辑id获取专辑信息
    CompletableFuture<AlbumInfo> completableFuture1 =
            CompletableFuture.supplyAsync(() -> {
                Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
                AlbumInfo albumInfo = albumInfoResult.getData();
                Assert.notNull(albumInfo,"专辑为空");

                //封装到albumInfoIndex
                BeanUtils.copyProperties(albumInfo,albumInfoIndex);
                return albumInfo;
     },threadPoolExecutor);

    //获取分类数据
    //专辑获取之后执行
    CompletableFuture<Void> completableFuture2 =
                    completableFuture1.thenAcceptAsync(albumInfo -> {
            //获取三级分类id
            Long category3Id = albumInfo.getCategory3Id();
            Result<BaseCategoryView> categoryViewResult = categoryFeignClient.getCategoryView(category3Id);
            BaseCategoryView baseCategoryView = categoryViewResult.getData();
            Assert.notNull(baseCategoryView,"分类数据为空");

            //封装三个分类id到albumInfoIndex
            Long category1Id = baseCategoryView.getCategory1Id();
            Long category2Id = baseCategoryView.getCategory2Id();
            albumInfoIndex.setCategory1Id(category1Id);
            albumInfoIndex.setCategory2Id(category2Id);
            albumInfoIndex.setCategory3Id(category3Id);
    },threadPoolExecutor);
    //获取专辑对应标签数据
    CompletableFuture<Void> completableFuture3 = CompletableFuture.runAsync(() -> {
        Result<List<AlbumAttributeValue>> albumAttributeValueResult = albumInfoFeignClient.findAlbumAttributeValue(albumId);
        List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueResult.getData();
        //封装 albumInfoIndex
        // List<AlbumAttributeValue> -->  List<AttributeValueIndex>
        if (!CollectionUtils.isEmpty(albumAttributeValueList)) {
            List<AttributeValueIndex> attributeValueIndexList =
                    albumAttributeValueList.stream().map(albumAttributeValue -> {
                        AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                        BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                        return attributeValueIndex;
                    }).collect(Collectors.toList());
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
        }

    },threadPoolExecutor);

    //获取用户信息
    CompletableFuture<Void> completableFuture4 = completableFuture1.thenAcceptAsync(albumInfo -> {
        Long userId = albumInfo.getUserId();
        Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(userId);
        UserInfoVo userInfoVo = userInfoVoResult.getData();
        //封装 albumInfoIndex
        albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
    },threadPoolExecutor);

    //  赋值：初始化统计信息
    int playStatNum = new Random().nextInt(100000);
    int subscribeStatNum = new Random().nextInt(100000000);
    int buyStatNum = new Random().nextInt(10000000);
    int commentStatNum = new Random().nextInt(1000000000);
    albumInfoIndex.setPlayStatNum(playStatNum);
    albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
    albumInfoIndex.setBuyStatNum(buyStatNum);
    albumInfoIndex.setCommentStatNum(commentStatNum);

    //等待所有任务都完成，汇总
    CompletableFuture.allOf(
            completableFuture1,
            completableFuture2,
            completableFuture3,
            completableFuture4
    ).join();

    //调用方法添加es
    albumInfoIndexRepository.save(albumInfoIndex);
}

3、专辑下架

  • 根据专辑id，把es存储专辑数据删除
1
2
3
4
5
6
7
8
9
10
11
/**
 * 下架专辑
 * @param albumId
 * @return
 */
@Operation(summary = "下架专辑")
@GetMapping("lowerAlbum/{albumId}")
public Result lowerAlbum(@PathVariable Long albumId) {
    searchService.lowerAlbum(albumId);
    return Result.ok();
}
1
2
3
4
5
//专辑下架
@Override
public void lowerAlbum(Long albumId) {
    albumInfoIndexRepository.deleteById(albumId);
}

4、利用mq实现专辑上下架

分析

  • 需求1：添加专辑时候，如果专辑状态公开状态，把专辑上架（添加es里面）
  • 需求2：修改专辑时候，如果把专辑状态修改为公开，把专辑上架

​ 如果把专辑状态修改为私密，把专辑下架

  • 需求3：删除专辑，把专辑下架（从es删除）
  • 这些操作通过发送mq消息异步实现

添加专辑上架

  • 找到专辑模块里面添加专辑的service方法
  • 根据是否公开判断，如果公开，发送mq消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
//保存专辑
//当前操作操作多张表，保证多张表数据一致性，添加事务
//当前这些表在一个数据库里面，这种事务称为本地事务
@Transactional
@Override
public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {
    //1 添加专辑基本信息 album_info
    AlbumInfo albumInfo = new AlbumInfo();
    // AlbumInfoVo 值--放到 AlbumInfo
    //String albumTitle = albumInfoVo.getAlbumTitle();
    //albumInfo.setAlbumTitle(albumTitle);
    BeanUtils.copyProperties(albumInfoVo,albumInfo);

    //专辑有几个值需要单独设置，前端没有传递过来的
    //TODO  userId 用户id，后面完善
    albumInfo.setUserId(1L);
    // 专辑状态，设置通过
    albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
    //设置收费专辑，免费试听集数，前3集
    String payType = albumInfo.getPayType();
    if(!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(payType)) {
       albumInfo.setTracksForFree(3);
    }
    //调用方法保存
    albumInfoMapper.insert(albumInfo);

    //2 添加专辑下面标签名称和标签值数据 album_attribute_value
    List<AlbumAttributeValueVo> albumAttributeValueVoList =
          albumInfoVo.getAlbumAttributeValueVoList();
    //非空判断
    if(!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
       albumAttributeValueVoList.stream().forEach(albumAttributeValueVo -> {
          AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
          // AlbumAttributeValueVo --  AlbumAttributeValue
          BeanUtils.copyProperties(albumAttributeValueVo,albumAttributeValue);
          //专辑id
          albumAttributeValue.setAlbumId(albumInfo.getId());

          albumAttributeValueMapper.insert(albumAttributeValue);
       });
    }

    //3 添加专辑四个统计数据 播放量，订阅量等 初始值 0 album_stat
    //播放量
    this.saveAlbumStat(albumInfo.getId(),SystemConstant.ALBUM_STAT_PLAY);
    //订阅量
    this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
    //购买量
    this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
    //评论数
    this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

    //判断专辑是否公开，如果公开，发送mq消息，实现专辑上架
    String isOpen = albumInfo.getIsOpen();
    // 发送上架消息
    if ("1".equals(isOpen)){
       rabbitService.sendMessage(MqConst.EXCHANGE_ALBUM, 
                               MqConst.ROUTING_ALBUM_UPPER, 
                               albumInfo.getId());
    }
}
  • 在service-search模块创建接收端
  • 接收发送过来消息（专辑id），根据专辑id调用方法实现专辑上架
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Component
public class SearchReceiver {

    @Autowired
    private SearchService searchService;
    
    @SneakyThrows
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = MqConst.EXCHANGE_ALBUM, durable = "true"),
            value = @Queue(value = MqConst.QUEUE_ALBUM_UPPER, durable = "true"),
            key = {MqConst.ROUTING_ALBUM_UPPER}
    ))
    public void upper_album(Long albumId, Message message, Channel channel) {
        if(albumId != null) {
            searchService.upperAlbum(albumId);
        }
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

删除专辑下架

  • 在专辑模块找到删除专辑的service方法
  • 发送mq消息，实现专辑下架
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
//删除专辑
@Override
public void removeAlbumInfo(String albumId) {
    //1 判断当前专辑下面是否包含声音，如果包含不能删除
    // select count(*) from track_info where album_id=?
    LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
    queryWrapper.eq(TrackInfo::getAlbumId,albumId);
    Long count = trackInfoMapper.selectCount(queryWrapper);
    if(count > 0) {//包含声音，如果包含不能删除
       throw new GuiguException(400, "该专辑下存在未删除声音！");
    }

    //2 如果专辑下面不包含声音，可以删除
    //2.1 删除专辑基本信息
    albumInfoMapper.deleteById(albumId);

    //2.2 删除专辑标签名称和标签值数据
    LambdaQueryWrapper<AlbumAttributeValue> queryWrapper1 =
                                         new LambdaQueryWrapper<>();
    queryWrapper1.eq(AlbumAttributeValue::getAlbumId,albumId);
    albumAttributeValueMapper.delete(queryWrapper1);

    //2.3 删除专辑四个统计数据
    LambdaQueryWrapper<AlbumStat> queryWrapper2 =
          new LambdaQueryWrapper<>();
    queryWrapper2.eq(AlbumStat::getAlbumId,albumId);
    albumStatMapper.delete(queryWrapper2);

    //发送mq消息
    //下架
    rabbitService.sendMessage(MqConst.EXCHANGE_ALBUM, 
          MqConst.ROUTING_ALBUM_LOWER, albumId);
}
  • 在service-search模块创建接收端
  • 接收发送过来消息（专辑id），根据专辑id调用方法实现专辑下架
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
 * 专辑下架
 *
 * @param albumId
 * @param message
 * @param channel
 */
@SneakyThrows
@RabbitListener(bindings = @QueueBinding(
        exchange = @Exchange(value = MqConst.EXCHANGE_ALBUM, durable = "true"),
        value = @Queue(value = MqConst.QUEUE_ALBUM_LOWER, durable = "true"),
        key = {MqConst.ROUTING_ALBUM_LOWER}
))
public void lowerGoods(Long albumId, Message message, Channel channel) {
    //业务处理
    if (null != albumId) {
        searchService.lowerAlbum(albumId);
    }

    //手动应答
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

5、批量添加专辑

在service-search模块添加方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
 * 批量上架
 * @return
 */
@Operation(summary = "批量上架")
@GetMapping("batchUpperAlbum")
public Result batchUpperAlbum(){
    //  循环
    for (long i = 1; i <= 1500; i++) {
        searchService.upperAlbum(i);
    }
    //  返回数据
    return Result.ok();
}