用idea进行数据同步

1. 声明对列和交换机

   你需要先在yaml文件当中进行rabbitmq的相关配置

rabbitmq:host:192.168.150.101        //消息件的地址port:5672                   //端口数据username:itcast             //用户名password:123321             //密码virtual-host:/              //虚拟机主机名

声明队列交换机，创建新的工具类，定义不同功能的交换机

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

• 生产者发送消息时，指定交换机为 HOTEL_EXCHANGE 、路由键为 HOTEL_INSERT_KEY 等。

• 消费者声明队列并绑定到 HOTEL_EXCHANGE 交换机，通过对应的路由键（HOTEL_INSERT_KEY / HOTEL_DELETE_KEY ）来订阅特定业务（新增修改 / 删除 ）的消息 。

定义队列交换机的对象和绑定关系：

1. 基于@Bean的方式

   这种方式适合在配置类当中集中管理队列，交换机，绑定关系的 Bean 定义，结构清晰，便于统一维护。

   import org.springframework.amqp.core.*;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   ​
   // 配置类注解，让 Spring 扫描并加载这些 Bean 定义
   @Configuration
   public class RabbitMqConfig {
   ​// 1. 定义交换机（对应 MqConstants 里的交换机）@Beanpublic TopicExchange hotelExchange() {// 创建一个 Topic 类型的交换机，名称用 MqConstants 里的return new TopicExchange(MqConstants.HOTEL_EXCHANGE);}
   ​// 2. 定义“新增和修改”队列@Beanpublic Queue hotelInsertQueue() {return new Queue(MqConstants.HOTEL_INSERT_QUEUE);}
   ​// 3. 定义“删除”队列@Beanpublic Queue hotelDeleteQueue() {return new Queue(MqConstants.HOTEL_DELETE_QUEUE);}
   ​// 4. 绑定“新增和修改”队列到交换机（设置路由键）@Beanpublic Binding bindHotelInsertQueue(TopicExchange hotelExchange, Queue hotelInsertQueue) {// 用 MqConstants 里的新增路由键，将队列绑定到交换机return BindingBuilder.bind(hotelInsertQueue).to(hotelExchange).with(MqConstants.HOTEL_INSERT_KEY);}
   ​// 5. 绑定“删除”队列到交换机（设置路由键）@Beanpublic Binding bindHotelDeleteQueue(TopicExchange hotelExchange, Queue hotelDeleteQueue) {// 用 MqConstants 里的删除路由键，将队列绑定到交换机return BindingBuilder.bind(hotelDeleteQueue).to(hotelExchange).with(MqConstants.HOTEL_DELETE_KEY);}
   }

   • 交换机类型：这里用了 TopicExchange（主题交换机 ），和 MqConstants 里 hotel.topic 对应，也可以根据实际需求换成 DirectExchange（直连 ）、FanoutExchange（扇形 / 广播 ）等。

   • Bean 依赖注入：绑定方法（如 bindHotelInsertQueue ）的参数，会由 Spring 自动注入对应的交换机、队列 Bean，要保证方法参数名或类型能匹配上容器里的 Bean 。

   • 路由键作用：通过 with 方法指定路由键，这样交换机就会根据路由键，把不同业务（新增修改、删除 ）的消息路由到对应的队列。

   1. 基于注解的方式

      这种方式更灵活，通常在消费者监听方法上直接声明队列、交换机和绑定关系，适合快速开发简单场景，或者临时新增队列绑定的情况。

      import org.springframework.amqp.rabbit.annotation.Exchange;
      import org.springframework.amqp.rabbit.annotation.Queue;
      import org.springframework.amqp.rabbit.annotation.QueueBinding;
      import org.springframework.amqp.rabbit.annotation.RabbitListener;
      import org.springframework.stereotype.Component;
      ​
      @Component // 让 Spring 扫描到这个组件
      public class HotelMqListener {
      ​// 监听“新增和修改”队列，同时声明队列、交换机、绑定关系@RabbitListener(bindings = @QueueBinding(value = @Queue(name = MqConstants.HOTEL_INSERT_QUEUE, durable = "true"), // 声明队列，durable = true 表示持久化exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = "topic"), // 声明交换机，类型 topickey = MqConstants.HOTEL_INSERT_KEY // 路由键))public void handleHotelInsert(String message) {// 这里写接收到“新增或修改”消息后的业务逻辑，比如更新 Elasticsearch 酒店数据System.out.println("收到酒店新增/修改消息：" + message);}
      ​// 监听“删除”队列，同时声明队列、交换机、绑定关系@RabbitListener(bindings = @QueueBinding(value = @Queue(name = MqConstants.HOTEL_DELETE_QUEUE, durable = "true"),exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = "topic"),key = MqConstants.HOTEL_DELETE_KEY))public void handleHotelDelete(String message) {// 这里写接收到“删除”消息后的业务逻辑，比如从 Elasticsearch 移除酒店数据System.out.println("收到酒店删除消息：" + message);}
      }

      • 注解参数说明：

        • @Queue：声明队列，name 指定队列名称，durable 设置是否持久化（建议生产环境设为 true ，防止 RabbitMQ 重启队列丢失 ）。

        • @Exchange：声明交换机，name 是交换机名称，type 指定类型（如 topic 、direct 、fanout 等 ）。

        • @QueueBinding：把队列、交换机、路由键绑定起来，让交换机知道如何把消息路由到这个队列。

      • 消费者方法：handleHotelInsert 和 handleHotelDelete 方法就是实际处理消息的逻辑，当对应的队列有消息进来时，方法会被触发执行。

      以下为Rabbitmq的消息发送的代码，当我们在进行增删改等操作的时候，我们只需在其方法内进行消息的发送

       // 新增酒店@PostMappingpublic void saveHotel(@RequestBody Hotel hotel) {hotelService.save(hotel);// 发送新增消息，用新增路由键rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId());}
      ​// 修改酒店@PutMappingpublic void updateById(@RequestBody Hotel hotel) {if (Objects.isNull(hotel.getId())) {throw new InvalidParameterException("id不能为空");}hotelService.updateById(hotel);// 发送修改消息，用修改路由键rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_UPDATE_KEY, hotel.getId());}
      ​// 删除酒店@DeleteMapping("/{id}")public void deleteById(@PathVariable("id") Long id) {hotelService.removeById(id);// 发送删除消息，用删除路由键rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_DELETE_KEY, id);}
      }

      当我们进行消息的发送个时候，我们所需要的参数有。交换机的名字。交换机的路由。需要发送的消息。

消息发送之后，再进行消息的监听。

消息的监听的时候，需要先进行接口的定义，分别定义新增修改和删除的业务接口，可以使用RabbitListener注解进行消息的监听。

 /*** 监听酒店新增或修改队列：从数据库查询最新数据，同步到 Elasticsearch* */@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)@Transactionalpublic void listenHotelInsertOrUpdate(Long id) {hotelRepository.save(hotel);}
​/*** 监听酒店删除队列：从 Elasticsearch 删除对应酒店数据* @param id 酒店ID*/@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)@Transactionalpublic void listenHotelDelete(Long id) {// 从 Elasticsearch 删除酒店hotelRepository.deleteById(id);System.out.println("酒店删除，同步 Elasticsearch 成功，酒店ID：" + id);}
定义过接口之后，我们就可以进行方法的实现@Overridepublic void deleteById(Long id) {// 1. 准备Request：创建删除请求，指定索引和文档IDDeleteRequest request = new DeleteRequest(INDEX_NAME, id.toString());
​// 2. 准备发送请求：执行删除操作（这里直接执行，也可做一些请求参数的额外设置，比如超时等）try {DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);if (response.getResult() != null) {System.out.println("Elasticsearch 文档删除成功，文档ID：" + id + "，结果：" + response.getResult());} else {System.err.println("Elasticsearch 文档删除失败，文档ID：" + id);}} catch (Exception e) {e.printStackTrace();System.err.println("删除 Elasticsearch 文档时发生异常，文档ID：" + id + "，异常信息：" + e.getMessage());}}
​@Overridepublic void insertById(Long id) {// 0. 根据id查询酒店数据：从数据库查询Hotel hotel = hotelDbService.getHotelById(id); // 需实现该方法，返回 Hotel 实体if (hotel == null) {System.err.println("根据ID查询酒店数据为空，ID：" + id);return;}
​// 1. 准备Request：创建索引请求，指定索引IndexRequest request = new IndexRequest(INDEX_NAME);request.id(id.toString()); // 设置文档ID，与酒店ID对应
​try {// 将 Hotel 对象转为 JSON 字符串，作为文档内容String hotelJson = objectMapper.writeValueAsString(hotel);request.source(hotelJson, XContentType.JSON);
​// 2. 准备DSL：这里 DSL 已经通过 request.source 等方式设置好了，//    若是复杂场景，可继续添加路由、超时等参数，比如 request.routing("...")
​// 3. 发送请求：执行索引（新增/更新）操作IndexResponse response = client.index(request, RequestOptions.DEFAULT);System.out.println("Elasticsearch 文档插入/更新成功，文档ID：" + id + "，结果：" + response.getResult());} catch (Exception e) {e.printStackTrace();System.err.println("插入/更新 Elasticsearch 文档时发生异常，文档ID：" + id + "，异常信息：" + e.getMessage());}}