自己学习netty的笔记和Demo

package com.zhihao.netty.rpc.clientconsumer.consumer; import cn.hutool.core.lang.UUID; import cn.hutool.core.util.ClassUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONUtil; import com.zhihao.netty.rpc.clientconsumer.consumer.common.NettyConsumerContext; import com.zhihao.netty.rpc.clientconsumer.consumer.handler.MyChannelInitializer; import com.zhihao.netty.rpc.entity.InvokeResult; import com.zhihao.netty.rpc.entity.MessageProtocol; import com.zhihao.netty.rpc.entity.RegistrationInfo; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOption; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import lombok.extern.slf4j.Slf4j; import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; /** * @Author: ZhiHao * @Date: 2022/6/29 17:37 * @Description: * @Versions 1.0 **/ @Slf4j public class NettyClient { private String ip; private Integer port; private String serviceName; private Map<String, ChannelHandlerContext> channelHandlerContextMap; private Map<String, SynchronousQueue<?>> resultConcurrentHashMap; public NettyClient(RegistrationInfo info) { if (StrUtil.isBlank(info.getIp())) { throw new RuntimeException("提供者注册信息有误!!!"); } this.ip = info.getIp(); this.port = info.getPort(); this.serviceName = info.getServiceName(); this.channelHandlerContextMap = NettyConsumerContext.channelHandlerContextMap; this.resultConcurrentHashMap = NettyConsumerContext.resultConcurrentHashMap; } public static NettyClient buildClient(RegistrationInfo info) { NettyClient nettyClient = new NettyClient(info); nettyClient.initNettyClient(); boolean flag = nettyClient.channelHandlerContextMap.containsKey(nettyClient.serviceName); log.info("initNettyClient===serviceName:{}, 初始化客户端结果:{}", nettyClient.serviceName, flag); return flag ? nettyClient : null; } public Object doRemotelyInvoke(Method method, Object[] args) throws TimeoutException { ChannelHandlerContext channelHandlerContext = channelHandlerContextMap.get(serviceName); if (null == channelHandlerContextMap) { throw new RuntimeException("服务提供者已不存在!!!"); } MessageProtocol protocol = new MessageProtocol(); String requestId = UUID.fastUUID().toString() + this.getCurrentTimeSecond(); String simpleName = method.getDeclaringClass().getSimpleName(); protocol.setClassName(StrUtil.lowerFirst(simpleName)); protocol.setMethodName(method.getName()); protocol.setMethodParameterTypes(method.getParameterTypes()); protocol.setMethodParameter(args); protocol.setRequestId(requestId); // 使用同步阻塞队列, 阻塞等待, 别的线程插入结果, 有结果线程在继续 SynchronousQueue<?> synchronousQueue = new SynchronousQueue<>(); // 以UUID+时间戳作为唯一key resultConcurrentHashMap.put(requestId, synchronousQueue); String jsonStr = JSONUtil.toJsonStr(protocol); channelHandlerContext.writeAndFlush(Unpooled.wrappedBuffer(jsonStr.getBytes(StandardCharsets.UTF_8))); log.info("doRemotelyInvoke===已经发送, jsonStr:{}!!!!", jsonStr); Class<?> returnType = method.getReturnType(); InvokeResult take = null; long second; try { second = this.getCurrentTimeSecond(); take = (InvokeResult) synchronousQueue.poll(30L, TimeUnit.SECONDS); second = this.getCurrentTimeSecond() - second; } catch (Exception e) { throw new RuntimeException(e); } if (second >= 30L) { resultConcurrentHashMap.remove(requestId); throw new TimeoutException("调用超时!!!"); } Integer resultCode = Optional.ofNullable(take).map(InvokeResult::getResultCode).orElse(-1); Object result = null; if (resultCode.equals(1)) { throw new RuntimeException(take.getFailMessage()); } else { String resultStr = Optional.ofNullable(take).map(InvokeResult::getInvokeResult) .map(Objects::toString).orElse(""); if (JSONUtil.isJsonArray(resultStr)) { result = JSONUtil.toList(resultStr, returnType); } else if (JSONUtil.isJsonObj(resultStr)) { result = JSONUtil.toBean(resultStr, returnType); } else if (ClassUtil.isBasicType(returnType)) { result = resultStr; } else { result = resultStr; } } return result; } private void initNettyClient() { // 以常量池中的服务名称作为锁对象 synchronized (serviceName.intern()) { try { // 不包含则初始化 if (!channelHandlerContextMap.containsKey(serviceName)) { new Thread(() -> { NioEventLoopGroup workerGroup = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap(); bootstrap = bootstrap.group(workerGroup) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE) .option(ChannelOption.TCP_NODELAY, Boolean.TRUE) .option(ChannelOption.SO_TIMEOUT, 3) .handler(new MyChannelInitializer(serviceName)); ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(ip, port)); channelFuture.addListener(el -> { if (el.isSuccess()) { log.info("initNettyClient===serviceName:{}, 连接成功!!!", serviceName); } else { log.error("initNettyClient===serviceName:{}, 连接失败!!!", serviceName); } }); try { channelFuture.sync().channel().closeFuture().sync(); } catch (Throwable e) { log.error("initNettyClient===serviceName:{}, 发生异常, 信息:{}", serviceName, e); } finally { workerGroup.shutdownGracefully(); } }).start(); } log.info("initNettyClient===serviceName:{}, 开始等待初始化客户端完成", serviceName); // 以常量池中不可变引用服务字符串 serviceName.intern().wait(5000L); } catch (Exception e) { log.error("initNettyClient===serviceName:{}, 初始化客户端发生异常, 信息:{}", serviceName, e); } } } private long getCurrentTimeSecond() { return TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()); } }