ElasticSearch + logstash (logstash-input-jdbc) + mysql 同步数据（二）

秒速五厘米 2022-05-19 03:20 298阅读 0赞

> ## **ElasticSearch + logstash (logstash-input-jdbc) + mysql 同步数据（二）** ##

### **应对上篇 elasticsearch + logstash + mysql同步数据服务实现，本篇介绍如何应用elasticsearch查询同步 mysql数据应用在springWeb中：** ###

> **1.elasticsearch开启的服务的版本是5.6.3 这里对应的引入包版本号应该一致：**

<dependency>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>transport</artifactId>
                    <version>5.6.3</version>
                </dependency>

> **2.调用elasticsearch API 查询方法总结**

import com.caox.sharding.utils.DateExtendUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.elasticsearch.action.get.GetResponse;
    import org.elasticsearch.action.index.IndexResponse;
    import org.elasticsearch.action.search.SearchRequestBuilder;
    import org.elasticsearch.action.search.SearchResponse;
    import org.elasticsearch.client.transport.TransportClient;
    import org.elasticsearch.common.settings.Settings;
    import org.elasticsearch.common.transport.InetSocketTransportAddress;
    import org.elasticsearch.common.unit.TimeValue;
    import org.elasticsearch.common.xcontent.XContentFactory;
    import org.elasticsearch.index.query.BoolQueryBuilder;
    import org.elasticsearch.index.query.QueryBuilder;
    import org.elasticsearch.index.query.QueryBuilders;
    import org.elasticsearch.search.SearchHit;
    import org.elasticsearch.transport.client.PreBuiltTransportClient;
    import org.junit.After;
    import org.junit.Before;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.net.InetAddress;
    import java.net.UnknownHostException;
    import java.util.Date;
    import java.util.Map;
    
    /**
     * Created by nazi on 2018/7/19.
     */
    @Slf4j
    public class ElasticSearchTest01 {
    
        private static final String INDEX = "article"; // 索引
        private static final String TYPE = "contact"; // 类型
    
        public final static String HOST = "127.0.0.1";
        public final static int PORT = 9300;//http请求的端口是9200，客户端是9300
        private TransportClient client = null;
    
        @SuppressWarnings({ "resource", "unchecked" })
        @Before
        public void getConnect() throws UnknownHostException {
            client = new PreBuiltTransportClient(Settings.EMPTY).addTransportAddresses(
                    new InetSocketTransportAddress(InetAddress.getByName(HOST),PORT));
            System.out.println("连接信息:" + client.toString());
        }
    
        @After
        public void closeConnect() {
            if(null != client) {
                System.out.println("执行关闭连接操作...");
                client.close();
            }
        }
    
        @SuppressWarnings("resource")
         @Test
        public void test1() throws UnknownHostException {
          //创建客户端
          TransportClient client = new PreBuiltTransportClient(Settings.EMPTY).addTransportAddresses(
                  new InetSocketTransportAddress(InetAddress.getByName(HOST),PORT));
            System.out.println("Elasticsearch connect info:" + client.toString());
            System.out.println("hello world!");
    
        //关闭客户端
           client.close();
    
        }
    
        @Test
        public void addIndex1() throws IOException {
            IndexResponse response = client.prepareIndex("msg", "tweet", "1").setSource(XContentFactory.jsonBuilder()
                    .startObject().field("userName", "张三")
                    .field("sendDate", new Date())
                    .field("msg", "你好李四")
                    .endObject()).get();
    
            System.out.println("索引名称:" + response.getIndex() + "\n类型:" + response.getType()
                    + "\n文档ID:" + response.getId() + "\n当前实例状态:" + response.status());
        }
    
        @Test
        public void getData1() {
            GetResponse getResponse = client.prepareGet("article", "jdbc", "8").get();
            System.out.println("索引库的数据:" + getResponse.getSourceAsString());
        }
    
        /**
         * 列表查询
         */
        @Test
        public void queryList() {
            try {
                String key = "8";
                QueryBuilder queryBuilder = QueryBuilders.multiMatchQuery(key, "name", "age","id","student_id");
    
                SearchResponse res = client.prepareSearch().setIndices(INDEX).setTypes(TYPE).setQuery(queryBuilder).get();
    
                System.out.println("查询到的总记录个数为：" + res.getHits().getTotalHits());
                for (int i = 0; i < res.getHits().getTotalHits(); i++) {
                   System.out.println("第" + (i + 1) + "条记录主要内容为：" + res.getHits().getAt(i).getSource().toString());
                }
            } catch (Exception e) {
               log.error("查询列表失败！:{}",e);
            }
        }
        /**
         * 单个精确值查找(termQuery)
         */
        @Test
        public void termQuery() {
            QueryBuilder queryBuilder = QueryBuilders.termQuery("age", 21);
    //        queryBuilder = QueryBuilders.termQuery("isDelete", true);
    //        queryBuilder = QueryBuilders.termQuery("my_title", "我的标题12323abcd");
            searchFunction(queryBuilder);
        }
    
        /**
         * 多个值精确查找(termsQuery)
         *
         * 一个查询相匹配的多个value
         */
        @Test
        public void termsQuery() {
            QueryBuilder queryBuilder = QueryBuilders.termsQuery("hehe","");
            searchFunction(queryBuilder);
        }
    
        /**
         * 查询相匹配的文档在一个范围(rangeQuery) 时间范围
         * 注意 这里的时间范围仅允许是UTC时间格式
         */
        @Test
        public void rangeQuery() {
            QueryBuilder queryBuilder = QueryBuilders
                    .rangeQuery("createtime") // 查询code字段
                    .from("2018-07-19T00:00:00.000Z")
                    .to("2018-07-19T23:59:59.000Z")
                    .includeLower(true)     //  包括下界
                    .includeUpper(true);   // 不包括上界
            searchFunction(queryBuilder);
        }
    
        /**
         * 查询相匹配的文档在一个范围(prefixQuery)
         */
        @Test
        public void prefixQuery() {
            QueryBuilder queryBuilder = QueryBuilders.prefixQuery("name", "hehe");
            searchFunction(queryBuilder);
        }
    
        /**
         * 通配符检索(wildcardQuery)
         *
         * 值使用用通配符，常用于模糊查询
         *
         * 匹配具有匹配通配符表达式（ (not analyzed ）的字段的文档。 支持的通配符：
         *     *，它匹配任何字符序列（包括空字符序列）
         *     ?，它匹配任何单个字符。
         *
         * 请注意，此查询可能很慢，因为它需要遍历多个术语。 为了防止非常慢的通配符查询，通配符不能以任何一个通配符*或？开头。
         */
        @Test
        public void wildcardQuery() {
            QueryBuilder queryBuilder = QueryBuilders.wildcardQuery("name", "*ehe*");
            queryBuilder = QueryBuilders.wildcardQuery("name", "*ehe");
    //        queryBuilder = QueryBuilders.wildcardQuery("name", "?ehe");
            searchFunction(queryBuilder);
        }
    
        /**
         * 正则表达式检索(regexpQuery) 不需要^、$
         */
        @Test
        public void regexpQuery() {
            QueryBuilder queryBuilder = QueryBuilders.regexpQuery("name", "he(he)?");
            searchFunction(queryBuilder);
        }
    
        /**
         * 使用模糊查询匹配文档查询(fuzzyQuery)
         */
        @Test
        public void fuzzyQuery() {
            QueryBuilder queryBuilder = QueryBuilders.fuzzyQuery("name", "heh");
            searchFunction(queryBuilder);
        }
        /**
         * Ids检索, 返回指定id的全部信息 (idsQuery)
         *
         * 在idsQuery(type)方法中，也可以指定具体的类型
         */
        @Test
        public void idsQuery() {
            QueryBuilder queryBuilder = QueryBuilders.idsQuery().addIds("8", "9", "10");
            searchFunction(queryBuilder);
        }
    
        /************************************************************ 全文检索 ************************************************************/
        /**
         * 单个匹配 (matchQuery)
         *
         * 感觉跟termQuery效果一样
         */
        @Test
        public void matchQuery() {
            QueryBuilder queryBuilder = QueryBuilders.matchQuery("name", "hehe");
            searchFunction(queryBuilder);
        }
    
        /**
         * 查询匹配所有文件 (matchAllQuery) 查询所有
         */
        @Test
        public void matchAllQuery() {
            QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();
            searchFunction(queryBuilder);
        }
    
        /**
         * 匹配多个字段, field可以使用通配符(multiMatchQuery)
         * 注意：搜索的 text 比如 8 要和 后面搜索的字段"id"、"age"、"student_id" 字段的类型对应
         */
        @Test
        public void multiMatchQuery() {
            QueryBuilder queryBuilder = QueryBuilders.multiMatchQuery("8", "id", "age", "student_id");
            queryBuilder = QueryBuilders.multiMatchQuery("hehe", "*name"); //字段使用通配符
            queryBuilder = QueryBuilders.multiMatchQuery("8", "*id"); //字段使用通配符
            searchFunction(queryBuilder);
        }
    
        /**
         * 字符串检索(queryString)
         *
         * 一个使用查询解析器解析其内容的查询。
         *  query_string查询提供了以简明的简写语法执行多匹配查询 multi_match queries ，布尔查询 bool queries ，提升得分 boosting ，模糊
         *  匹配 fuzzy matching ，通配符 wildcards，正则表达式 regexp 和范围查询 range queries 的方式。
         *
         *  支持参数达10几种
         */
        @Test
        public void queryString() {
    //        QueryBuilder queryBuilder = QueryBuilders.queryStringQuery("*hehe");   //通配符查询
            QueryBuilder queryBuilder = QueryBuilders.queryStringQuery("hehe");
            searchFunction(queryBuilder);
        }
        /**
         * 多个条件查询 匹配条件
         * must 相当于and，就是都满足
         * should 相当于or，满足一个或多个
         * must_not 都不满足
         */
        @Test
        public void testQueryBuilder2() {
            // "科技视频"分词的结果是"科技", "视频", "频"
    
            // 通配符查询
            BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
            /**
             * and
             */
    //        queryBuilder.must(QueryBuilders.matchQuery("name", "hehe"));
    //        queryBuilder.must(QueryBuilders.matchQuery("student_id", 8));
    
            /**
             * or
             */
    //        queryBuilder.should(QueryBuilders.matchQuery("name", "hehe"));
    //        queryBuilder.should(QueryBuilders.matchQuery("student_id", 8));
    //        queryBuilder.minimumShouldMatch(2); // 最少匹配数
    
            /**
             * not
             */
            queryBuilder.mustNot(QueryBuilders.matchQuery("name", "hehe"));
            queryBuilder.mustNot(QueryBuilders.matchQuery("student_id", 8));
            searchFunction(queryBuilder);
        }
    
    
        /**
         * 类型检索(typeQuery)
         *
         * 查询该类型下的所有数据
         */
        @Test
        public void typeQuery() {
            QueryBuilder queryBuilder = QueryBuilders.typeQuery(TYPE);
            searchFunction(queryBuilder);
        }
    
        /**
         * 查询遍历抽取
         *
         * 查询结果是根据分值排序(从大到小)
         *
         * @param queryBuilder
         */
        private void searchFunction(QueryBuilder queryBuilder) {
            SearchRequestBuilder requestBuilder = client.prepareSearch().setIndices(INDEX).setTypes(TYPE)
                    .setScroll(new TimeValue(60000)).setQuery(queryBuilder);
            // 相当于分页参数  setFrom => pageNo; setSize=> pageCount
            SearchResponse response = requestBuilder.setFrom(0).setSize(100).execute().actionGet();
            System.out.println("--------------查询结果：----------------------");
            // 查询出记录的总条数
            System.out.println("总共的条数："+response.getHits().totalHits);
            System.out.println("--------------显示列表：----------------------");
            for (SearchHit hit : response.getHits()) {
                System.out.println("分值：" + hit.getScore()); // 相关度
                Map<String, Object> map = hit.getSource();
                for (String sKey : map.keySet()) {
                    System.out.println(sKey + ": " + map.get(sKey));
                }
                System.out.println("--------------");
            }
            System.out.println("-----------------------------------");
        }
    
        @Test
        public void test2() throws Exception{
            System.out.println(DateExtendUtil.formatDate2UTCDateString("2018-07-19 00:00:00"));
        }
    }

> ** 3.注意问题 **

（1）时间范围查询 这里的时间格式必须是UTC标准格式，否则出错

public void rangeQuery() {
            QueryBuilder queryBuilder = QueryBuilders
                    .rangeQuery("createtime") // 查询code字段
                    .from("2018-07-19T00:00:00.000Z")
                    .to("2018-07-19T23:59:59.000Z")
                    .includeLower(true)     //  包括下界
                    .includeUpper(true);   // 不包括上界
            searchFunction(queryBuilder);
        }

（2）其中查询遍历的公共方法：

private void searchFunction(QueryBuilder queryBuilder) {
            SearchRequestBuilder requestBuilder = client.prepareSearch().setIndices(INDEX).setTypes(TYPE)
                    .setScroll(new TimeValue(60000)).setQuery(queryBuilder);
            // 相当于分页参数  setFrom => pageNo; setSize=> pageCount
            SearchResponse response = requestBuilder.setFrom(0).setSize(100).execute().actionGet();
            System.out.println("--------------查询结果：----------------------");
            // 查询出记录的总条数
            System.out.println("总共的条数："+response.getHits().totalHits);
            System.out.println("--------------显示列表：----------------------");
            for (SearchHit hit : response.getHits()) {
                System.out.println("分值：" + hit.getScore()); // 相关度
                Map<String, Object> map = hit.getSource();
                for (String sKey : map.keySet()) {
                    System.out.println(sKey + ": " + map.get(sKey));
                }
                System.out.println("--------------");
            }
            System.out.println("-----------------------------------");
        }

对应分页查询 参数：

起始页码setFrom => pageNo; 每页数量setSize=> pageCount

记录的总条数：response.getHits().totalHits

（3）时间格式处理的公共方法

public final static String UTC_DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
    
    
    /**
    * 将日期Str转换成UTC格式的时间字符串
    */
    public static String formatDate2UTCDateString(String dateStr) throws Exception {
    	Date date = parseString2Date(dateStr);
    	return formatDate2String(date, UTC_DATE_FORMAT);
    }
    
    
    /**
    * 将日期转换成指定格式的字符串
    * 
    * @param date
    * @param formatString
    * @return
    */
    public static String formatDate2String(Date date, String formatString) {
    	return new SimpleDateFormat(formatString).format(date);
    }