【設計模式】【第六章】【查詢ElasticSearch 大量資料場景】【疊代器模式】

文章目錄

• ​​建立design-demo項目​​

• ​​建立EsController​​
• ​​建立EsService​​
• ​​建立EsServiceimpl​​
• ​​建立EsQueryProcessor​​
• ​​建立EsSqlQuery​​
• ​​建立EsSqlResult​​

建立design-demo項目

項目代碼：​​https://gitee.com/java_wxid/java_wxid/tree/master/demo/design-demo​​ 項目結構如下（示例）：

建立EsController

代碼如下（示例）：

package com.example.designdemo.controller;

import com.example.designdemo.service.EsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class EsController {
    @Autowired
    private EsService esService;

    @PostMapping("es")
    public Boolean query(@RequestParam String query, Long fetchSize) {

        return esService.query(query, fetchSize);
    }

}
      

建立EsService

代碼如下（示例）：

package com.example.designdemo.service;

/**
 * @author zhiwei Liao
 * @Description
 * @Date create in 2022/9/12 0012 21:03
 */
public interface EsService {

    Boolean query(String query, Long fetchSize);
}
      

建立EsServiceimpl

代碼如下（示例）：

package com.example.designdemo.service.impl;

import com.example.designdemo.esquery.EsQueryProcessor;
import com.example.designdemo.service.EsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.stream.Stream;

@Service
public class EsServiceimpl implements EsService {

    @Autowired
    private EsQueryProcessor esQueryProcessor;
    
    public Boolean query(String query, Long fetchSize) {
        Stream<Map<String, Object>> mapStream = esQueryProcessor
                .scrollEsStream(query, fetchSize);
        mapStream.forEach(x -> System.out.println(x));
        return true;
    }
}
      

建立EsQueryProcessor

代碼如下（示例）：

package com.example.designdemo.esquery;

import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
/**
 * @Author: zhiwei Liao
 * @Date: 2022/9/29 21:29
 * @Description:
 */
@Component
public class EsQueryProcessor {
    //1. 我們要用stream 傳回 為了節省記憶體
    public Stream<Map<String, Object>> scrollEsStream(String query, Long fetchSize) {
        return StreamSupport.stream(Spliterators
                .spliteratorUnknownSize(new ScrollIterator(query, fetchSize), 0), false);
    }


    //2. 我們要 疊代器
    private class ScrollIterator implements Iterator<Map<String, Object>> {
        private String scrollId;
        private List<String> columns;
        Iterator<Map<String, Object>> iterator;
        RestTemplate restTemplate = new RestTemplate(); // 真是項目中使用resttemplate的時候
        //一定是進行過我們的 bean 配置注入的。這裡邊直接用new關鍵字是為了通路我們的es 接口。

        //構造函數進行第一次查詢，并且初始化我們後續需要使用的 columns 和 iterator 和 scroll
        public ScrollIterator(String query, Long fetchSize) {
            EsSqlResult esSqlResult = restTemplate.postForObject("http://localhost:9200/_sql?format=json",
                    new EsSqlQuery(query, fetchSize), EsSqlResult.class);//第一次通路的結果出來了
            this.scrollId = esSqlResult.getCursor();
            this.columns = esSqlResult.getColumns()
                    .stream().map(x->x.get("name"))
                    .collect(Collectors.toList());
            this.iterator = convert(columns, esSqlResult).iterator();
        }

        // hasNext 根據 是否 scrollId 為null進行後續的 第二次，第三次，，，的通路，直到 scrollId 為null
        @Override
        public boolean hasNext() {
            return iterator.hasNext() || scrollNext();
        }
        private boolean scrollNext() {
            if(iterator == null || this.scrollId == null) {
                return false;
            }
            EsSqlResult esSqlResult = restTemplate.postForObject("http://localhost:9200/_sql?format=json",
                    new EsSqlQuery(this.scrollId), EsSqlResult.class);//第二次通路的結果出來了
            this.scrollId = esSqlResult.getCursor();
            this.iterator = convert(columns, esSqlResult).iterator();
            return iterator.hasNext();
        }

        @Override
        public Map<String, Object> next() {
            return iterator.next();
        }
    }



    //3. 傳回結果傳統一點 List<map>
    private List<Map<String, Object>> convert(List<String> columns, EsSqlResult esSqlResult) {
        List<Map<String, Object>> results = new ArrayList<>();
        for(List<Object> row : esSqlResult.getRows()) {
            Map<String, Object> map = new HashMap<>();
            for(int i = 0; i < columns.size(); i++) {
                map.put(columns.get(i), row.get(i));
            }
            results.add(map);
        }
        return results;
    }
}
      

建立EsSqlQuery

package com.example.designdemo.esquery;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
/**
 * @Author: zhiwei Liao
 * @Date: 2022/9/29 21:29
 * @Description:
 */
@JsonIgnoreProperties
public class EsSqlQuery {
    private String query;
    private Long fetchSize;
    private String cursor;

    public EsSqlQuery(String cursor) {
        this.cursor = cursor;
    }

    public EsSqlQuery(String query, Long fetchSize) {
        this.query = query;
        this.fetchSize = fetchSize;
    }

    public String getQuery() {
        return query;
    }

    public void setQuery(String query) {
        this.query = query;
    }

    public Long getFetchSize() {
        return fetchSize;
    }

    public void setFetchSize(Long fetchSize) {
        this.fetchSize = fetchSize;
    }

    public String getCursor() {
        return cursor;
    }

    public void setCursor(String cursor) {
        this.cursor = cursor;
    }
}
      

建立EsSqlResult

package com.example.designdemo.esquery;

import java.util.List;
import java.util.Map;
/**
 * @Author: zhiwei Liao
 * @Date: 2022/9/29 21:29
 * @Description:
 */
public class EsSqlResult {
    private List<Map<String, String>> columns;
    private List<List<Object>> rows;
    private String cursor;

    public List<Map<String, String>> getColumns() {
        return columns;
    }

    public void setColumns(List<Map<String, String>> columns) {
        this.columns = columns;
    }

    public List<List<Object>> getRows() {
        return rows;
    }

    public void setRows(List<List<Object>> rows) {
        this.rows = rows;
    }

    public String getCursor() {
        return cursor;
    }

    public void setCursor(String cursor) {
        this.cursor = cursor;
    }
}