当前位置：首页 > news >正文

2.微服务灰度发布落地实践(agent实现)

news 2025/6/27 6:49:28

前言

据上一篇，设计方案的分析，综合考虑，最终决定,客户端采用agent方案实现，具本原因不再赘述，
感觉兴趣的小伙伴可以回头了解一下.该篇主要讲java agent的实现,灰度agent客户端的基础框架实现

java agent的介绍

java Agent 是一种允许你在 Java 应用程序启动时或运行时修改其字节码的技术。
它通过 JVM 提供的 java.lang.instrument 包来实现，可以用于各种用途，
如性能监控、代码增强、AOP（面向切面编程）、日志记录等。
Java Agent 的核心功能是能够在类加载到 JVM 之前对其进行修改，
而无需修改应用程序的源代码。启动时代理（Premain Agent）：在 JVM 启动时加载的代理。你需要通过 -javaagent 参数指定代理的 JAR 文件路径，并提供一个 premain 方法作为入口点。

public static void premain(String agentArgs, Instrumentation inst) {// 在这里注册 ClassFileTransformer 或执行其他初始化逻辑
}

基础实现

agent的实现的基础，是基于开源项目skywalking的插件核心实现，没必要重复造轮子,要做的是专注
去实现灰度相关的功能.

agent端 http服务实现

为了方便灰度管理端，给agent端实时推送数据，agent端，基于netty实现单简的http服务，；会提供接口，供服务端对实例状态的控制，以及其它服务实例状态的推送;实例在收发消息或路由时，依赖这些信息作相应处理。
如果不太了解这块功能的作用，可以去第一篇看一下整个系统的架构图;下需要agent http server代码片段

SimpleChannelInboundHandler agentServerHttpInboundHandler = new AgentServerHttpInboundHandler();
agentChannelInitializer = new AgentServerChannelInitializer(agentServerHttpInboundHandler);
Thread serverThread = new Thread(new RunnableWithExceptionProtection(() -> {LOGGER.info("dbcat gray agent netty server.");ServerBootstrap serverBootstrap = new ServerBootstrap();bossGroup = new NioEventLoopGroup(1, new DefaultNamedThreadFactory("server-boss"));workerGroup = new NioEventLoopGroup(4, new DefaultNamedThreadFactory("server-worker"));serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_KEEPALIVE, true).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childHandler(agentChannelInitializer);int port = Config.Agent.SERVER_PORT;try {ChannelFuture f = serverBootstrap.bind(port).sync();LOGGER.info("dbcat gray agent server started, port is {}.", port);f.channel().closeFuture().sync();LOGGER.info("dbcat gray agent server closed, port is {}.", port);} catch (InterruptedException e) {LOGGER.error("dbcat gray agent server start failed", e);Thread.currentThread().interrupt();}
}, t -> {
}), "gray-agent-server");
serverThread.setDaemon(true);
serverThread.start();

agent端api接口

上面实现agent端的http服务，灰度服务端如果需要控制agent客户端，必需还要提供api接口，才能进行交互
http 接口实现片段

   public class AgentServerHttpInboundHandler extends SimpleChannelInboundHandler<FullHttpRequest> {private static final ILog log = LogManager.getLogger(AgentServerHttpInboundHandler.class);private static final Gson GSON = new GsonBuilder().disableHtmlEscaping().create();@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) {ctx.flush();}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {//收到http请求QueryStringDecoder queryStringDecoder = new QueryStringDecoder(request.uri());String requestUri = queryStringDecoder.rawPath();String method = request.method().name();if (!"POST".equals(method.toUpperCase())) {complete(ctx, request,  RestResult.buildFailure(HttpResponseStatus.METHOD_NOT_ALLOWED.toString(),HttpResponseStatus.METHOD_NOT_ALLOWED.code()));return;}//通过url配置接口对应用实现Endpoint endpoint = EndpointManager.getEndpoint(requestUri);if (endpoint == null) {complete(ctx, request, RestResult.buildFailure(HttpResponseStatus.NOT_FOUND.toString(),HttpResponseStatus.NOT_FOUND.code()));return;}Throwable throwable = null;RestResult result = null;try {ByteBuf content = request.content();byte[] requestBodyBytes = new byte[content.readableBytes()];content.readBytes(requestBodyBytes);String requestBodyString = new String(requestBodyBytes, "UTF-8");ParameterizedType parameterizedType = ((ParameterizedType) endpoint.getClass().getGenericInterfaces()[0]);Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();Object data =  GSON.fromJson(requestBodyString,actualTypeArguments[0]);//调用相应的接口result = endpoint.invoke(data);} catch (Throwable e) {log.error(String.format("接口响应异常%s", requestUri), e);throwable = e;} finally {if (throwable == null) {complete(ctx, request, result);} else {complete(ctx, request, RestResult.buildFailure(throwable.getMessage(),HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));}}}}

Endpoint api 接口类，通过spi 方式加载实现

    public class EndpointManager {private static final Map<String, Endpoint> ENDPOINT_INVOKER_MAP = new HashMap<>();static {for (final Endpoint endpointInvoker : ServiceLoader.load(Endpoint.class, AgentClassLoader.getDefault())) {ENDPOINT_INVOKER_MAP.put(endpointInvoker.path(), endpointInvoker);}}public static Endpoint<?, ?> getEndpoint(String requestUri) {return ENDPOINT_INVOKER_MAP.get(requestUri);}}

Endpoint 的实某个实现，如更新实例状态接口

public class EnvStatusUpdateEndpoint implements Endpoint<InstanceEnvUpdateRequest, List<ExecuteResponse>> {private static final ILog log = LogManager.getLogger(EnvStatusUpdateEndpoint.class);@Overridepublic String path() {return "/env-status/update";}@Overridepublic RestResult<List<ExecuteResponse>> invoke(InstanceEnvUpdateRequest request) throws Exception {log.info("更新实例信息");ServerInstance.getInstance().setEnvStatus(request.getEnvStatus());List<ExecuteResponse> executeResponses = MQConnectionManager.restartAll(request.getStrategy());return RestResult.buildSuccess(executeResponses);}
}

到此，agent客户端的基础框加已实现，下一篇将开始介绍相关核心组件灰度的增强实现