dubbo中NettyServer的作用是什么,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractServer.java
public abstract class AbstractServer extends AbstractEndpoint implements Server { protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler"; private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class); ExecutorService executor; private InetSocketAddress localAddress; private InetSocketAddress bindAddress; private int accepts; private int idleTimeout; public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); localAddress = getUrl().toInetSocketAddress(); String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { bindIp = ANYHOST_VALUE; } bindAddress = new InetSocketAddress(bindIp, bindPort); this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS); this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT); try { doOpen(); if (logger.isInfoEnabled()) { logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress()); } } catch (Throwable t) { throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t); } //fixme replace this with better method DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort())); } protected abstract void doOpen() throws Throwable; protected abstract void doClose() throws Throwable; @Override public void reset(URL url) { if (url == null) { return; } try { if (url.hasParameter(ACCEPTS_KEY)) { int a = url.getParameter(ACCEPTS_KEY, 0); if (a > 0) { this.accepts = a; } } } catch (Throwable t) { logger.error(t.getMessage(), t); } try { if (url.hasParameter(IDLE_TIMEOUT_KEY)) { int t = url.getParameter(IDLE_TIMEOUT_KEY, 0); if (t > 0) { this.idleTimeout = t; } } } catch (Throwable t) { logger.error(t.getMessage(), t); } try { if (url.hasParameter(THREADS_KEY) && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) { ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor; int threads = url.getParameter(THREADS_KEY, 0); int max = threadPoolExecutor.getMaximumPoolSize(); int core = threadPoolExecutor.getCorePoolSize(); if (threads > 0 && (threads != max || threads != core)) { if (threads < core) { threadPoolExecutor.setCorePoolSize(threads); if (core == max) { threadPoolExecutor.setMaximumPoolSize(threads); } } else { threadPoolExecutor.setMaximumPoolSize(threads); if (core == max) { threadPoolExecutor.setCorePoolSize(threads); } } } } } catch (Throwable t) { logger.error(t.getMessage(), t); } super.setUrl(getUrl().addParameters(url.getParameters())); } @Override public void send(Object message, boolean sent) throws RemotingException { Collection<Channel> channels = getChannels(); for (Channel channel : channels) { if (channel.isConnected()) { channel.send(message, sent); } } } @Override public void close() { if (logger.isInfoEnabled()) { logger.info("Close " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress()); } ExecutorUtil.shutdownNow(executor, 100); try { super.close(); } catch (Throwable e) { logger.warn(e.getMessage(), e); } try { doClose(); } catch (Throwable e) { logger.warn(e.getMessage(), e); } } @Override public void close(int timeout) { ExecutorUtil.gracefulShutdown(executor, timeout); close(); } @Override public InetSocketAddress getLocalAddress() { return localAddress; } public InetSocketAddress getBindAddress() { return bindAddress; } public int getAccepts() { return accepts; } public int getIdleTimeout() { return idleTimeout; } @Override public void connected(Channel ch) throws RemotingException { // If the server has entered the shutdown process, reject any new connection if (this.isClosing() || this.isClosed()) { logger.warn("Close new channel " + ch + ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process."); ch.close(); return; } Collection<Channel> channels = getChannels(); if (accepts > 0 && channels.size() > accepts) { logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts); ch.close(); return; } super.connected(ch); } @Override public void disconnected(Channel ch) throws RemotingException { Collection<Channel> channels = getChannels(); if (channels.isEmpty()) { logger.warn("All clients has disconnected from " + ch.getLocalAddress() + ". You can graceful shutdown now."); } super.disconnected(ch); } }
AbstractServer的构造器会从url读取bindAddress、accepts、idleTimeout,然后执行doOpen方法;close方法会关闭executor,执行父类close方法,然后执行doClose方法;connected方法会先判断channels是否超出accepts值,超过则直接close;disconnected则执行父类disconnected方法
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyServer.java
public class NettyServer extends AbstractServer implements Server { private static final Logger logger = LoggerFactory.getLogger(NettyServer.class); /** * the cache for alive worker channel. * <ip:port, dubbo channel> */ private Map<String, Channel> channels; /** * netty server bootstrap. */ private ServerBootstrap bootstrap; /** * the boss channel that receive connections and dispatch these to worker channel. */ private io.netty.channel.Channel channel; private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; public NettyServer(URL url, ChannelHandler handler) throws RemotingException { // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants. // the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); } /** * Init and start netty server * * @throws Throwable */ @Override protected void doOpen() throws Throwable { bootstrap = new ServerBootstrap(); bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true)); final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); channels = nettyServerHandler.getChannels(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE) .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .childHandler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { // FIXME: should we use getTimeout()? int idleTimeout = UrlUtils.getIdleTimeout(getUrl()); NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug .addLast("decoder", adapter.getDecoder()) .addLast("encoder", adapter.getEncoder()) .addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS)) .addLast("handler", nettyServerHandler); } }); // bind ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); channelFuture.syncUninterruptibly(); channel = channelFuture.channel(); } @Override protected void doClose() throws Throwable { try { if (channel != null) { // unbind. channel.close(); } } catch (Throwable e) { logger.warn(e.getMessage(), e); } try { Collection<org.apache.dubbo.remoting.Channel> channels = getChannels(); if (channels != null && channels.size() > 0) { for (org.apache.dubbo.remoting.Channel channel : channels) { try { channel.close(); } catch (Throwable e) { logger.warn(e.getMessage(), e); } } } } catch (Throwable e) { logger.warn(e.getMessage(), e); } try { if (bootstrap != null) { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } catch (Throwable e) { logger.warn(e.getMessage(), e); } try { if (channels != null) { channels.clear(); } } catch (Throwable e) { logger.warn(e.getMessage(), e); } } @Override public Collection<Channel> getChannels() { Collection<Channel> chs = new HashSet<Channel>(); for (Channel channel : this.channels.values()) { if (channel.isConnected()) { chs.add(channel); } else { channels.remove(NetUtils.toAddressString(channel.getRemoteAddress())); } } return chs; } @Override public Channel getChannel(InetSocketAddress remoteAddress) { return channels.get(NetUtils.toAddressString(remoteAddress)); } @Override public boolean canHandleIdle() { return true; } @Override public boolean isBound() { return channel.isActive(); } }
NettyServer继承了AbstractServer,其实现了doOpen、doClose方法;doOpen方法会创建netty的ServerBootstrap、bossGroup、workerGroup;doClose方法会关闭channel,关闭bossGroup、workerGroup
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyTransporter.java
public class NettyTransporter implements Transporter { public static final String NAME = "netty"; @Override public Server bind(URL url, ChannelHandler listener) throws RemotingException { return new NettyServer(url, listener); } @Override public Client connect(URL url, ChannelHandler listener) throws RemotingException { return new NettyClient(url, listener); } }
NettyTransporter实现了Transporter接口,其bind方法创建的是NettyServer
NettyServer继承了AbstractServer,其实现了doOpen、doClose方法;doOpen方法会创建netty的ServerBootstrap、bossGroup、workerGroup;doClose方法会关闭channel,关闭bossGroup、workerGroup
看完上述内容,你们掌握dubbo中NettyServer的作用是什么的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。