使用Feign 的远程调用,把mysql数据导入es

要把数据库数据导入到elasticsearch中，包括下面几步：

1）将商品微服务中的分页查询商品接口定义为一个FeignClient，放到feign-api模块中

2）搜索服务编写一个测试业务，实现下面功能：

调用item-service提供的FeignClient，分页查询商品 PageDTO<Item>

将查询到的商品封装为一个ItemDoc对象，放入ItemDoc集合

将ItemDoc集合批量导入elasticsearch中

注意：数据库中的商品数量多达9万多个，不可查询索引导入。一定要分页导入。

第一步，分页查询item接口对外暴露，在feign-api中定义接口

 @FeignClient("itemservice")
public interface ItemFeignClient {@GetMapping(path = "/item/list")public ResponseEntity<PageDTO<Item>> list(@RequestParam(value = "page", defaultValue = "1") Integer page, @RequestParam(value = "size", defaultValue = "10") Integer size);}

第二步 ，在es服务和数据服务分别引入fegin-api依赖

<dependency><groupId>com.hmall</groupId><artifactId>feign-api</artifactId><version>1.0</version>
</dependency>

es服务配置文件:

spring:application:name: searchserviceelasticsearch:rest:uris: http://192.168.xxx.xxx:xxxx

第三步，es服务入口类中调用feign,要开启feign扫描

@SpringBootApplication
@EnableFeignClients(basePackages = "com.hmall.api")
public class SearchApplication {public static void main(String[] args) {SpringApplication.run(SearchApplication.class, args);}
}

第四步，编写一个测试类，分页查询并批量导入

@SpringBootTest
@RunWith(SpringRunner.class) //Junit 4 需要  ,  Junit 5 不需要
public class BulkItemImportTest {@Autowiredprivate ItemFeignClient itemFeignClient;@Autowiredprivate RestHighLevelClient restHighLevelClient;@Testpublic void bulkItemImportTest() throws IOException {Integer page = 1, size = 1000;while (true) {//1. 查询商品列表ResponseEntity<PageDTO<Item>> responseEntity = itemFeignClient.list(page, size);PageDTO<Item> pageDTO = responseEntity.getBody();//如果查询的数据未空代表, 所有数据已经完成导入, 终止循环if (pageDTO.getList() == null || pageDTO.getList().size() == 0) {break;}//2. 导入数据到ES//2.1 创建请求对象BulkRequest request = new BulkRequest();//2.2 封装DSL语句pageDTO.getList().stream().forEach(item -> {ItemDoc itemDoc = new ItemDoc(item);request.add(new IndexRequest("hmall_item").id(item.getId() + "").source(JSON.toJSONString(itemDoc), XContentType.JSON));});//2.3 发送请求restHighLevelClient.bulk(request, RequestOptions.DEFAULT);page++;}}
}