从零开发短视频电商 Low Level Client(推荐)连接OpenSearch进行CRUD

2023-12-10 08:44
文章标签 进行 视频 推荐 连接 开发 low level client 电商 crud opensearch

本文主要是介绍从零开发短视频电商 Low Level Client(推荐)连接OpenSearch进行CRUD，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

      • 依赖
      • 初始化客户端
      • 发起请求
      • 请求参数
      • 请求头
      • 设置超时时间
      • 设置线程数
      • 设置用户名密码
      • 结果解析
      • 节点选择器
      • 配置嗅探器
      • 整体示例
      • 问题
      • 参考

OpenSearch开发环境安装Docker和Docker-Compose两种方式

依赖

<dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-client</artifactId><version>7.13.4</version>  <!-- 建议就是这个版本 -->
</dependency>
<!-- 或者 -->
<dependency><groupId>org.opensearch.client</groupId><artifactId>opensearch-java</artifactId><version>2.8.1</version>
</dependency>

初始化客户端

// 构建客户端
RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200, "http"),new HttpHost("localhost", 9201, "http")).build();

发起请求

  • performRequest： 是同步请求方法: 将阻塞调用线程，并在请求成功时返回响应，或在请求失败时引发异常
  • performRequestAsync: 是异步方法：接收一个ResponseListener对象作为参数。如果请求成功，则该参数使用响应进行调用；如果请求失败，则使用异常进行调用
// 同步请求Request request = new Request("GET","/posts/_search");Response response = restClient.performRequest(request); // 执行同步请求response.toString();// 异步请求Request request = new Request("GET", "/posts/_search");restClient.performRequestAsync(request, new ResponseListener() {@Overridepublic void onSuccess(Response response) {log.info("异步请求成功!" + response.toString());}@Overridepublic void onFailure(Exception e) {log.error("异步请求失败!");e.printStackTrace();}});

请求参数

// 第一种
request.addParameter("pretty","true");
// 第二种
request.setEntity(new NStringEntity("{\"json\":\"text\"}",ContentType.APPLICATION_JSON));
// 第三种
request.setJsonEntity("{\"json\":\"text\"}");

请求头

RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();builder.addHeader("Authorization", "Bearer " + "my-token");builder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024 * 1024));COMMON_OPTIONS = builder.build();
Request request = new Request("GET", "/");
request.setOptions(COMMON_OPTIONS);

设置超时时间

 RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {@Overridepublic RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder builder) {return builder.setConnectTimeout(50000) // 连接超时默认1s .setSocketTimeout(10000); // 套接字超时默认30s.setConnectionRequestTimeout(10000);}});

设置线程数

Apache HTTP异常客户端默认启动一个调度程序线程，连接管理器使用多个工作线程。

        RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200)).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {return httpAsyncClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(threadNumber).build());}});

设置用户名密码

// 创建凭证提供程序，设置用户名和密码
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("admin", "admin"));// 使用 RestClient 构建器连接到 OpenSearch
RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200, "http")).setHttpClientConfigCallback(httpClientBuilder -> {// 配置连接超时，连接建立后两个节点之间数据传输的套接字超时和连接请求超时// 连接超时：客户端和服务器建立连接的最长时间RequestConfig.Builder requestConfigBuilder = RequestConfig.custom().setConnectTimeout(5000)  // 连接超时为5秒.setSocketTimeout(10000) // 套接字超时为10秒.setConnectionRequestTimeout(10000); // 连接请求超时为10秒httpClientBuilder.setDefaultRequestConfig(requestConfigBuilder.build());// 设置凭证提供程序httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);return httpClientBuilder;}).build();

结果解析

            Response response = restClient.performRequest(new Request("GET", "/"));// 已执行请求的信息RequestLine requestLine = response.getRequestLine();// Host返回的信息HttpHost httpHost = response.getHost();// 响应状态行，从中解析状态代码int statusCode = response.getStatusLine().getStatusCode();// 响应头，可以通过getHeader(string)按名称获取Header[] headers = response.getHeaders();String responseBody = EntityUtils.toString(response.getEntity());

节点选择器

在默认情况下，客户端以轮询的方式将每个请求发送到配置的各个节点中
ES允许用户自由选择要连接的节点，通过初始化客户端来配置节点选择器，以便筛选节点。该功能在启用嗅探器时可以用来防止HTTP请求只命中专用的主节点。
配置后，对于每个请求，客户端都通过节点选择器来筛选备选节点。

        RestClientBuilder builder = RestClient.builder(new HttpHost("localhost",9200,"http"));builder.setNodeSelector(new NodeSelector(){@Overridepublic void select(Iterable<Node> nodes){boolean foundOne = false;for(Node node : nodes){String rackId = node.getAttributes().get("rack_id").get(0);if("targetId".equals(rackId)){foundOne = true;break;}}if(foundOne){Iterator<Node> nodesIt = nodes.iterator();while(nodesIt.hasNext()){Node node = nodesIt.next();String rackId = node.getAttributes().get("rack_id").get(0);if("targetId".equals(rackId) == false){nodesIt.remove();}}}}

配置嗅探器

嗅探器允许自动发现运行中ES集群中的节点，并将其设置为现有的RestClient实例
默认i情况下，嗅探器使用nodes info API检索属于集群的节点并采用jackson解析获得JSON响应

  <dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-client-sniffer</artifactId><version>${elasticsearch.version}</version></dependency>

创建RestClient实例就可以采用嗅探器与其互联。嗅探器利用RestClient提供的定期机制（默认定期时间为5min)，从集群中获取当前节点的列表，通过调用RestClient类中的setNodes方法来更新。

整体示例

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseException;
import org.elasticsearch.client.RestClient;import java.io.IOException;public class OpenSearchExample {public static void main(String[] args) throws IOException {// Connect to OpenSearchfinal CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("admin", "admin"));RestClient restClient = RestClient.builder(new HttpHost("10.12.23.1", 9200, "http")).setHttpClientConfigCallback(httpClientBuilder -> {RequestConfig.Builder requestConfigBuilder = RequestConfig.custom().setConnectTimeout(5000).setSocketTimeout(10000).setConnectionRequestTimeout(10000);httpClientBuilder.setDefaultRequestConfig(requestConfigBuilder.build());httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);return httpClientBuilder;}).build();try {// Delete IndexdeleteIndex(restClient, "my_index");// Create IndexcreateIndex(restClient, "my_index");// Index DocumentindexDocument(restClient, "{\"index\":{\"_index\":\"my_index\",\"_id\":1}}\n{ \"field\": \"value\" }\n");// Get DocumentgetDocument(restClient, "my_index");// Delete DocumentdeleteDocument(restClient, "my_index", "1");// Delete IndexdeleteIndex(restClient, "my_index");} catch (ResponseException e) {e.printStackTrace();// Handle response exceptionSystem.err.println("Error: " + e.getResponse().getStatusLine().getReasonPhrase());} finally {// Close the clientrestClient.close();}}private static void createIndex(RestClient restClient, String index) throws IOException {// Create Index requestRequest request = new Request("PUT", "/" + index);// Execute the requestrestClient.performRequest(request);}private static void indexDocument(RestClient restClient, String s ) throws IOException {// Index Document requestRequest request = new Request("POST", "/_bulk" );request.setJsonEntity(s);// Execute the requestrestClient.performRequest(request);}private static void getDocument(RestClient restClient, String index) throws IOException {// Get Document requestRequest request = new Request("GET", "/" + index +  "/_search");// Execute the requestResponse response = restClient.performRequest(request);// Handle the responseSystem.out.println("Document found: " + EntityUtils.toString(response.getEntity()));}private static void deleteDocument(RestClient restClient, String index,  String id) throws IOException {// Delete Document requestRequest request = new Request("DELETE", "/" + index  + "/_doc/" + id);// Execute the requestrestClient.performRequest(request);}private static void deleteIndex(RestClient restClient, String index) throws IOException {// Delete Index requestRequest request = new Request("DELETE", "/" + index);// Execute the requestrestClient.performRequest(request);}
}

问题

异常如下：

Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested targetat java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439)at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)at java.base/sun.security.validator.Validator.validate(Validator.java:264)at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:285)at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:144)at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1335)... 19 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested targetat java.base/sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:146)at java.base/sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:127)

解决方案

因为证书问题，我们用的是测试环境，就不要费劲的去下载私有证书再安装了，直接配置opensearch支持http即可。

opensearch.yml

plugins.security.ssl.http.enabled: false

或者直接禁用安全插件。

参考

  • https://www.cnblogs.com/openmind-ink/p/13951767.html

这篇关于从零开发短视频电商 Low Level Client(推荐)连接OpenSearch进行CRUD的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/476597。

相关文章

Python在二进制文件中进行数据搜索的实战指南

Python在二进制文件中进行数据搜索的实战指南

《Python在二进制文件中进行数据搜索的实战指南》在二进制文件中搜索特定数据是编程中常见的任务,尤其在日志分析、程序调试和二进制数据处理中尤为重要,下面我们就来看看如何使用Python实现这一功能吧... 目录简介1. 二进制文件搜索概述2. python二进制模式文件读取（rb）2.1 二进制模式与文本
阅读更多...
SQL Server 中的表进行行转列场景示例

SQL Server 中的表进行行转列场景示例

《SQLServer中的表进行行转列场景示例》本文详细介绍了SQLServer行转列（Pivot）的三种常用写法,包括固定列名、条件聚合和动态列名,文章还提供了实际示例、动态列数处理、性能优化建议... 目录一、常见场景示例二、写法 1：PIVOT（固定列名）三、写法 2：条件聚合（CASE WHEN）四、
阅读更多...
VSCode开发中有哪些好用的插件和快捷键

VSCode开发中有哪些好用的插件和快捷键

《VSCode开发中有哪些好用的插件和快捷键》作为全球最受欢迎的编程工具,VSCode的快捷键体系是提升开发效率的核心密码,：本文主要介绍VSCode开发中有哪些好用的插件和快捷键的相关资料,文中... 目录前言1、vscode插件1.1 Live-server1.2 Auto Rename Tag1.3
阅读更多...
Agent开发核心技术解析以及现代Agent架构设计

Agent开发核心技术解析以及现代Agent架构设计

《Agent开发核心技术解析以及现代Agent架构设计》在人工智能领域,Agent并非一个全新的概念,但在大模型时代,它被赋予了全新的生命力,简单来说,Agent是一个能够自主感知环境、理解任务、制定... 目录一、回归本源：到底什么是Agent？二、核心链路拆解：Agent的"大脑"与"四肢"1. 规划模
阅读更多...
Python+FFmpeg实现视频自动化处理的完整指南

Python+FFmpeg实现视频自动化处理的完整指南

《Python+FFmpeg实现视频自动化处理的完整指南》本文总结了一套在Python中使用subprocess.run调用FFmpeg进行视频自动化处理的解决方案,涵盖了跨平台硬件加速、中间素材处理... 目录一、 跨平台硬件加速：统一接口设计1. 核心映射逻辑2. python 实现代码二、 中间素材处
阅读更多...
JavaWeb项目创建、部署、连接数据库保姆级教程(tomcat)

JavaWeb项目创建、部署、连接数据库保姆级教程(tomcat)

《JavaWeb项目创建、部署、连接数据库保姆级教程(tomcat)》：本文主要介绍如何在IntelliJIDEA2020.1中创建和部署一个JavaWeb项目,包括创建项目、配置Tomcat服务... 目录简介：一、创建项目二、tomcat部署1、将tomcat解压在一个自己找得到路径2、在idea中添加
阅读更多...
SpringBoot整合Kafka启动失败的常见错误问题总结(推荐)

SpringBoot整合Kafka启动失败的常见错误问题总结(推荐)

《SpringBoot整合Kafka启动失败的常见错误问题总结(推荐)》本文总结了SpringBoot项目整合Kafka启动失败的常见错误,包括Kafka服务器连接问题、序列化配置错误、依赖配置问题、... 目录一、Kafka服务器连接问题1. Kafka服务器无法连接2. 开发环境与生产环境网络不通二、序
阅读更多...
Python+wxPython开发一个文件属性比对工具

Python+wxPython开发一个文件属性比对工具

《Python+wxPython开发一个文件属性比对工具》在日常的文件管理工作中,我们经常会遇到同一个文件存在多个版本,或者需要验证备份文件与源文件是否一致,下面我们就来看看如何使用wxPython模... 目录引言项目背景与需求应用场景核心需求运行结果技术选型程序设计界面布局核心功能模块关键代码解析文件大
阅读更多...
C++多线程开发环境配置方法

C++多线程开发环境配置方法

《C++多线程开发环境配置方法》文章详细介绍了如何在Windows上安装MinGW-w64和VSCode,并配置环境变量和编译任务,使用VSCode创建一个C++多线程测试项目,并通过配置tasks.... 目录下载安装 MinGW-w64下载安装VS code创建测试项目配置编译任务创建 tasks.js
阅读更多...
JAVA SpringBoot集成Jasypt进行加密、解密的详细过程

JAVA SpringBoot集成Jasypt进行加密、解密的详细过程

《JAVASpringBoot集成Jasypt进行加密、解密的详细过程》文章详细介绍了如何在SpringBoot项目中集成Jasypt进行加密和解密,包括Jasypt简介、如何添加依赖、配置加密密钥... 目录Java (SpringBoot) 集成 Jasypt 进行加密、解密 - 详细教程一、Jasyp
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2