data-service

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
+distributionUrl=https\://dangjian-web.oss-cn-chengdu.aliyuncs.com/lib/gradle-7.5.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

src/main/java/com/sxtvs/open/api/news/controller/YoumeiDataController.java

@@ -31,7 +31,7 @@ public class YoumeiDataController {
 
     @RequestMapping("search")
     @LoginRequired
-    public Page<YoumeiEsData> search(@RequestBody DataRequestDTO dataRequestDTO){
+    public Page<YoumeiEsData> search(@RequestBody DataRequestDTO dataRequestDTO) throws Exception {
         return youmeiDataService.search(dataRequestDTO);
     }
 

src/main/java/com/sxtvs/open/api/news/entity/YoumeiEsData.java

@@ -33,8 +33,6 @@ public class YoumeiEsData{
 
     private String summary;
 
-    private String url;
-
     public YoumeiEsData() {
 
     }
@@ -52,7 +50,6 @@ public class YoumeiEsData{
         data.setSourceWebsite(youmeiData.getSourceWebsite());
         Map<String, Object> mp = getMapFromRaw(youmeiData.getRawData());
         data.setSummary((String) mp.get("summary"));
-        data.setUrl((String) mp.get("url"));
         data.setCategory(youmeiData.getCategory());
         return data;
     }

src/main/java/com/sxtvs/open/api/news/service/impl/YoumeiDataServiceImpl.java

@@ -22,6 +22,7 @@ import com.sxtvs.open.api.news.entity.YoumeiOffset;
 import com.sxtvs.open.api.news.mapper.YoumeiDataMapper;
 import com.sxtvs.open.api.news.service.IYoumeiDataService;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.sxtvs.open.core.es.EsClientPool;
 import com.sxtvs.open.core.sls.AliyunLogger;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
@@ -47,8 +48,8 @@ import java.util.Map;
 @Service
 @Slf4j
 public class YoumeiDataServiceImpl extends ServiceImpl<YoumeiDataMapper, YoumeiData> implements IYoumeiDataService {
-    @Resource
-    private ElasticsearchClient elasticsearchClient;
+//    @Resource
+//    private ElasticsearchClient elasticsearchClient;
 
     @Resource
     private YoumeiOffsetServiceImpl youmeiOffsetService;
@@ -56,9 +57,13 @@ public class YoumeiDataServiceImpl extends ServiceImpl<YoumeiDataMapper, YoumeiD
     @Resource
     private AliyunLogger logger;
 
+    @Resource
+    private EsClientPool esClientPool;
+
 
-    public Page<YoumeiEsData> search(DataRequestDTO dataRequestDTO) {
+    public Page<YoumeiEsData> search(DataRequestDTO dataRequestDTO) throws Exception {
 
+        ElasticsearchClient elasticsearchClient = esClientPool.getClient();
         SearchResponse<YoumeiEsData> response = null;
         Page<YoumeiEsData> page = new Page<>(dataRequestDTO.getPage(), dataRequestDTO.getPageSize());
         int offset = (int) ((page.getCurrent() - 1) * page.getSize());
@@ -121,35 +126,38 @@ public class YoumeiDataServiceImpl extends ServiceImpl<YoumeiDataMapper, YoumeiD
     }
 
     public void putData() {
+
         log.info("ES 入数据开始");
         YoumeiOffset youmeiOffset = youmeiOffsetService.getById(4);
         long offset = youmeiOffset.getOffset() + 1L;
         long max = 0L;
-        while (true) {
-            List<CategoryData> youmeiDataList = baseMapper.getCategoryList(offset, 1000);
-            log.info("本次数据{}条，offset从{}开始", youmeiDataList.size(), offset);
-            if (youmeiDataList.size() == 0) {
-                break;
-            }
-            max = youmeiDataList.get(youmeiDataList.size() - 1).getOffset();
-            List<BulkOperation> bulkOperations = new ArrayList<>();
-
-            youmeiDataList.forEach(a -> {
-                YoumeiEsData youmeiEsData = YoumeiEsData.of(a);
-                bulkOperations.add(BulkOperation.of(b -> b.index(c -> c.id(String.valueOf(youmeiEsData.getOffset())).document(youmeiEsData))));
-            });
+        try {
+            ElasticsearchClient elasticsearchClient = esClientPool.getClient();
+            while (true) {
+                List<CategoryData> youmeiDataList = baseMapper.getCategoryList(offset, 1000);
+                log.info("本次数据{}条，offset从{}开始", youmeiDataList.size(), offset);
+                if (youmeiDataList.size() == 0) {
+                    break;
+                }
+                max = youmeiDataList.get(youmeiDataList.size() - 1).getOffset();
+                List<BulkOperation> bulkOperations = new ArrayList<>();
+
+                youmeiDataList.forEach(a -> {
+                    YoumeiEsData youmeiEsData = YoumeiEsData.of(a);
+                    bulkOperations.add(BulkOperation.of(b -> b.index(c -> c.id(String.valueOf(youmeiEsData.getOffset())).document(youmeiEsData))));
+                });
 
-            try {
                 elasticsearchClient.bulk(BulkRequest.of(x -> x.index("news_data").operations(bulkOperations)));
-            } catch (IOException e) {
-                logger.error("error", e);
-            }
 
-            if (youmeiDataList.size() < 1000) {
-                break;
-            }
+                if (youmeiDataList.size() < 1000) {
+                    break;
+                }
 
-            offset += 1000;
+                offset += 1000;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            logger.error("error", e);
         }
         if (max > 0L) {
             youmeiOffset.setOffset(max);

src/main/java/com/sxtvs/open/core/conf/EsClientConfig.java

@@ -6,6 +6,12 @@ import co.elastic.clients.transport.ElasticsearchTransport;
 import co.elastic.clients.transport.rest_client.RestClientTransport;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.pool2.BasePooledObjectFactory;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.http.Header;
 import org.apache.http.HttpHost;
 import org.apache.http.auth.AuthScope;
@@ -20,6 +26,8 @@ import org.apache.http.ssl.SSLContexts;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -43,7 +51,7 @@ import java.security.cert.CertificateFactory;
 @ConfigurationProperties(prefix = "elasticsearch") //配置的前缀
 @Configuration
 @Slf4j
-public class EsClientConfig {
+public class EsClientConfig  {
     @Setter
     private String hosts;
 
@@ -56,6 +64,16 @@ public class EsClientConfig {
     @Setter
     private String apikey;
 
+    @Setter
+    private int maxTotal;
+
+    @Setter
+    private int maxIdle;
+
+    @Setter
+    private int minIdle ;
+
+
     /**
      * 解析配置的字符串，转为HttpHost对象数组
      * @return
@@ -126,7 +144,7 @@ public class EsClientConfig {
         return new RestClientTransport(client, new JacksonJsonpMapper());
     }
 
-    /*private static ElasticsearchTransport getElasticsearchTransport(String apiKey, HttpHost...hosts) {
+    private static ElasticsearchTransport getElasticsearchTransport(String apiKey, HttpHost...hosts) {
         // 将ApiKey放入header中
         Header[] headers = new Header[] {new BasicHeader("Authorization", "ApiKey " + apiKey)};
 
@@ -149,5 +167,44 @@ public class EsClientConfig {
     public ElasticsearchClient clientByApiKey() throws Exception {
         ElasticsearchTransport transport = getElasticsearchTransport(apikey, toHttpHost());
         return new ElasticsearchClient(transport);
-    }*/
+    }
+
+    @Bean
+    public GenericObjectPoolConfig<ElasticsearchClient> config() {
+        GenericObjectPoolConfig<ElasticsearchClient> poolConfig = new GenericObjectPoolConfig<>();
+        poolConfig.setMinIdle(minIdle);
+        poolConfig.setMaxTotal(maxTotal);
+        poolConfig.setMaxIdle(maxIdle);
+        poolConfig.setJmxEnabled(false);
+        return poolConfig;
+    }
+
+    @Bean
+    public GenericObjectPool<ElasticsearchClient> pool(
+            PooledObjectFactory<ElasticsearchClient> factory,
+            GenericObjectPoolConfig<ElasticsearchClient> config) {
+        return new GenericObjectPool<>(factory, config);
+    }
+
+    @Bean
+    public PooledObjectFactory<ElasticsearchClient> factory(){
+        return new BasePooledObjectFactory<>() {
+            @Override
+            public ElasticsearchClient create() {
+                ElasticsearchTransport transport = getElasticsearchTransport(apikey, toHttpHost());
+                return new ElasticsearchClient(transport);
+            }
+
+            @Override
+            public PooledObject<ElasticsearchClient> wrap(ElasticsearchClient obj) {
+                return new DefaultPooledObject<>(obj);
+            }
+
+            @Override
+            public void destroyObject(PooledObject<ElasticsearchClient> pooledObject) throws Exception {
+                ElasticsearchClient highLevelClient = pooledObject.getObject();
+
+            }
+        };
+    }
 }

src/main/java/com/sxtvs/open/core/es/EsClientPool.java

@@ -0,0 +1,34 @@
+package com.sxtvs.open.core.es;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import jakarta.annotation.Resource;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.springframework.stereotype.Component;
+
+@Component
+public class EsClientPool {
+    @Resource
+    private GenericObjectPool<ElasticsearchClient> genericObjectPool;
+
+    /**
+     * 获得对象
+     *
+     * @return
+     * @throws Exception
+     */
+    public ElasticsearchClient getClient() throws Exception {
+        // 从池中取一个对象
+        return genericObjectPool.borrowObject();
+    }
+
+    /**
+     * 归还对象
+     *
+     * @param client
+     */
+    public void returnClient(ElasticsearchClient client) {
+        // 使用完毕之后，归还对象
+        genericObjectPool.returnObject(client);
+    }
+
+}

src/main/resources/application.yml

@@ -64,3 +64,6 @@ elasticsearch:
   username: elastic
   passwd: Iqyb_f3RJ6V16IG3lS0t
   apikey: RGM0S1hvWUI0aUtoa3RTR245SHo6cm9wZDBOUVBTaFM4UENHZE9BTUt3dw==
+  max-total: 8
+  max-idle: 8
+  min-idle: 1

src/test/data-service.http

@@ -17,15 +17,17 @@ Content-Type: application/json;charset=UTF-8
 Authorization: b6af95077bafeccb32ae7a65dadee47d
 
 {
-"keywords": "刘三姐扮演者黄婉秋病危昏迷",
-  "category": "分类",
-  "city": "城市",
-  "start": "2023-02-20 15:52:52",
-  "end": "2023-02-22 15:52:52",
+"keywords": "",
+  "city": "",
+  "start": "2023-01-01 15:52:52",
+  "end": "2023-02-23 15:52:52",
   "page": 0,
   "pageSize": 10
 }
 
 ### 数据入es
 GET http://localhost/news/data/put/MWEZOWQ2NDYXMJU0NW
-Authorization: b6af95077bafeccb32ae7a65dadee47d
+
+
+### 分类
+GET http://localhost/news/data/category