nodejs源码分析之connect

theanarkh

发布于 2020-07-27 22:25:11

7440

发布于 2020-07-27 22:25:11

今天我们来分析connect函数。connect是发起tcp连接的api。本质上是对底层tcp协议connect函数的封装。我们看一下nodejs里做了什么事情。我们首先看一下connect函数的入口定义。

// connect(options, [cb])
// connect(port, [host], [cb])
// connect(path, [cb]);
// 对socket connect的封装
function connect(...args) {
  // 处理参数
  var normalized = normalizeArgs(args);
  var options = normalized[0];
  // 申请一个socket表示一个连接
  var socket = new Socket(options);
  // 设置连接超时时间
  if (options.timeout) {
    socket.setTimeout(options.timeout);
  }
  // 调用socket的connect
  return Socket.prototype.connect.call(socket, normalized);
}

从代码中可以发现，connect函数是对Socket对象的封装。Socket表示一个tcp连接。我们分成三部分分析。 1 new Socket 2 setTimeout 3 Socket的connect

1 new Socket 我们看看新建一个Socket对象，做了什么事情。

function Socket(options) {
  if (!(this instanceof Socket)) return new Socket(options);
  // 是否正在建立连接，即三次握手中
  this.connecting = false;
  // 触发close事件时，该字段标记是否由于错误导致了close事件
  this._hadError = false;
  // 对应的底层handle，比如tcp
  this._handle = null;
  // 定时器id
  this[kTimeout] = null;
  options = options || {};
  // 双工
  stream.Duplex.call(this, options);
  // 还不能读写，先设置成false
  // these will be set once there is a connection
  this.readable = this.writable = false;
  this.on('finish', onSocketFinish);
  this.on('_socketEnd', onSocketEnd);
  // 是否允许单工
  this.allowHalfOpen = options && options.allowHalfOpen || false;
}

其实也没有做太多的事情，就是初始化一些属性。

2 setTimeout

Socket.prototype.setTimeout = function(msecs, callback) {
  // 清除之前的，如果有的话
  clearTimeout(this[kTimeout]);
  // 0代表清除
  if (msecs === 0) {
    if (callback) {
      this.removeListener('timeout', callback);
    }
  } else {
    // 开启一个定时器，超时时间是msecs，超时回调是_onTimeout
    this[kTimeout] = setUnrefTimeout(this._onTimeout.bind(this), msecs);
    // 监听timeout事件，定时器超时时，底层会调用nodejs的回调，nodejs会调用用户的回调callback
    if (callback) {
      this.once('timeout', callback);
    }
  }
  return this;
};

setTimeout做的事情就是设置一个超时时间，如果超时则执行回调，在回调里再触发用户传入的回调。我们看一下超时处理函数_onTimeout。

Socket.prototype._onTimeout = function() {
  this.emit('timeout');
};

直接触发timeout函数，回调用户的函数。

3 connect函数

// 建立连接，即三次握手
Socket.prototype.connect = function(...args) {
  let normalized;
  /* 忽略参数处理 */
  var options = normalized[0];
  var cb = normalized[1];

  if (this.write !== Socket.prototype.write)
    this.write = Socket.prototype.write;

  this._handle = new TCP(TCPConstants.SOCKET);
  this._handle.onread = onread;
  // 连接成功，执行的回调
  if (cb !== null) {
    this.once('connect', cb);
  }
  // 正在连接
  this.connecting = true;
  this.writable = true;
  // 可能需要dns解析，解析成功再发起连接
  lookupAndConnect(this, options);
  return this;
};

connect 函数主要是三个逻辑 1 首先创建一个底层的handle，比如我们这里是tcp（对应tcp_wrap.cc的实现）。 2 设置一些回调 3 做dns解析（如果需要的话），然后发起三次握手。我们不展开dns解析的逻辑，这个留给分析dns模块的时候。我们直接看dns解析成功（或者不需要dns）时的逻辑。

function internalConnect(
  self, 
  // 需要连接的远端ip、端口
  address, 
  port, 
  addressType, 
  // 用于和对端连接的本地ip、端口（如果不设置，则操作系统自己决定）
  localAddress, 
  localPort) {
  var err;
  // 如果传了本地的地址或端口，则tcp连接中的源ip和端口就是传的，否则由操作系统自己选
  if (localAddress || localPort) {
      // ip v4
    if (addressType === 4) {
      localAddress = localAddress || '0.0.0.0';
      // 绑定地址和端口到handle，类似设置handle对象的两个属性
      err = self._handle.bind(localAddress, localPort);
    } else if (addressType === 6) {
      localAddress = localAddress || '::';
      err = self._handle.bind6(localAddress, localPort);
    }

    // 绑定是否成功
    err = checkBindError(err, localPort, self._handle);
    if (err) {
      const ex = exceptionWithHostPort(err, 'bind', localAddress, localPort);
      self.destroy(ex);
      return;
    }
  }
  if (addressType === 6 || addressType === 4) {
    // 新建一个请求对象，是一个c++对象
    const req = new TCPConnectWrap();
    // 设置一些列属性
    req.oncomplete = afterConnect;
    req.address = address;
    req.port = port;
    req.localAddress = localAddress;
    req.localPort = localPort;
    // 调用底层对应的函数
    if (addressType === 4)
      err = self._handle.connect(req, address, port);
    else
      err = self._handle.connect6(req, address, port);
  }
  // 非阻塞调用，可能在还没发起三次握手之前就报错了，而不是三次握手出错，这里进行出错处理
  if (err) {
    // 获取socket对应的底层ip端口信息
    var sockname = self._getsockname();
    var details;

    if (sockname) {
      details = sockname.address + ':' + sockname.port;
    }

    const ex = exceptionWithHostPort(err, 'connect', address, port, details);
    self.destroy(ex);
  }
}

这里的代码比较多，除了错误处理外，主要的逻辑是bind和connect。bind函数的逻辑很简单（即使是底层的bind），他就是在底层的一个对象上设置了两个字段的值。所以我们主要来分析connect。我们把关于connect的这段逻辑拎出来。

    const req = new TCPConnectWrap();
    // 设置一些列属性
    req.oncomplete = afterConnect;
    req.address = address;
    req.port = port;
    req.localAddress = localAddress;
    req.localPort = localPort;
    // 调用底层对应的函数
    self._handle.connect(req, address, port);

void TCPWrap::Connect(const FunctionCallbackInfo<Value>& args) {
  Environment* env = Environment::GetCurrent(args);

  TCPWrap* wrap;
  ASSIGN_OR_RETURN_UNWRAP(&wrap,
                          args.Holder(),
                          args.GetReturnValue().Set(UV_EBADF));

  CHECK(args[0]->IsObject());
  CHECK(args[1]->IsString());
  CHECK(args[2]->IsUint32());
  Local<Object> req_wrap_obj = args[0].As<Object>();
  // 要连接的ip和端口
  node::Utf8Value ip_address(env->isolate(), args[1]);
  int port = args[2]->Uint32Value();

  sockaddr_in addr;
  int err = uv_ip4_addr(*ip_address, port, &addr);

  if (err == 0) {
    // 新建一个request代表本次的connect操作
    ConnectWrap* req_wrap =  new ConnectWrap(env, 
                                             req_wrap_obj, 
                                             AsyncWrap::PROVIDER_TCPCONNECTWRAP);
    err = uv_tcp_connect(req_wrap->req(),
                         &wrap->handle_,
                         reinterpret_cast<const sockaddr*>(&addr),
                         AfterConnect);
    req_wrap->Dispatched();
    if (err)
      delete req_wrap;
  }

  args.GetReturnValue().Set(err);
}

我们不深入底层分析connect函数的实现，有兴趣的可以参考之前的一些文章。这里主要是申请一个request对象，然后针对该handle，进行connect操作（libuv中的handle和request）。因为是非阻塞式调用，所以设置了回调AfterConnect。假设连接建立，这时候就会执行AfterConnect。AfterConnect就会执行js层的oncomplete函数，oncomplete函数指向的是afterConnect。

function afterConnect(status, handle, req, readable, writable) {
  var self = handle.owner;

  handle = self._handle;

  self.connecting = false;
  self._sockname = null;
  // 连接成功
  if (status === 0) {
    self.readable = readable;
    self.writable = writable;
    self.emit('connect');

    if (readable && !self.isPaused())
      self.read(0);
    }
  }
  // 错误处理
}

连接成功后js层调用了self.read(0)注册等待可读事件（可参考之前的文章记一次nodejs问题排查）。至此，connect函数分析完毕。本文对connect函数进行了粗略的分析，如果有兴趣，欢迎交流。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2020-07-27，如有侵权请联系 cloudcommunity@tencent.com 删除

socket编程