Python 与 Java 集成实战——Jython、RPC、微服务跨语言通信方案

适读人群：Java + Python 混合技术栈的工程师、做 AI 工程的 Java 背景开发者 | 阅读时长：约17分钟 | 核心价值：掌握 Python 与 Java 系统集成的三种工程方案及选型原则

---

我的 Java + Python 双栈经历

我本来是 Java 工程师，转 AI 之后做的是 Python。但现实是，大多数公司的核心业务系统是 Java 写的，AI 能力是 Python 写的，两个语言必须协同工作。

有一段时间，我同时维护着一个 Java Spring Boot 的核心业务服务，和一个 Python FastAPI 的 AI 推理服务。两边都需要数据交换：业务侧把待处理的文本推给 AI 侧，AI 侧把推理结果回写到业务数据库。

最初的方案极其简陋——AI 服务直接连业务数据库，从里面捞数据、写数据。这个方案稳定运行了两个月，然后 DBA 找来了，说某天深夜数据库慢查询全是 AI 侧的脚本在做全表扫，把数据库拖慢了。

从那次开始，我们重新设计了跨语言通信方案，系统性地解决了 Java 和 Python 之间的集成问题。今天来讲这套经验。

---

一、三种集成方案概览

Python 与 Java 的集成，主要有三种路线：

方案	适用场景	复杂度	性能
Jython	在 Java 内嵌入 Python 脚本	低	中（有解释器开销）
HTTP/REST	独立服务，HTTP 通信	中	中（序列化开销）
gRPC	高性能跨语言 RPC	高	高（二进制序列化）
消息队列	异步解耦，大量数据交换	中	高（异步）

---

二、Jython——在 JVM 内嵌 Python

Jython 是运行在 JVM 上的 Python 实现，允许 Java 程序直接调用 Python 代码（或反之）。

适用场景：需要在 Java 程序里运行 Python 脚本，且脚本逻辑不复杂、不需要 C 扩展的 Python 库。

最大限制：Jython 目前停留在 Python 2.7 / Python 3 早期，无法使用 NumPy、PyTorch 等 C 扩展库——这在 AI 工程场景里几乎是致命的。

<!-- Java pom.xml -->
<dependency>
    <groupId>org.python</groupId>
    <artifactId>jython-standalone</artifactId>
    <version>2.7.3</version>
</dependency>

// Java 调用 Python 脚本（Jython）
import org.python.util.PythonInterpreter;
import org.python.core.PyObject;

public class JythonExample {
    public static void main(String[] args) {
        PythonInterpreter interpreter = new PythonInterpreter();
        
        // 直接执行 Python 代码
        interpreter.exec("result = 2 + 2");
        PyObject result = interpreter.get("result");
        System.out.println("Result: " + result);  // 4
        
        // 执行 Python 文件
        interpreter.execfile("script.py");
        
        // 调用 Python 函数
        interpreter.exec(
            "def process_text(text):\n" +
            "    return text.upper()\n"
        );
        PyObject func = interpreter.get("process_text");
        PyObject processed = func.__call__(new PyObject[]{
            interpreter.eval("'hello world'")
        });
        System.out.println("Processed: " + processed);
    }
}

我的建议：在 AI 工程场景里，基本不推荐用 Jython。 它的 Python 版本太老，无法使用现代 AI 库。只有在极特殊的场景（比如已有的 Python 2 脚本集成）才考虑。

---

三、HTTP REST API——最常用的方案

对于绝大多数 Java + Python 集成场景，独立部署两个服务，通过 HTTP 通信是最简单、最成熟、最推荐的方案。

Python 侧：FastAPI 推理服务

# src/main.py
from contextlib import asynccontextmanager
from typing import Optional
import logging
import time

from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# 模型管理
class ModelManager:
    def __init__(self):
        self._model = None
        self._loaded = False

    def load(self):
        """延迟加载模型"""
        if not self._loaded:
            logger.info("加载 AI 模型...")
            # import torch
            # self._model = torch.load("model.pt")
            self._loaded = True
            logger.info("模型加载完成")

    def predict(self, text: str) -> dict:
        if not self._loaded:
            raise RuntimeError("模型未加载")
        # 推理逻辑
        return {"label": "positive", "confidence": 0.95}


model_manager = ModelManager()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载模型
    model_manager.load()
    yield
    # 关闭时清理资源


app = FastAPI(title="AI 推理服务", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["POST", "GET"],
)


class PredictRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=10000)
    request_id: Optional[str] = None


class PredictResponse(BaseModel):
    request_id: Optional[str]
    label: str
    confidence: float
    latency_ms: float


@app.post("/v1/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
    start = time.time()
    try:
        result = model_manager.predict(request.text)
        latency = (time.time() - start) * 1000
        return PredictResponse(
            request_id=request.request_id,
            label=result["label"],
            confidence=result["confidence"],
            latency_ms=round(latency, 2),
        )
    except Exception as e:
        logger.error(f"推理失败: {e}")
        raise HTTPException(status_code=500, detail="推理服务内部错误")


@app.get("/health")
async def health():
    return {"status": "ok", "model_loaded": model_manager._loaded}

Java 侧：调用 Python 服务

// Java Spring Boot 调用 Python AI 服务
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.time.Duration;

@Service
public class AIInferenceClient {
    
    private final WebClient webClient;
    
    public AIInferenceClient(WebClient.Builder builder) {
        this.webClient = builder
            .baseUrl("http://ai-service:8000")
            .defaultHeader("Content-Type", "application/json")
            .build();
    }
    
    public record PredictRequest(String text, String requestId) {}
    public record PredictResponse(String requestId, String label, double confidence, double latencyMs) {}
    
    public Mono<PredictResponse> predict(String text, String requestId) {
        var request = new PredictRequest(text, requestId);
        
        return webClient.post()
            .uri("/v1/predict")
            .bodyValue(request)
            .retrieve()
            .onStatus(status -> status.is5xxServerError(), response -> 
                Mono.error(new RuntimeException("AI 服务内部错误"))
            )
            .bodyToMono(PredictResponse.class)
            .timeout(Duration.ofSeconds(30))
            .retry(2);  // 失败自动重试2次
    }
    
    // 批量推理（并发）
    public java.util.List<PredictResponse> batchPredict(java.util.List<String> texts) {
        var requests = texts.stream()
            .map(t -> predict(t, java.util.UUID.randomUUID().toString()))
            .collect(java.util.stream.Collectors.toList());
        
        return reactor.core.publisher.Flux.fromIterable(requests)
            .flatMap(mono -> mono, 10)  // 最多10个并发
            .collectList()
            .block(Duration.ofMinutes(2));
    }
}

---

四、gRPC——高性能跨语言 RPC

当 HTTP REST 的性能不满足需求时（比如每秒需要处理数千次推理），gRPC 是更好的选择：

pip install grpcio grpcio-tools

// proto/inference.proto
syntax = "proto3";
package inference;

service InferenceService {
  rpc Predict(PredictRequest) returns (PredictResponse);
  rpc BatchPredict(stream PredictRequest) returns (stream PredictResponse);
}

message PredictRequest {
  string text = 1;
  string request_id = 2;
}

message PredictResponse {
  string request_id = 1;
  string label = 2;
  float confidence = 3;
  float latency_ms = 4;
}

# Python gRPC Server
import grpc
import asyncio
import time
from concurrent import futures
import inference_pb2
import inference_pb2_grpc


class InferenceServicer(inference_pb2_grpc.InferenceServiceServicer):

    async def Predict(self, request, context):
        start = time.time()
        # 模型推理
        label, confidence = "positive", 0.95
        latency = (time.time() - start) * 1000

        return inference_pb2.PredictResponse(
            request_id=request.request_id,
            label=label,
            confidence=confidence,
            latency_ms=latency,
        )

    async def BatchPredict(self, request_iterator, context):
        async for request in request_iterator:
            response = await self.Predict(request, context)
            yield response


async def serve():
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))
    inference_pb2_grpc.add_InferenceServiceServicer_to_server(
        InferenceServicer(), server
    )
    server.add_insecure_port("[::]:50051")
    await server.start()
    print("gRPC 服务启动: port 50051")
    await server.wait_for_termination()

asyncio.run(serve())

---

五、消息队列——异步解耦方案

对于不需要实时响应的场景（比如文档分析、批量报告生成），消息队列是最优雅的跨语言集成方式：

# Python 消费者（处理 Java 推送的任务）
import json
import logging
import redis

logger = logging.getLogger(__name__)


class AITaskConsumer:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.input_queue = "ai:tasks:input"
        self.output_queue = "ai:results:output"

    def process_task(self, task: dict) -> dict:
        """执行 AI 推理任务"""
        text = task["text"]
        task_id = task["task_id"]
        # 实际推理逻辑
        return {
            "task_id": task_id,
            "label": "positive",
            "confidence": 0.95,
        }

    def run(self):
        logger.info("AI 任务消费者启动...")
        while True:
            # 阻塞等待任务（超时5秒）
            item = self.redis.blpop(self.input_queue, timeout=5)
            if item is None:
                continue

            _, raw = item
            task = json.loads(raw)
            logger.info(f"处理任务: {task['task_id']}")

            try:
                result = self.process_task(task)
                result["status"] = "success"
            except Exception as e:
                result = {"task_id": task["task_id"], "status": "failed", "error": str(e)}

            # 结果推送到输出队列（Java 侧消费）
            self.redis.rpush(self.output_queue, json.dumps(result))
            logger.info(f"任务完成: {task['task_id']}")


consumer = AITaskConsumer()
# consumer.run()

---

六、踩坑实录

踩坑实录1：HTTP 超时设置不合理，Java 请求大量积压

现象：AI 推理服务偶尔变慢时，Java 侧有大量请求在等待，最终超时、线程池耗尽。
原因：Java 的 HTTP 客户端没有设置合理超时，一个请求慢，把整个线程池都占满了。
解法：设置严格的超时（connect timeout + read timeout），并使用熔断器（Resilience4j）。

踩坑实录2：序列化格式不一致，Java 解析 Python 返回的数据报错

现象：Python 返回了 {"value": 1.0000000000000002}，Java 反序列化 float 精度不一致。
原因：Python float 和 Java double 精度表示方式有细微差异。
解法：接口返回金额等精度敏感数据时，统一用字符串格式（"1.00"），不用浮点数。

踩坑实录3：Python 推理服务无状态，但 Java 误以为有状态

现象：Java 侧把某个请求发给了某个 Python 实例，然后期望下次请求还发到同一实例（因为有上下文）。
原因：负载均衡把请求路由到了不同的 Python 实例，上下文丢失。
解法：AI 推理服务应该设计为完全无状态，所有上下文都在每次请求里携带，不依赖服务端 session。

---

七、选型建议

集成场景	推荐方案
简单脚本嵌入（不需要 AI 库）	Jython（不推荐 AI 场景）
独立服务，实时调用	HTTP REST（FastAPI）
高性能实时推理（>500 QPS）	gRPC
批量异步处理	Redis/RabbitMQ 消息队列
共享数据，间接集成	共享数据库（谨慎使用）

作为一个 Java + Python 双栈工程师，我感受最深的是：不要让两种语言的代码紧密耦合，保持清晰的服务边界，HTTP 或 gRPC 都是比 Jython 嵌入更健康的集成方式。语言各司其职，Java 做业务、Python 做 AI，接口清晰，独立部署，这是长期可维护的基础。