项目中ES使用---并使用mq解决业务上ES与数据库的数据不一致的问题_es与数据库一致性

概述：

1、当数据量过大时，将数据存放到es中，增加查询的效率，降低和数据库的交互。【一次性导入】
2、使用mq同步es数据，避免DB中数据和Es中的不一致。【消息队列持续同步db中的数据】
3、注意springboot和es的版本问题，不然项目一定起不来【elk6.几不支持springboot2.3点几的版本】
我用的是：springboot-2.2.10.RELEASE；es-6.4版本
关键导包

        <!--集成es-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>
        <!--集成rabbitMq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
1
2
3
4
5
6
7
8
9
10

理解图

mq同步数据图

coding

1、创建Es索引，定义分词和业务需要展示的字段

【1】创建es索引字段类（每个注解的意思注释在代码中）

package com.xusj.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;

import java.math.BigDecimal;
import java.util.Date;

/**
 * @author xusj
 * <br>CreateDate 2022/5/14 13:32
 */
// 定义es中名称（相当于数据库中的表的概念）
@Document(indexName = "movie", type = "movie")
public class MovieSearch {
    /**
     * 主键，es中必须设置主键（相当于数据库中的主键【这里最好和数据库的主键一致】）
     */
    @Id
    private Integer movieId;
    /**
     * 该字段为自己添加的字段，为了方便，以前可能是两个字段在查询，现在可以和到一个字段
     * 进行操作
     * 注意：这个字段并不是数据库中有的字段，是我们自定义的字段，如果你需要对多个字段进行分词
     * 		可以使用一个新的字段存入到es中，统一分词
     */
    // 注解：type（相当于数据库中的数据类型）
    // analyzer：分词器ik_max_word
    // index默认为true：使用改字段查询的意思
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String allText;

    /**
     * 根据业务要求添加的字段，电影类型，并不是所有业务中的字段都在数据库中，
     * 有一些需要搜索的结果应该是自己手动拼接起来的
     * 注：业务中需要的展示的数据进行拼接保存到es（*业务来找数据，而不是数据去找业务）
     */
    @Field(type = FieldType.Text, = false)
    private String movieType;

    /**
     * 影片名称，index为false==》不需要将其建为索引，单需要出现在查询的结果中
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieName;

    /**
     * 上映时间
     */
    @Field(type = FieldType.Date, index = false)
    private Date movieReleaseTime;

    /**
     * 首次上映地区
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieAreaDesc;

    /**
     * 影片描述
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieInfo;

    /**
     * 电影版本
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieVersion;

    /**
     * 影片海报
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieImg;

    /**
     * 影片英文名
     */
    @Field(type = FieldType.Keyword, index = false)
    private String movieEname;

    /**
     * 评分
     */
    @Field(type = FieldType.Double, index = false)
    private BigDecimal movieScore;

    public String getMovieType() {
        return movieType;
    }

    public void setMovieType(String movieType) {
        this.movieType = movieType;
    }

    public Integer getMovieId() {
        return movieId;
    }

    public void setMovieId(Integer movieId) {
        this.movieId = movieId;
    }

    public String getAllText() {
        return allText;
    }

    public void setAllText(String allText) {
        this.allText = allText;
    }

    public String getMovieName() {
        return movieName;
    }

    public void setMovieName(String movieName) {
        this.movieName = movieName;
    }

    public Date getMovieReleaseTime() {
        return movieReleaseTime;
    }

    public void setMovieReleaseTime(Date movieReleaseTime) {
        this.movieReleaseTime = movieReleaseTime;
    }

    public String getMovieAreaDesc() {
        return movieAreaDesc;
    }

    public void setMovieAreaDesc(String movieAreaDesc) {
        this.movieAreaDesc = movieAreaDesc;
    }

    public String getMovieInfo() {
        return movieInfo;
    }

    public void setMovieInfo(String movieInfo) {
        this.movieInfo = movieInfo;
    }

    public String getMovieVersion() {
        return movieVersion;
    }

    public void setMovieVersion(String movieVersion) {
        this.movieVersion = movieVersion;
    }

    public String getMovieImg() {
        return movieImg;
    }

    public void setMovieImg(String movieImg) {
        this.movieImg = movieImg;
    }

    public String getMovieEname() {
        return movieEname;
    }

    public void setMovieEname(String movieEname) {
        this.movieEname = movieEname;
    }

    public BigDecimal getMovieScore() {
        return movieScore;
    }

    public void setMovieScore(BigDecimal movieScore) {
        this.movieScore = movieScore;
    }

    //    /**
//     * 票房金额
//     */
//    private Long movieBoxOffice;
//
//    /**
//     * 影片时长（分钟）
//     */
//    private Integer movieTime;
//
//    /**
//     * 想要观看人数
//     */
//    private Long movieWantCount;
//
//    /**
//     * 影片上映年份
//     */
//    private Integer movieReleaseYear;
//
//    /**
//     * 累计评论人数
//     */
//    private Long commentUserCount;
}

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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205

【2】编写方法，创建到es中

@Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

@GetMapping("search/createIndex")
    public String createIndex() {
        // 使用es的模板将其索引进行创建
        elasticsearchTemplate.createIndex(MovieSearch.class);
        // 将其put进es中
        elasticsearchTemplate.putMapping(MovieSearch.class);

        return "导入成功";
    }
1
2
3
4
5
6
7
8
9
10
11
12

【3】可以使用Kibana查看使用创建成功

GET movie/_mapping【查看字段是否设置ok（自己设置的index/)mapping）】

2、导入数据到es中

【1】定义接口继承，es存储库

package com.xusj.search.repository;

import com.xusj.domain.MovieSearch;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

/**
 * 定义es的存储库
 * ElasticsearchRepository<MovieSearch, Integer>，es中的存储库，注意泛型定义
 * 1.定义的es索引字段；2.库主键
 * @author xusj
 * <br>CreateDate 2022/5/14 17:43
 */
public interface MovieRepository extends ElasticsearchRepository<MovieSearch, Integer> {
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

【2】编写方法同步到es中

@Autowired
private MovieRepository movieRepository;

@Override
    public void init() {
        // 查询出数据库中的数据，用于存放到数据库中
        List<MyMovieVo> myMovieVoList = movieApi.initMovie();

        System.out.println(myMovieVoList);
        // 将数据初始化到es中,封装es中对应的（注意使用的必须要是第一步中es中创建的那些字段）
        List<MovieSearch> collect = myMovieVoList.stream().map(item -> {
            MovieSearch movieSearch = new MovieSearch();
            BeanUtils.copyProperties(item, movieSearch);
            movieSearch.setMovieAreaDesc(item.getMovieAreaStr());
            movieSearch.setMovieType(item.getMovieTypeStr());
            // 这个是我们自定义的分词
            movieSearch.setAllText(item.getMovieEname() + " " + item.getMovieName() + " " + item.getMovieInfo());
            return movieSearch;
        }).collect(Collectors.toList());

        // 使用【1】中定义的接口。导入es中
        movieRepository.saveAll(collect);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

【3】可以使用Kibana查看使用创建成功

GET movie/_search

3、使用mq实时同步数据库数据，避免数据不一致

注意!!!!
这里的前提是以数据量小的前提下可使用消息队列做同步。

如果数据量巨大，消息队列积压数据过多，开启的消费者太多又超级占用MySQL资源导致MySQL崩溃，最好的做法是使用canal或者otter监听MySQL二进制日志来同步。
感谢评论区guangqianglu7985大佬提醒！！！

【1】生产者（业务方）----定义方法，在对数据的修改添加删除后都将消息放到队列中

// 使用 rabbitTemplate
@Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 写一个统一的方法，将影院的crud操作发送到mq中
     */
    private void sendMsg(Map<String, String> msg) {
        // 将msg放到对应的交换机中，和key中
        rabbitTemplate.convertAndSend(UserConstants.MQ_CINEMA_UPDATE_EXCHANGE,
                UserConstants.MQ_CINEMA_UPDATE_ROUTING_KEY, msg);

    }

// 例子，修改数据
/**
* 新增影片信息
* @param movie 影片信息
* @return 结果
*/
@Override
public int insertMovie(Movie movie) {
    int row=movieMapper.insertMovie(movie);
    // 避免消息堆积，消息尽量短小
    sendMsg(去拿到新增movie的主键id，封装map发送消息【map的key写死movieId便于区分，value使用电影主键id】);
    return row;
}
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

【2】消费者（es存储方）----消费id，取到id查询对应业务方的最新数据更新到es中

    /**
     * 定义一个方法去监听消费
     */
    @RabbitListener(bindings = {
            @QueueBinding(
                    exchange = @Exchange(name = UserConstants.MQ_CINEMA_UPDATE_EXCHANGE,
                            type = ExchangeTypes.TOPIC),
                    value = @Queue(value = UserConstants.MQ_CINEMA_UPDATE_QUEUE, durable = "true"),
                    key = {UserConstants.MQ_CINEMA_UPDATE_ROUTING_KEY}
            )
    })
    public void receiveCinemaMsg(Map<String, String> msg) {
        // 这是消息中的key
        String cinemaId = msg.get("cinemaId");
        // 接收到消息后，拿到id去查找具体的消息，然后更新es中的数据
        api远程去调用业务方去查询
        // MovieRepository.save(null);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18