校园选课助手【5】-解决Elasticsearch和MySQL同步

未优化前：课程检索库中的数据经过一次同步后固定不变，用户搜索到对应的课程后点击跳转到课程详情页面，进行选课

存在的问题：

1. 用户使用不友好，搜索就是为了进行选课，应该返回对应课程的余量，如果没有余量了，进行友好提示，如果有余量，跳转到详情页面进行选课。
2. 数据不一致，可能修改了课程信息，而检索库中的数据依旧没有更新。

解决方案：
引入消息队列，当CourseController将数据写入mysql后，需要自己再往MQ发条消息，说"数据更新了"，EsController收到消息去更新索引库。

1.声明队列和交换机

@Configuration
public class MqConfig {
	//交换机名称
    public static final String EXCHANGE_NAME = "course.topic";
    //新增和修改队列
    public static final String INSERT_QUEUE_NAME = "course.insert.queue";
    //删除队列
    public static final String DELETE_QUEUE_NAME = "course.delete.queue";
    //RoutingKey
    public static final String INSERT_KEY = "course.insert";
    public static final String DELETE_KEY = "course.delete";
    
@Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(EXCHANGE_NAME,true,false);
    }

    @Bean
    public Queue insertQueue(){
        return new Queue(INSERT_QUEUE_NAME,true);
    }

    @Bean
    public Queue deleteQueue(){
        return new Queue(DELETE_QUEUE_NAME,true);
    }

    /**
     * 绑定队列和交换机关系
     */
    @Bean
    public Binding insertQueueBinding(){
        return BindingBuilder
                .bind(insertQueue())
                .to(topicExchange())
                .with(INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding(){
        return BindingBuilder
                .bind(deleteQueue())
                .to(topicExchange())
                .with(DELETE_KEY);
    }

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

2.发送消息

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.security.InvalidParameterException;

@RestController
@RequestMapping("Course")
public class courseController {

    @Autowired
    private ICourseService courseService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostMapping
    public void saveCourse(@RequestBody Course course){
        courseService.save(course);
        rabbitTemplate.convertAndSend("course.topic","course.insert", course.getId());
    }

    @PutMapping()
    public void updateById(@RequestBody course course){
        if (course.getId() == null) {
            throw new InvalidParameterException("id不能为空");
        }
        courseService.updateById(course);

        // 发送MQ消息
        rabbitTemplate.convertAndSend("course.topic","course.insert", course.getId());
    }

    @DeleteMapping("/{id}")
    public void deleteById(@PathVariable("id") Long id) {
        courseService.removeById(id);

        // 发送MQ消息
        rabbitTemplate.convertAndSend("course.topic", "course.delete", id);
    }   
}

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

3.监听消息队列并进行处理

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


@Component
public class CourseListener {

    @Resource
    ICourselService courseService;

    /**
     * 监听课程新增或者修改的业务
     * id接受一个Long，因为发送过来的是一个Long id
     */
    @RabbitListener(queues = "course.insert.queue“)
    public void listenCourseInsertAndUpdate(Long id){
        CourseService.insertDocById(id);
    }

    /**
     * 监听课程删除业务
     */
    @RabbitListener(queues = "course.delete.queue“)
    public void listenCourseDelete(Long id){
        CourseService.deleteDocById(id);
    }
}

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

4.实现上述service功能

新增或修改课程逻辑

• 1）根据id查询课程数据Item
• 2）将Item封装为ItemDoc
• 3）将ItemDoc序列化为JSON
• 4）创建IndexRequest，指定索引库名和id
• 5）准备请求参数，也就是JSON文档
• 6）发送请求

删除课程逻辑

• 1）准备Request对象，因为是删除，这次是DeleteRequest对象。要指定索引库名和id
• 3）发送请求。因为是删除，所以是client.delete()方法

import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;

@Service
public class CourseService extends ServiceImpl<CourseMapper, Course> implements ICourseService {

    @Resource
    RestHighLevelClient client;

    @Override
    public void insertDocById(Long id) {
        try {
            //0.根据ID查数据，并转为文档类型
            Course Course = getById(id);
            CourseDoc CourseDoc = new CourseDoc(Course);
            //1.准备request
            IndexRequest request = new IndexRequest("Course").id(CourseDoc.getId().toString());
            //2.准备DSL
            request.source(JSON.toJSONString(CourseDoc), XContentType.JSON);
            //3.发送请求
            client.index(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deleteDocById(Long id) {

        try {
            //1.准备request
            DeleteRequest request = new DeleteRequest("Course",id.toString());
            //2.发送请求
            client.delete(request,RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
}
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