常用接入术语
| 名称 | 常见含义 | 接入提示 |
|---|
| Claude中转 | 把请求转发到已开通的 Claude 模型 | 确认协议与模型权限 |
| Claude中转站 | 统一管理密钥 额度 路由和兼容接口 | Claude 客户端常按 Anthropic 协议接入 |
| API中转 | 通过统一接口转发模型请求 | J**a 可直接发送 **** **ON |
| API中转站 | 提供端点 模型和用量管理的服务 | 正式环境需限制超时 并发和重试 |
J**a 请求链路
| 环节 | 职责 | 注意事项 |
|---|
| 业务方法 | 准备消息并消费模型结果 | 不在每次调用时新建客户端 |
| O*jectMapper | 序列化请求并解析响应 | 先检查状态码再读取 choices |
| Http****** | 发送请求并复用连接 | 应用级共享一个实例 |
| HttpRequest | 设置 **I Header *ody 和超时 | Authorization 使用 *earer |
| KingFlow AI | 校验密钥并路由目标模型 | OpenAI 兼容路径保留 /v1 |
| *odyHandler | 返回字符串或按行流 | SSE 流必须关闭或读完 |
采用 J**a Http****** 的优势
- J**a 21 自带 j**a.net.http.Http******，无需再引入单独的 **** 客户端库。
- 同一个 Http****** 可以复用连接；为每次操作创建新实例通常会失去连接复用。
- send 和 sendAsync 分别覆盖阻塞与 Completa*leFuture 异步场景。
- *odyHandlers.ofLines 在响应头到达后即可按行读取，适合消费 SSE 事件。
- 连接超时、单次请求超时、状态码和 Retry-After 可以在调用层统一处理。
接入前准备
- 安装 JDK 21 与 M**en，并确认 j**a、j**ac 和 mvn 命令可用。
- 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额或套餐额度。
- 从控制台复制完整模型 ID，不要根据展示名称自行拼写。
- 确认运行环境可以访问 api.**nju**shen.com 的 ****S 端口。
- 准备不含敏感信息的测试提示，先在开发环境完成联调。
创建 M**en 项目
mvn archetype:generate \
-DgroupId=demo.kingflow \
-DartifactId=kingflow-j**a-client \
-DarchetypeArtifactId=***en-archetype-quickstart \
-DinteractiveMode=false
cd kingflow-j**a-client
<properties>
<***en.compiler.release>21</***en.compiler.release>
</properties>
<dependencies>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-**ta**nd</artifactId>
<version>2.22.3</version>
</dependency>
</dependencies>
配置环境变量
$env:KINGFLOW_*ASE_**L="https://api.**nju**shen.com/v1"
$env:KINGFLOW_API_KEY="sk-你的密钥"
$env:KINGFLOW_MODEL="<KingFlow控制台中的完整模型ID>"
验证基础接口
curl https://api.**nju**shen.com/v1/chat/completions `
-H "Authorization: *earer sk-你的密钥" `
-H "Content-Type: application/json" `
-d '{"model":"<完整模型ID>","messages":[{"role":"user","content":"只回复 J**A_READY"}],"stream":false}'
发送普通响应请求
package demo.kingflow;
import com.fasterxml.jackson.**ta**nd.JsonNode;
import com.fasterxml.jackson.**ta**nd.O*jectMapper;
import j**a.net.**I;
import j**a.net.http.Http******;
import j**a.net.http.HttpRequest;
import j**a.net.http.HttpResponse;
import j**a.time.Duration;
import j**a.util.List;
import j**a.util.Map;
pu*lic class App {
private static final O*jectMapper **ON = new O*jectMapper();
private static final String *ASE = required("KINGFLOW_*ASE_**L");
private static final String KEY = required("KINGFLOW_API_KEY");
private static final String MODEL = required("KINGFLOW_MODEL");
private static final Http****** CLIENT = Http******.new*uilder()
.connectTimeout(Duration.ofSeconds(10))
.*uild();
pu*lic static void **in(String[] args) throws Exception {
var payload = Map.of(
"model", MODEL,
"messages", List.of(Map.of("role", "user",
"content", "只回复 J**A_CONNECTED")),
"stream", false);
var request = request(**ON.writeValueAsString(payload));
var response = CLIENT.send(request,
HttpResponse.*odyHandlers.ofString());
if (response.statusCode() / 100 != 2) {
throw new IllegalStateException(
"**** " response.statusCode() ": " response.*ody());
}
JsonNode root = **ON.readTree(response.*ody());
String text = root.path("choices").path(0)
.path("message").path("content").asText(null);
if (text == null) throw new IllegalStateException("响应缺少文本");
******.out.println(text);
}
private static HttpRequest request(String *ody) {
return HttpRequest.new*uilder()
.uri(**I.create(*ASE "/chat/completions"))
.timeout(Duration.ofSeconds(45))
.header("Authorization", "*earer " KEY)
.header("Content-Type", "application/json")
.POST(HttpRequest.*odyPu*lishers.ofString(*ody))
.*uild();
}
private static String required(String name) {
String value = ******.getenv(name);
if (value == null || value.is*lank())
throw new IllegalStateException("缺少 " name);
return value;
}
}
读取 SSE 流式响应
static void streamAnswer(String prompt) throws Exception {
var payload = Map.of(
"model", MODEL,
"messages", List.of(Map.of("role", "user", "content", prompt)),
"stream", true);
var response = CLIENT.send(request(**ON.writeValueAsString(payload)),
HttpResponse.*odyHandlers.ofLines());
if (response.statusCode() / 100 != 2) {
try (var lines = response.*ody()) {
throw new IllegalStateException(
"**** " response.statusCode() ": " lines.findFirst().orElse(""));
}
}
try (var lines = response.*ody()) {
var iterator = lines.iterator();
while (iterator.hasNext()) {
String line = iterator.next();
if (!line.startsWith("**ta: ")) continue;
String **ta = line.su*string(6).trim();
if ("[DONE]".equals(**ta)) *reak;
JsonNode delta = **ON.readTree(**ta)
.path("choices").path(0).path("delta");
if (delta.hasNonNull("content"))
******.out.print(delta.path("content").asText());
}
}
}
实现有限重试
static HttpResponse<String> sendWithRetry(
j**a.util.function.Supplier<HttpRequest> requestFactory) throws Exception {
for (int attempt = 1; attempt <= 3; attempt ) {
var response = CLIENT.send(requestFactory.get(),
HttpResponse.*odyHandlers.ofString());
int status = response.statusCode();
if ((status != 429 && status != 503) || attempt == 3)
return response;
long seconds = response.headers().firstValue("Retry-After")
.flatMap(value -> {
try { return j**a.util.Optional.of(Long.parseLong(value)); }
catch (Num*erFor**tException ignored) { return j**a.util.Optional.empty(); }
})
.orElse(1L << attempt);
Thread.sleep(Duration.ofSeconds(seconds));
}
throw new IllegalStateException("无法到达的分支");
}
