RocketMQ源码分析1--Remoting

ServerBootstrap childHandler =
         this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
             .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .option(ChannelOption.SO_REUSEADDR, true)
             .option(ChannelOption.SO_KEEPALIVE, false)
             .childOption(ChannelOption.TCP_NODELAY, true)
             .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
             .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
             .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline()
                         .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                             new HandshakeHandler(TlsSystemConfig.tlsMode))
                         .addLast(defaultEventExecutorGroup,
                             new NettyEncoder(), // 指定消息编码器
                             new NettyDecoder(), // 指定消息解码器
                             new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                             new NettyConnectManageHandler(),
                             new NettyServerHandler()
                         );
                 }
             });

pipeline.fireChannelRead(readBuf.get(i));

public final ChannelPipeline fireChannelRead(Object msg) {
     AbstractChannelHandlerContext.invokeChannelRead(head, msg);
     return this;
 }

//RequestCode定义:当表示请求操作代码时候，请求接收方根据代码执行相应操作；
//当表示应答结果代码时候，0表示成功，非0表示错误代码。
private int code;
private LanguageCode language = LanguageCode.JAVA; // 请求和应答方语言
private int version = 0; //请求和应答方程序版本
private int opaque = RequestId.getAndIncrement();//请求发起方做tcp连接上的线程复用。
private int flag = 0; //通信层标志位
private String remark;
//把下面customHeader中的信息填充到这里，见 makeCustomHeaderToNet
//"extFields":{"topic":"yyztest2","queueId":"3","consumerGroup":"yyzGroup2","commitOffset":"28"}
private HashMap<String, String> extFields;//请求自定义字段

/**
 * header data
 * 例如CONSUMER_SEND_MSG_BACK消息，
 * customHeader 为ConsumerSendMsgBackRequestHeader  填充见 MQClientAPIImpl.consumerSendMessageBack
 */
private transient CommandCustomHeader customHeader;

/**
 * body部分
 */
private transient byte[] body;

Message msg = new Message("Topic2Test",// topic
                        "TagA",// tag
                        ("Hello RocketMQ ").getBytes()// body
                );

SendResult sendResult = producer.send(msg);

public static final int SEND_MESSAGE_V2 = 310;

/**
 * This container holds all processors per request code, aka, for each incoming request, we may look up the
 * responding processor in this map to handle the request.
 */
protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable =
    new HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>>(64);

/**
 * This map caches all on-going requests.
 */
protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable = new ConcurrentHashMap<Integer, ResponseFuture>(256);

 /**
  * addr:为服务端地址，RPC调用需要制定服务端地址这样客户端才可以发送请求
  * request: RemotingCommand封装的消息
  */
 public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
        final Channel channel = this.getAndCreateChannel(addr);
        if (channel != null && channel.isActive()) {
            try {
                if (this.rpcHook != null) {
                    this.rpcHook.doBeforeRequest(addr, request);
                }
                System.out.println("invoke sync! request: "+request.toString());
                // 同步通信，通信回应存储在reponse里
                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis);
                if (this.rpcHook != null) {
                    this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                }
                return response;
            }
            ...
        }
    }
}

public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
       final long timeoutMillis)
       throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
       final int opaque = request.getOpaque();

       try {
           final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, null, null);
           // 客户端使用resonseTable,服务端使用processortable
           this.responseTable.put(opaque, responseFuture);
           final SocketAddress addr = channel.remoteAddress();
           channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
               @Override
               public void operationComplete(ChannelFuture f) throws Exception {
                   if (f.isSuccess()) {
                       responseFuture.setSendRequestOK(true);
                       return;
                   } else {
                       responseFuture.setSendRequestOK(false);
                   }

                   responseTable.remove(opaque);
                   responseFuture.setCause(f.cause());
                   responseFuture.putResponse(null);
                   log.warn("send a request command to channel <" + addr + "> failed.");
               }
           });

           RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
           if (null == responseCommand) {
               if (responseFuture.isSendRequestOK()) {
                   throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                       responseFuture.getCause());
               } else {
                   throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
               }
           }

           return responseCommand;
       } finally {
           this.responseTable.remove(opaque);
       }
   }