常见的数据同步方案有三种：

• 同步调用

• 异步通知

• 监听binlog

方案一：同步调用（基本不会用，问题太多）

 基本步骤如下：

• hotel-demo对外提供接口，用来修改elasticsearch中的数据

• 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口

• 优点：实现简单，粗暴

• 缺点：业务耦合度高

方案二：异步通知（一般会选择这种方案）

 流程如下：

• hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息

• hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改

• 优点：低耦合，实现难度一般

• 缺点：依赖mq的可靠性

方案三：监听binlog

 流程如下：

• 给mysql开启binlog功能

• mysql完成增、删、改操作都会记录在binlog中

• hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容

• 优点：完全解除服务间耦合

• 缺点：开启binlog增加数据库负担、实现复杂度高，要求数据库只能是mysql

在此，只演示方案二：

1）引入依赖，改配置

在消费方和生产方，引入rabbitmq的依赖：

org.springframework.bootspring-boot-starter-amqp

 在消费方和生产方，修改application.yaml配置文件

spring:rabbitmq:host:192.168.177.132 port:5672username:itcastpassword:123321virtual-host:/

2）在消费方，声明队列交换机名称

package cn.itcast.hotel.constatnts;public class MqConstants {/*** 交换机*/public final static String HOTEL_EXCHANGE = "hotel.topic";/*** 监听新增和修改的队列*/public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";/*** 监听删除的队列*/public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";/*** 新增或修改的RoutingKey*/public final static String HOTEL_INSERT_KEY = "hotel.insert";/*** 删除的RoutingKey*/public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

3）在消费方，声明队列交换机

定义配置类，声明队列、交换机：

package cn.itcast.hotel.config;import cn.itcast.hotel.constants.MqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MqConfig {//交换机@Beanpublic TopicExchange topicExchange(){return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);}//队列@Beanpublic Queue insertQueue(){return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);}//队列@Beanpublic Queue deleteQueue(){return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);}//绑定@Beanpublic Binding insertQueueBinding(){return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);}//绑定@Beanpublic Binding deleteQueueBinding(){return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);}
}

4）在生产方，操作mysql的增、删、改业务中分别给MQ发送消息：

1.在conroller层，注入MQ对象

@Autowired
private RabbitTemplate rabbitTemplate;

2.在增、删、改业务中分别给MQ发送消息

5）当消费方（操作es侧），接收到MQ的消息：

• 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库

• 删除消息：根据传递的hotel的id删除索引库中的一条数据

1.在消费方的service中添加新增、删除业务

void deleteById(Long id);void insertById(Long id);

2.service的实现类，实现业务：

@Autowired
private RestHighLevelClient client;@Override
public void deleteById(Long id) {try {// 1.准备RequestDeleteRequest request = new DeleteRequest("hotel", id.toString());// 2.发送请求client.delete(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}
}@Override
public void insertById(Long id) {try {// 0.根据id查询酒店数据Hotel hotel = getById(id);// 转换为文档类型HotelDoc hotelDoc = new HotelDoc(hotel);// 1.准备Request对象IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());// 2.准备Json文档request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);// 3.发送请求client.index(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}
}

3.编写监听器

package cn.itcast.hotel.mq;import cn.itcast.hotel.constants.MqConstants;
import cn.itcast.hotel.service.IHotelService;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class HotelListener {@Autowiredprivate IHotelService hotelService;/*** 监听酒店新增或修改的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)public void listenHotelInsertOrUpdate(Long id){hotelService.insertById(id);}/*** 监听酒店删除的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)public void listenHotelDelete(Long id){hotelService.deleteById(id);}
}