# http-client

**Repository Path**: dufafei/http-client

## Basic Information

- **Project Name**: http-client
- **Description**: 数据集成子项目 - HTTP客户端
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-01-13
- **Last Updated**: 2026-07-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# HTTP客户端

## 概述

本项目为数据集成子项目，专为 HTTP 处理/输出插件 设计。

提供高性能、高扩展性的 HTTP 请求能力。

## 功能

### 1. 支持认证方式扩展

提供灵活的认证接口，方便扩展自定义的鉴权逻辑：

```java
public interface HttpAuth {

    String getId();

    HttpClient getClient();

    void setClient(HttpClient client);

    /** 初始化认证信息 */
    void init();

    /** 请求前添加认证信息 */
    HttpRequestInfo preAuth(HttpRequestInfo request) throws Exception;

    /** 请求后添加认证信息 比如 digest 认证 */
    HttpResponseInfo postAuth(HttpRequestInfo request, HttpResponseInfo response, Schema schema, Row row, HttpCallback callback)     throws Exception;
}
```

**已内置实现**：Basic、Bearer、Digest、API Key、OAuth2。

### 2. 支持认证信息缓存

支持对认证凭证进行本地缓存，可以有效避免高频请求下的重复鉴权操作。

显著降低网络开销与认证服务压力的同时，大幅提升高并发场景下的数据同步性能。

支持引入 Redis 作为全局共享缓存，实现跨节点的认证状态统一管理与复用，避免多节点重复发起鉴权请求。

```java
public interface HttpAuthCache {

    /**
     * 获取单个key是否存在
     */
    boolean containsKey(String key);

    /**
     * 获取单个认证属性
     */
    String get(String key);

    /**
     * 批量获取认证信息
     */
    Map<String, String> get();

    /**
     * 设置认证信息
     */
    void set(Map<String, String> map);

    /**
     * 原子化递增
     */
    Integer incrementAndGet(String key);

    /**
     * 删除单个认证属性
     */
    void remove(String key);

    /**
     * 清除认证信息
     */
    void clear();
}
```

### 3. 支持请求体解析扩展

提供请求体解析接口，支持自定义序列化方式：

```java
public interface HttpBodyParser {

    String getName();

    HttpEntity getRequestEntity(Charset contentCharset, String body, List<HttpRequestFile> files) throws IOException;
}
```

### 4. 支持模版变量解析

ETL 流程支持在 HTTP 模板的 Path、Param、Header 及 Body 中配置变量。

通过调用 `HttpClient.prepare` 进行预处理，可将变量名转化为底层索引。

在实际请求阶段，系统基于索引提取数据并完成动态替换，实现模板与数据的解耦。

**需要特别强调的是，由于数组基于连续内存分配且通过下标直接寻址，其取值效率远高于 Map 结构。**

```java
public static String replace(String text, Schema schema, Row row) {
        if (text != null) {
            PlaceholderParser pl = new PlaceholderParser("${", "}",
                    placeholder -> {
                        int index = Integer.parseInt(placeholder);
                        StringValue value = row.getString(schema, index);
                        return value == null ? null : value.getString();
                    });
            return pl.parse(text);
        } else {
            return null;
        }
}
```

### 5. 支持脚本处理

支持在请求前后注入自定义脚本，实现动态逻辑处理：

- **前置脚本 (Pre-request Script)**：在认证之前执行，允许动态修改请求内容。
- **后置脚本 (Post-request Script)**：认证完成后执行，允许动态修改返回内容。

**执行顺序**

1. 变量解析与替换
2. 前置脚本 (Pre-request Script)
3. 前置认证 (`preAuth`)
4. 发送网络请求 (`execute`)
5. 后置认证处理 (`postAuth`，包含可能的重试闭环)
6. 后置脚本 (Post-request Script)

**前置脚本示例 (JavaScript)**

```javascript
// 1. 从上下文中获取当前行数据（row）和请求对象（request）
var userId = row.getStringValue(schema, 0); 
var timestamp = new Date().getTime();

// 2. 模拟一个简单的签名逻辑（实际业务中可能是 MD5/HMAC 等）
var rawStr = userId + "_" + timestamp;
var signature = "SIGN_" + rawStr;

// 3. 将动态生成的参数添加到请求头中
request.addHeader("X-Timestamp", "" + timestamp);
request.addHeader("X-Signature", signature);

// 4. 打印日志（输出到 LoggerWriter 中，方便调试）
print("前置脚本执行成功，已添加签名头: " + signature);
```

**测试代码示例**

```java
Schema schema = new Schema();
StringField field = new StringField();
field.setName("user_id");
schema.addField(field);
Row row = new Row(1);
row.setValue(0, StringValue.fromString("dufafei"));

DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
HttpAuthCache authCache = new HttpAuthMapCache();

HttpRequestInfo re = new HttpRequestInfo();
HttpDigestAuth digestAuth = new HttpDigestAuth();
digestAuth.setUsername("sa");
digestAuth.setPassword("123456");
re.setAuth(digestAuth);
re.setAuthCache(authCache);
re.setMethod(HttpMethod.POST.getName());
re.setPath("http://localhost:8080/token/SaCheckHttpDigest");
// 设置前置脚本
re.setPreScript(preScript);

HttpResponseInfo res = client.send(re, schema, row);
System.out.println(res.getBody());
```

### 6. 支持脚本中使用函数

#### 6.1.函数执行引擎

通过**动态代码生成与编译执行**的方式保障大数据量下的极致性能。

```java
Schema schema = new Schema();
schema.addField(new Field("name", new StringType()));

Row row = new Row(1);
row.setValue(0, StringValue.fromString("dff"));

ELEngine elEngine = new ELEngine();
String exp = "concat('test', start_with(name, 'aa'))";
// 编译代码并加载
EL elInterface = elEngine.compile(schema, exp);
System.out.println(elInterface.eval(row));
```

表达式编译后的代码如下：

```java
package com.gitee.dufafei.elInterface.compile;

import com.gitee.dufafei.el.ELInterfaceInterface;
import com.gitee.dufafei.elInterface.context.ELGlobalContext;
import com.gitee.dufafei.elInterface.func.Function;
import com.gitee.dufafei.schema.row.Row;
import com.gitee.dufafei.schema.values.StringValue;
import com.gitee.dufafei.schema.values.ValueInterface;
import java.lang.Override;

public class Fel_1 implements ELInterface {
  private final Function f_19 = ELGlobalContext.getFunction(18);

  private final ValueInterface c_1 = StringValue.fromString("test");

  private final Function f_9 = ELGlobalContext.getFunction(8);

  private final ValueInterface c_2 = StringValue.fromString("aa");

  @Override
  public ValueInterface eval(Row row) {
    ValueInterface v_1 = row.getValue(0);
    ValueInterface tmp_1 = f_9.eval(v_1, c_2);
    ValueInterface tmp_2 = f_19.eval(c_1, tmp_1);
    return tmp_2;
  }
}
```

#### 6.2 脚本中使用函数

在脚本中使用函数，可以复用函数的能力，大大增强脚本的可用性。

```java
// 1. 准备数据与引擎
ELEngine engine = new ELEngine();

Schema schema = new Schema();
StringField field = new StringField();
field.setName("name");
schema.addField(field);

Row row = new Row(1);
row.setValue(0, StringValue.fromString("dff"));

// 2. 绑定数据流
Map<String, Object> params = new HashMap<>();
params.put("el", engine);
params.put("schema", schema);
params.put("row", row);

// 3. 编译并执行脚本
ScriptProcessor processor = ScriptProcessor.getInstance();
CompiledScript compiledScript = processor.compile(
        "var node = el.parse(schema, \"concat(name,'x')\"); " +
        "var compiled = el.compile(node); " +
        "print(compiled.eval(row));"
);
processor.eval(compiledScript, params);
```

制台输出日志：

<span style="color: red;"> INFO com.gitee.dufafei.plugin.core.PluginRegistry - Registering plugin type: function<br> INFO com.gitee.dufafei.plugin.core.PluginRegistry - Scanning local plugins in package: com.gitee.dufafei.el.funcs<br> INFO com.gitee.dufafei.plugin.core.PluginRegistry - Found 1 resource URLs for package: com.gitee.dufafei.el.funcs<br> INFO com.gitee.dufafei.plugin.core.PluginRegistry - Registering Plugin{type='function', id='concat', version='1.0'}<br> INFO com.gitee.dufafei.el.context.ELGlobalContext - Function 'concat' registered.<br>   INFO com.gitee.dufafei.script.LoggerWriter - dffx </span>

### 7. 支持连接池监控

提供连接池指标监控接口，方便排查性能瓶颈：

```java
public interface HttpPoolClientMXBean {
    /** 已租用连接数：当前正在被 HTTP 请求占用的连接数量。若长期接近 Max，说明连接池压力大。 */
    int getLeasedConnections();

    /** 待定/等待连接数：正在等待获取连接的线程数。若 > 0，说明连接池已成为瓶颈。 */
    int getPendingConnections();

    /** 可用连接数：当前处于空闲状态、可立即复用的连接数量。 */
    int getAvailableConnections();

    /** 最大连接数：该统计范围允许的最大连接总数。 */
    int getMaxConnections();
}
```

**监控接入方式:**

通过集成 **Jolokia** 将 JMX 协议桥接为 HTTP 协议，利用 REST API 可便捷地获取连接池实时状态。

**启动参数添加：**

```
-javaagent:D:\Work\project\http-client\lib\jolokia-jvm-1.7.2.jar=port=8778,host=localhost
```

**监控 API 示例：**

http://ip:8778/jolokia/read/org.apache.http.client:type=HttpClientPool,name=DefaultHttpPoolClient/MaxConnections

**返回：**

```json
{
    "request": {
        "mbean": "org.apache.http.client:name=DefaultHttpPoolClient,type=HttpClientPool",
        "attribute": "MaxConnections",
        "type": "read"
    },
    "value": 50,
    "timestamp": 1782772115,
    "status": 200
}
```

## 使用

使用示例：

```java
// 1. 测试 Multipart 文件上传
@Test
public void test1() throws Exception {
    DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
    HttpRequestInfo request = new HttpRequestInfo();
    request.setAuth(new HttpNoAuth());
    request.setMethod(HttpMethod.POST.getName());
    request.setPath("http://localhost:8080/http/test1");
    request.setContentType(HttpContentType.MULTIPART_FORM_DATA.getName());
    request.setBody("""
            {"id":1,"name":"读法非"}
            """);

    HttpRequestFile file = new HttpRequestFile("file", "test.text",
            new FileInputStream(new File(System.getProperty("user.dir") + "/src/test/java/test.text")));
    request.addFile(file);

    HttpResponseInfo res = client.send(request, null, null);
    System.out.println(res.getBody());
}

// 2. 测试 Form 表单提交
@Test
public void test2() throws Exception {
    DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
    HttpRequestInfo request = new HttpRequestInfo();
    request.setMethod(HttpMethod.POST.getName());
    request.setPath("http://localhost:8080/http/test2");
    request.setContentType(HttpContentType.APPLICATION_X_WWW_FORM_URLENCODED.getName());
    request.setBody("""
            {"id":1,"name":"读法非"}
            """);

    HttpResponseInfo res = client.send(request, null, null);
    System.out.println(res.getBody());
}

// 3. 测试 JSON 提交
@Test
public void test3() throws Exception {
    DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
    HttpRequestInfo request = new HttpRequestInfo();
    request.setMethod(HttpMethod.POST.getName());
    request.setPath("http://localhost:8080/http/test3");
    request.setContentType(HttpContentType.APPLICATION_JSON.getName());
    request.setBody("""
            {"id":1,"name":"读法非a"}
            """);

    HttpResponseInfo res = client.send(request, null, null);
    System.out.println(res.getBody());
}

// 4. 测试二进制流上传
@Test
public void test4() throws Exception {
    DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
    HttpRequestInfo request = new HttpRequestInfo();
    request.setMethod(HttpMethod.POST.getName());
    request.setPath("http://localhost:8080/http/test4");
    request.setContentType(HttpContentType.APPLICATION_OCTET_STREAM.getName());

    HttpRequestFile file = new HttpRequestFile("file", "test.text",
            new FileInputStream(new File(System.getProperty("user.dir") + "/src/test/java/test.text")));
    request.addFile(file);

    HttpResponseInfo res = client.send(request, null, null);
    System.out.println(res.getBody());
}

// 5. 测试文件下载
@Test
public void test5() throws Exception {
    DefaultHttpClient client = new DefaultHttpClient(new HttpSettings());
    HttpRequestInfo request = new HttpRequestInfo();
    request.setMethod(HttpMethod.GET.getName());
    request.setPath("http://localhost:8080/http/test5");
    request.setContentType(HttpContentType.APPLICATION_OCTET_STREAM.getName());
    request.addParam("fileName", System.getProperty("user.dir") + "/src/test/java/test.text");

    FileDownloadCallback callback = new FileDownloadCallback(
            System.getProperty("user.dir") + "/src/test/java/", "test1.text");
    HttpResponseInfo res = client.send(request, null, null, callback);
    
    System.out.println(res.getBody());
}
```

## 声明

本项目代码受版权保护。除明确授权外，**保留所有权利（All Rights Reserved）**。

- **授权范围**：本项目仅供内部学习、测试及非商业性质的个人研究使用。
- **禁止事项**：未经授权，严禁任何形式的二次开源、分发或作为衍生作品发布于公共代码托管平台。