1、An abstract IPC service. IPC calls take a single {@link Writable} as a parameter, and return a {@link Writable} as their value. A service runs on a port and is defined by a parameter class and a value class.
2、共有5个内部类ExceptionsHandler，Call，Listener，Responder，Connection,Handler
3、很多内部属性，位置比较散乱
private final boolean authorize;
private boolean isSecurityEnabled;
private ExceptionsHandler exceptionsHandler = new ExceptionsHandler();
private String bindAddress;
private int port; // port we listen on
private int handlerCount; // number of handler threads
private int readThreads; // number of read threads
private Class<? extends Writable> paramClass; // class of call parameters
private int maxIdleTime; // the maximum idle time after which a client may be disconnected
private int thresholdIdleConnections; // the number of idle connections after which we will start cleaning up idle connections
int maxConnectionsToNuke; // the max number of connections to nuke during a cleanup
protected RpcInstrumentation rpcMetrics;
private Configuration conf;
private SecretManager<TokenIdentifier> secretManager;
private int maxQueueSize;
private final int maxRespSize;
private int socketSendBufferSize;
private final boolean tcpNoDelay; // if T then disable Nagle‘s Algorithm
volatile private boolean running = true; // true while server runs
private BlockingQueue<Call> callQueue; // queued calls
private List<Connection> connectionList =
Collections.synchronizedList(new LinkedList<Connection>());
//maintain a list of client connections
private Listener listener = null;
private Responder responder = null;
private int numConnections = 0;
private Handler[] handlers = null;
4、这个Call和Client里边的Call是不一样的，
A call queued for handling.
private Writable param; // the parameter passed
private Connection connection; // connection to client
private long timestamp; // the time received when response is null ； the time served when response is not null
private ByteBuffer response; // the response for this call
5、Connection
Reads calls from a connection and queues them for handling.
private boolean rpcHeaderRead = false; // if initial rpc header is read
private boolean headerRead = false; //if the connection header that follows version is read.
private SocketChannel channel;
private ByteBuffer data;
private ByteBuffer dataLengthBuffer;
private LinkedList<Call> responseQueue;
private volatile int rpcCount = 0; // number of outstanding rpcs
private long lastContact;
private int dataLength;
private Socket socket;// Cache the remote host & port info so that even if the socket is disconnected, we can say where it used to connect to.
private String hostAddress;
private int remotePort;
private InetAddress addr;
ConnectionHeader header = new ConnectionHeader();
Class<?> protocol;
boolean useSasl;
SaslServer saslServer;
private AuthMethod authMethod;
private boolean saslContextEstablished;
private boolean skipInitialSaslHandshake;
private ByteBuffer rpcHeaderBuffer;
private ByteBuffer unwrappedData;
private ByteBuffer unwrappedDataLengthBuffer;
UserGroupInformation user = null;
public UserGroupInformation attemptingUser = null; // user name before auth
// Fake ‘call‘ for failed authorization response
private final int AUTHROIZATION_FAILED_CALLID = -1;
private final Call authFailedCall =
new Call(AUTHROIZATION_FAILED_CALLID, null, this);
private ByteArrayOutputStream authFailedResponse = new ByteArrayOutputStream();
// Fake ‘call‘ for SASL context setup
private static final int SASL_CALLID = -33;
private final Call saslCall = new Call(SASL_CALLID, null, this);
private final ByteArrayOutputStream saslResponse = new ByteArrayOutputStream();
private boolean useWrap = false;
6、ExceptionsHandler manages Exception groups for special handling e.g., terse exception group for concise logging messages
7、Handles queued calls .
8、Listens on the socket. Creates jobs for the handler threads
Listener里边有一个内部类，Reader
相应的属性有
private ServerSocketChannel acceptChannel = null; //the accept channel
private Selector selector = null; //the selector that we use for the server
private Reader[] readers = null;
private int currentReader = 0;
private InetSocketAddress address; //the address we bind at
private Random rand = new Random();
private long lastCleanupRunTime = 0; //the last time when a cleanup connec-
//-tion (for idle connections) ran
private long cleanupInterval = 10000; //the minimum interval between
//two cleanup runs
private int backlogLength = conf.getInt("ipc.server.listen.queue.size", 128);
private ExecutorService readPool;
9、Responder
Sends responses of RPC back to clients
相应的属性有
private Selector writeSelector;
private int pending; // connections waiting to register
final static int PURGE_INTERVAL = 900000; // 15mins