背景：
一般场景 python服务经常作为java下游的 算法服务或者 数据处理服务
但是使用http 去调用比较不灵活，通过注册到nacos上进行微服务调用才是比较爽的
1.定义feginapi的接口定义
java端
定义接口请求和响应 主要关注
CommonResult 结构 和 python要一直 ，不然序列号是有问题的
CommonResult<List<SocialRelationRefDTO>> 和python的 接口的结构要一致
@FeignClient(name = ApiConstants.PYTHON_WEB_SERVER)
@Tag(name = "RPC 服务 - 调用py信息分析接口")
public interface PythonInfoAnalysisApi {
String PREFIX = "analysis";
@PostMapping(PREFIX + "/get_article_similar_relation_ref")
CommonResult<List<SocialRelationRefDTO>> getArticleSimilarRelationRef(@RequestBody List<SocialRelationRefDTO> dto);
}
CommonResult 结构
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Data;
import org.springframework.util.Assert;
import java.io.Serializable;
import java.util.Objects;
/**
* 通用返回
*
* @param <T> 数据泛型
*/
@Data
public class CommonResult<T> implements Serializable {
/**
* 错误码
*
* @see ErrorCode#getCode()
*/
private Integer code;
/**
* 返回数据
*/
private T data;
/**
* 错误提示，用户可阅读
*
* @see ErrorCode#getMsg() ()
*/
private String msg;
/**
* 将传入的 result 对象，转换成另外一个泛型结果的对象
*
* 因为 A 方法返回的 CommonResult 对象，不满足调用其的 B 方法的返回，所以需要进行转换。
*
* @param result 传入的 result 对象
* @param <T> 返回的泛型
* @return 新的 CommonResult 对象
*/
public static <T> CommonResult<T> error(CommonResult<?> result) {
return error(result.getCode(), result.getMsg());
}
public static <T> CommonResult<T> error(Integer code, String message) {
Assert.isTrue(!GlobalErrorCodeConstants.SUCCESS.getCode().equals(code), "code 必须是错误的！");
CommonResult<T> result = new CommonResult<>();
result.code = code;
result.msg = message;
return result;
}
public static <T> CommonResult<T> error(ErrorCode errorCode) {
return error(errorCode.getCode(), errorCode.getMsg());
}
public static <T> CommonResult<T> success(T data) {
CommonResult<T> result = new CommonResult<>();
result.code = GlobalErrorCodeConstants.SUCCESS.getCode();
result.data = data;
result.msg = "success";
return result;
}
public static boolean isSuccess(Integer code) {
return Objects.equals(code, GlobalErrorCodeConstants.SUCCESS.getCode());
}
@JsonIgnore // 避免 jackson 序列化
public boolean isSuccess() {
return isSuccess(code);
}
@JsonIgnore // 避免 jackson 序列化
public boolean isError() {
return !isSuccess();
}
// ========= 和 Exception 异常体系集成 =========
/**
* 判断是否有异常。如果有，则抛出 {@link ServiceException} 异常
*/
public void checkError() throws ServiceException {
if (isSuccess()) {
return;
}
// 业务异常
throw new ServiceException(code, msg);
}
/**
* 判断是否有异常。如果有，则抛出 {@link ServiceException} 异常
* 如果没有，则返回 {@link #data} 数据
*/
@JsonIgnore // 避免 jackson 序列化
public T getCheckedData() {
checkError();
return data;
}
public static <T> CommonResult<T> error(ServiceException serviceException) {
return error(serviceException.getCode(), serviceException.getMessage());
}
}
python端
CommonResult :结构
1 import json 2 from typing import TypeVar, Generic, Optional 3 from dataclasses import dataclass 4 5 from flask import jsonify 6 7 T = TypeVar('T') 8 9 10 @dataclass 11 class CommonResult(Generic[T]): 12 code: Optional[int] = None 13 data: T = None 14 msg: str = None 15 16 @staticmethod 17 def error(code: int, message: str) -> 'CommonResult': 18 assert code is not None, "code must be provided" 19 return CommonResult(code=code, msg=message) 20 21 @staticmethod 22 def success(data: T): 23 result = CommonResult(code=0, data=data, msg="success") 24 return jsonify(result.to_dict()) 25
36 37 def get_checked_data(self) -> T: 38 self.check_error() 39 return self.data 40 41 def to_dict(self): 42 return { 43 "code": self.code, 44 "data": self.data, 45 "msg": self.msg 46 }
接口端：
@analysis_controller_bp.route('/build_media_topological_graph', methods=['POST']) def build_media_topological_graph(): data = request.json result = social_relation_service.build_media_topological_graph(data) common_result = CommonResult.success(result) print(common_result) return common_result # 将CommonResult转换为字典并返回JSON响应
python 启动 类加上 注册nacos和 心跳检查
import os import socket import threading import time import nacos from flask import Flask from route.route import register_route from util.nacos_util import NacosInstance import logging from conf import NACOS_SERVER_ADDRESS, NACOS_SERVER_NAMESPACE app = Flask(__name__) app.config["JSON_AS_ASCII"] = False client = nacos.NacosClient(NACOS_SERVER_ADDRESS, namespace=NACOS_SERVER_NAMESPACE) hostname = socket.gethostname() ip_address = socket.gethostbyname(hostname) instance = NacosInstance( NACOS_SERVER_ADDRESS, namespace=NACOS_SERVER_NAMESPACE, service_name="python-web-server", instance_ip=ip_address, instance_port=5000, cluster='DEFAULT' ) def service_register(): # 创建 Nacos 实 # 注册服务 # instance.register() instance.register() def heartbeat_task(): while True: instance.heartbeat() time.sleep(5) heartbeat_thread = threading.Thread(target=heartbeat_task) heartbeat_thread.start() app = Flask(__name__) register_route(app) if __name__ == '__main__': service_register() app.run(debug=True, host='0.0.0.0', port=5000) # 当应用退出时，注销服务 deregister = instance.deregister() print(deregister)
python注册nacos 核心方法