SOFABolt 是蚂蚁集团开发的一套基于 Netty 实现的网络通信框架

/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package com.alipay.remoting.rpc; import java.io.FileInputStream; import java.io.InputStream; import java.net.InetSocketAddress; import java.security.KeyStore; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLEngine; import javax.net.ssl.TrustManagerFactory; import com.alipay.remoting.*; import com.alipay.remoting.config.BoltGenericOption; import com.alipay.remoting.config.BoltServerOption; import org.slf4j.Logger; import com.alipay.remoting.codec.Codec; import com.alipay.remoting.config.ConfigManager; import com.alipay.remoting.constant.Constants; import com.alipay.remoting.exception.RemotingException; import com.alipay.remoting.log.BoltLoggerFactory; import com.alipay.remoting.rpc.protocol.UserProcessor; import com.alipay.remoting.rpc.protocol.UserProcessorRegisterHelper; import com.alipay.remoting.util.IoUtils; import com.alipay.remoting.util.NettyEventLoopUtil; import com.alipay.remoting.util.RemotingUtil; import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; import io.netty.channel.WriteBufferWaterMark; import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.SslHandler; import io.netty.handler.flush.FlushConsolidationHandler; import io.netty.handler.timeout.IdleStateHandler; /** * Server for Rpc. * * Usage: * You can initialize RpcServer with one of the three constructors: * {@link #RpcServer(int)}, {@link #RpcServer(int, boolean)}, {@link #RpcServer(int, boolean, boolean)} * Then call start() to start a rpc server, and call stop() to stop a rpc server. * * Notice: * Once rpc server has been stopped, it can never be start again. You should init another instance of RpcServer to use. * * @author jiangping * @version $Id: RpcServer.java, v 0.1 2015-8-31 PM5:22:22 tao Exp $ */ public class RpcServer extends AbstractRemotingServer { /** logger */ private static final Logger logger = BoltLoggerFactory .getLogger("RpcRemoting"); /** server bootstrap */ private ServerBootstrap bootstrap; /** channelFuture */ private ChannelFuture channelFuture; /** connection event handler */ private ConnectionEventHandler connectionEventHandler; /** connection event listener */ private ConnectionEventListener connectionEventListener = new ConnectionEventListener(); /** user processors of rpc server */ private ConcurrentHashMap<String, UserProcessor<?>> userProcessors = new ConcurrentHashMap<String, UserProcessor<?>>( 4); /** boss event loop group, boss group should not be daemon, need shutdown manually*/ private final EventLoopGroup bossGroup = NettyEventLoopUtil .newEventLoopGroup( 1, new NamedThreadFactory( "Rpc-netty-server-boss", false)); /** worker event loop group. Reuse I/O worker threads between rpc servers. */ private static final EventLoopGroup workerGroup = NettyEventLoopUtil .newEventLoopGroup( Runtime .getRuntime() .availableProcessors() * 2, new NamedThreadFactory( "Rpc-netty-server-worker", true)); /** address parser to get custom args */ private RemotingAddressParser addressParser; /** connection manager */ private DefaultServerConnectionManager connectionManager; /** rpc remoting */ protected RpcRemoting rpcRemoting; /** rpc codec */ private Codec codec = new RpcCodec(); static { if (workerGroup instanceof NioEventLoopGroup) { ((NioEventLoopGroup) workerGroup).setIoRatio(ConfigManager.netty_io_ratio()); } else if (workerGroup instanceof EpollEventLoopGroup) { ((EpollEventLoopGroup) workerGroup).setIoRatio(ConfigManager.netty_io_ratio()); } } /** * Construct a rpc server with random port <br> * the random port will determined after server startup * Note:<br> * You can only use invoke methods with params {@link Connection}, for example {@link #invokeSync(Connection, Object, int)} <br> * Otherwise {@link UnsupportedOperationException} will be thrown. */ public RpcServer() { this(false); } /** * Construct a rpc server with random port <br> * the random port will determined after server startup * Note:<br> * You can only use invoke methods with params {@link Connection}, for example {@link #invokeSync(Connection, Object, int)} <br> * Otherwise {@link UnsupportedOperationException} will be thrown. * * @param manageConnection true to enable connection management feature */ public RpcServer(boolean manageConnection) { this(0, manageConnection); } /** * Construct a rpc server. <br> * * Note:<br> * You can only use invoke methods with params {@link Connection}, for example {@link #invokeSync(Connection, Object, int)} <br> * Otherwise {@link UnsupportedOperationException} will be thrown. */ public RpcServer(int port) { this(port, false); } /** * Construct a rpc server. <br> * * Note:<br