предисловие
В этой серии статей будет подробно рассказано о постепенном обогащении функций NIO, что проложит путь к объяснению библиотеки Reactor-Netty.
О методологии программирования на Java: совместное использование видео Reactor и Webflux, Rxjava и Reactor завершены, адрес станции b следующий:
Интерпретация исходного кода Rxjava и совместное использование:вооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооо
Интерпретация и совместное использование исходного кода Reactor:вооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооооо
Интерпретация исходного кода этой серии основана на деталях API JDK11, которые могут отличаться от других версий.Пожалуйста, решите проблему с версией JDK самостоятельно.
Интерпретация канала
Продолжение предыдущей статьиНекоторые сведения об исходном коде BIO to NIO на NIO
Дайте каналу возможность поддерживать сетевые сокеты
Наша первоначальная цель — улучшить сокет. Исходя из этого основного требования, нет условий для создания условий. Поэтому, чтобы позволить каналу иметь возможность сетевых сокетов, мы определяемjava.nio.channels.NetworkChannelинтерфейс. Без лишних слов давайте посмотрим на определение этого интерфейса:
public interface NetworkChannel extends Channel
{
NetworkChannel bind(SocketAddress local) throws IOException;
SocketAddress getLocalAddress() throws IOException;
<T> NetworkChannel setOption(SocketOption<T> name, T value) throws IOException;
<T> T getOption(SocketOption<T> name) throws IOException;
Set<SocketOption<?>> supportedOptions();
}
пройти черезbind(SocketAddress)метод будетsocketпривязать к локальномуSocketAddress, возвращаемый методом getLocalAddress()socketадрес привязки,
пройти черезsetOption(SocketOption,Object)а такжеgetOption(SocketOption)Настройки метода и запросыsocketПоддерживаемые параметры конфигурации.
bind
Далее мы смотрим наjava.nio.channels.ServerSocketChannelАбстрактный класс и класс его реализацииsun.nio.ch.ServerSocketChannelImplПодробности его реализации.
Во-первых, давайте посмотрим на его реализацию bind:
//sun.nio.ch.ServerSocketChannelImpl#bind
@Override
public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
synchronized (stateLock) {
ensureOpen();
//通过localAddress判断是否已经调用过bind
if (localAddress != null)
throw new AlreadyBoundException();
//InetSocketAddress(0)表示绑定到本机的所有地址，由操作系统选择合适的端口
InetSocketAddress isa = (local == null)
? new InetSocketAddress(0)
: Net.checkAddress(local);
SecurityManager sm = System.getSecurityManager();
if (sm != null)
sm.checkListen(isa.getPort());
NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
Net.bind(fd, isa.getAddress(), isa.getPort());
//开启监听，s如果参数backlog小于1，默认接受50个连接
Net.listen(fd, backlog < 1 ? 50 : backlog);
localAddress = Net.localAddress(fd);
}
return this;
}
Давайте посмотрим, как реализованы методы bind и listen в Net.
Net.bind
//sun.nio.ch.Net#bind(java.io.FileDescriptor, java.net.InetAddress, int)
public static void bind(FileDescriptor fd, InetAddress addr, int port)
throws IOException
{
bind(UNSPEC, fd, addr, port);
}
static void bind(ProtocolFamily family, FileDescriptor fd,
InetAddress addr, int port) throws IOException
{
//如果传入的协议域不是IPV4而且支持IPV6,则使用ipv6
boolean preferIPv6 = isIPv6Available() &&
(family != StandardProtocolFamily.INET);
bind0(fd, preferIPv6, exclusiveBind, addr, port);
}
private static native void bind0(FileDescriptor fd, boolean preferIPv6,
boolean useExclBind, InetAddress addr,
int port)
throws IOException;
bind0 реализован как нативный метод:
JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_bind0(JNIEnv *env, jclass clazz, jobject fdo, jboolean preferIPv6,
jboolean useExclBind, jobject iao, int port)
{
SOCKETADDRESS sa;
int sa_len = 0;
int rv = 0;
//将java的InetAddress转换为c的struct sockaddr
if (NET_InetAddressToSockaddr(env, iao, port, &sa, &sa_len,
preferIPv6) != 0) {
return;//转换失败，方法返回
}
//调用bind方法:int bind(int sockfd, struct sockaddr* addr, socklen_t addrlen)
rv = NET_Bind(fdval(env, fdo), &sa, sa_len);
if (rv != 0) {
handleSocketError(env, errno);
}
}
Сокет — это центр обмена информацией между пользовательскими программами и ядром.Он не имеет такой информации, как адреса сетевых протоколов и номера портов.При осуществлении сетевого взаимодействия сокет должен быть связан с адресом. Во многих случаях ядро автоматически привязывает нам адрес, но иногда пользователям может потребоваться выполнить процесс привязки самостоятельно, чтобы удовлетворить потребности практических приложений; Наиболее типичная ситуация заключается в том, что серверному процессу необходимо выполнить привязку к известному адресу или порту, чтобы дождаться подключения клиентов. Для клиента во многих случаях метод связывания не нужно вызывать, он автоматически связывается ядром;
Обратите внимание здесь, привязка привязана, при наличии соединения будет создан новый сокет, после чего сервер сможет работать с новым сокетом. Следуйте здесьacceptметод. Зависит отsun.nio.ch.ServerSocketChannelImpl#bindНаконец, мы знаем, что егоNet.listen(fd, backlog < 1 ? 50 : backlog)Включить мониторинг.Если отставание параметра меньше 1, по умолчанию принимается 50 подключений. Следовательно, обратим внимание наNet.listenДетали метода.
Net.listen
//sun.nio.ch.Net#listen
static native void listen(FileDescriptor fd, int backlog) throws IOException;
Знать,Net.listenдаnativeметод, исходный код выглядит следующим образом:
JNIEXPORT void JNICALL
Java_sun_nio_ch_Net_listen(JNIEnv *env, jclass cl, jobject fdo, jint backlog)
{
if (listen(fdval(env, fdo), backlog) < 0)
handleSocketError(env, errno);
}
Вы можете видеть, что нижний слой - это вызовlistenосуществленный,listenФункция обычно называетсяbindпозже позвонитьacceptВызванный ранее, его прототип функции:int listen(int sockfd, int backlog)Возвращаемое значение: 0 в случае успеха, -1 в случае неудачи.
Обратим внимание на другие детали в операции привязки, в началеensureOpen()Оценка метода:
//sun.nio.ch.ServerSocketChannelImpl#ensureOpen
// @throws ClosedChannelException if channel is closed
private void ensureOpen() throws ClosedChannelException {
if (!isOpen())
throw new ClosedChannelException();
}
//java.nio.channels.spi.AbstractInterruptibleChannel#isOpen
public final boolean isOpen() {
return !closed;
}
еслиsocketзакрыть, броситьClosedChannelException.
давайте посмотрим еще разNet#checkAddress:
//sun.nio.ch.Net#checkAddress(java.net.SocketAddress)
public static InetSocketAddress checkAddress(SocketAddress sa) {
if (sa == null)//地址为空
throw new NullPointerException();
//非InetSocketAddress类型地址
if (!(sa instanceof InetSocketAddress))
throw new UnsupportedAddressTypeException(); // ## needs arg
InetSocketAddress isa = (InetSocketAddress)sa;
//地址不可识别
if (isa.isUnresolved())
throw new UnresolvedAddressException(); // ## needs arg
InetAddress addr = isa.getAddress();
//非ip4和ip6地址
if (!(addr instanceof Inet4Address || addr instanceof Inet6Address))
throw new IllegalArgumentException("Invalid address type");
return isa;
}
Как видно из вышесказанного, bind first проверяетServerSocketТо ли закрывать, то ли привязывать адрес, если ни привязывать, ни закрывать, проверять привязкуsocketaddressЭто правильно или законно; затем через класс Net toolbindа такжеlisten, завершает фактическийServerSocketПривязка адреса и включение мониторинга, если привязка включена, параметр меньше1, по умолчанию принимается50связь.
В отличие от БИО, с которыми мы связались в первой статье, давайте рассмотрим некоторыеaccept()Реализация метода:
//sun.nio.ch.ServerSocketChannelImpl#accept()
@Override
public SocketChannel accept() throws IOException {
acceptLock.lock();
try {
int n = 0;
FileDescriptor newfd = new FileDescriptor();
InetSocketAddress[] isaa = new InetSocketAddress[1];
boolean blocking = isBlocking();
try {
begin(blocking);
do {
n = accept(this.fd, newfd, isaa);
} while (n == IOStatus.INTERRUPTED && isOpen());
} finally {
end(blocking, n > 0);
assert IOStatus.check(n);
}
if (n < 1)
return null;
//针对接受连接的处理通道socketchannelimpl，默认为阻塞模式
// newly accepted socket is initially in blocking mode
IOUtil.configureBlocking(newfd, true);
InetSocketAddress isa = isaa[0];
//构建SocketChannelImpl，这个具体在SocketChannelImpl再说
SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
// check permitted to accept connections from the remote address
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
//检查地址和port权限
sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
} catch (SecurityException x) {
sc.close();
throw x;
}
}
//返回socketchannelimpl
return sc;
} finally {
acceptLock.unlock();
}
}
заaccept(this.fd, newfd, isaa), вызовите accept, чтобы получить установленное соединение в сокете, как мы узнали ранее в BIO, функция в итоге вызовет: int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);
- Если в очереди слушающего сокета fd нет ожидающих соединений, и сокет не помечен как неблокирующий, accept() будет блокироваться до тех пор, пока не появится соединение;
- Если сокет помечен как неблокирующий и в очереди нет ожидающих соединений, accept() возвращает ошибку EAGAIN или EWOULDBLOCK.
здесьbegin(blocking);а такжеend(blocking, n > 0);Модель сотрудничества, в которой мы находимсяInterruptibleChannel и прерываемый ввод-выводЭто уже было описано в этой статье.Я упомяну это здесь еще раз, чтобы каждый мог увидеть его применение.Фокус здесь заключается в ожидании процесса подключения, во время которого могут быть аварийные прерывания.Если процесс завершится нормально, он будет выполняться нормально. Логика, не делай вид, что этот Канал вот-вот закончится,end(blocking, n > 0)Второй параметр завершен только для того, чтобы судить, закончился ли процесс ожидания, не расширяя объем функций.
supportedOptions
давайте посмотрим еще разNetworkChannelДругие методы достижения, сначала посмотрите наsupportedOptions:
//sun.nio.ch.ServerSocketChannelImpl#supportedOptions
@Override
public final Set<SocketOption<?>> supportedOptions() {
return DefaultOptionsHolder.defaultOptions;
}
//sun.nio.ch.ServerSocketChannelImpl.DefaultOptionsHolder
private static class DefaultOptionsHolder {
static final Set<SocketOption<?>> defaultOptions = defaultOptions();
private static Set<SocketOption<?>> defaultOptions() {
HashSet<SocketOption<?>> set = new HashSet<>();
set.add(StandardSocketOptions.SO_RCVBUF);
set.add(StandardSocketOptions.SO_REUSEADDR);
if (Net.isReusePortAvailable()) {
set.add(StandardSocketOptions.SO_REUSEPORT);
}
set.add(StandardSocketOptions.IP_TOS);
set.addAll(ExtendedSocketOptions.options(SOCK_STREAM));
//返回不可修改的HashSet
return Collections.unmodifiableSet(set);
}
}
Давайте взглянем на некоторые из приведенных выше конфигураций:
//java.net.StandardSocketOptions
//socket接受缓存大小
public static final SocketOption<Integer> SO_RCVBUF =
new StdSocketOption<Integer>("SO_RCVBUF", Integer.class);
//是否可重用地址
public static final SocketOption<Boolean> SO_REUSEADDR =
new StdSocketOption<Boolean>("SO_REUSEADDR", Boolean.class);
//是否可重用port
public static final SocketOption<Boolean> SO_REUSEPORT =
new StdSocketOption<Boolean>("SO_REUSEPORT", Boolean.class);
//Internet协议（IP）标头（header）中的服务类型（ToS）。
public static final SocketOption<Integer> IP_TOS =
new StdSocketOption<Integer>("IP_TOS", Integer.class);
Реализация setOption
Зная приведенную выше поддерживаемую конфигурацию, давайте посмотримsetOptionДетали реализации:
//sun.nio.ch.ServerSocketChannelImpl#setOption
@Override
public <T> ServerSocketChannel setOption(SocketOption<T> name, T value)
throws IOException
{
Objects.requireNonNull(name);
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
ensureOpen();
if (name == StandardSocketOptions.IP_TOS) {
ProtocolFamily family = Net.isIPv6Available() ?
StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;
Net.setSocketOption(fd, family, name, value);
return this;
}
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
isReuseAddress = (Boolean)value;
} else {
// no options that require special handling
Net.setSocketOption(fd, Net.UNSPEC, name, value);
}
return this;
}
}
Здесь вы можете увидетьsupportedOptions().contains(name)Функция , во-первых, будет выполняться оценка конфигурации поддержки, а затем будет выполняться обычная логика настройки. В основном он выполняет настройки конфигурации сокета.Net.setSocketOption, тут только китайские комментарии к коду делайте, вся логика процесса не слишком сложная.
static void setSocketOption(FileDescriptor fd, ProtocolFamily family,
SocketOption<?> name, Object value)
throws IOException
{
if (value == null)
throw new IllegalArgumentException("Invalid option value");
// only simple values supported by this method
Class<?> type = name.type();
if (extendedOptions.isOptionSupported(name)) {
extendedOptions.setOption(fd, name, value);
return;
}
//非整形和布尔型，则抛出断言错误
if (type != Integer.class && type != Boolean.class)
throw new AssertionError("Should not reach here");
// special handling
if (name == StandardSocketOptions.SO_RCVBUF ||
name == StandardSocketOptions.SO_SNDBUF)
{
//判断接受和发送缓冲区大小
int i = ((Integer)value).intValue();
if (i < 0)
throw new IllegalArgumentException("Invalid send/receive buffer size");
}
//缓冲区有数据，延迟关闭socket的的时间
if (name == StandardSocketOptions.SO_LINGER) {
int i = ((Integer)value).intValue();
if (i < 0)
value = Integer.valueOf(-1);
if (i > 65535)
value = Integer.valueOf(65535);
}
//UDP单播
if (name == StandardSocketOptions.IP_TOS) {
int i = ((Integer)value).intValue();
if (i < 0 || i > 255)
throw new IllegalArgumentException("Invalid IP_TOS value");
}
//UDP多播
if (name == StandardSocketOptions.IP_MULTICAST_TTL) {
int i = ((Integer)value).intValue();
if (i < 0 || i > 255)
throw new IllegalArgumentException("Invalid TTL/hop value");
}
// map option name to platform level/name
OptionKey key = SocketOptionRegistry.findOption(name, family);
if (key == null)
throw new AssertionError("Option not found");
int arg;
//转换配置参数值
if (type == Integer.class) {
arg = ((Integer)value).intValue();
} else {
boolean b = ((Boolean)value).booleanValue();
arg = (b) ? 1 : 0;
}
boolean mayNeedConversion = (family == UNSPEC);
boolean isIPv6 = (family == StandardProtocolFamily.INET6);
//设置文件描述符的值及其他
setIntOption0(fd, mayNeedConversion, key.level(), key.name(), arg, isIPv6);
}
getOption
Далее посмотримgetOptionИсходный код выглядит следующим образом:
//sun.nio.ch.ServerSocketChannelImpl#getOption
@Override
@SuppressWarnings("unchecked")
public <T> T getOption(SocketOption<T> name)
throws IOException
{
Objects.requireNonNull(name);
//非通道支持选项，则抛出UnsupportedOperationException
if (!supportedOptions().contains(name))
throw new UnsupportedOperationException("'" + name + "' not supported");
synchronized (stateLock) {
ensureOpen();
if (name == StandardSocketOptions.SO_REUSEADDR && Net.useExclusiveBind()) {
// SO_REUSEADDR emulated when using exclusive bind
return (T)Boolean.valueOf(isReuseAddress);
}
//假如获取的不是上面的配置，则委托给Net来处理
// no options that require special handling
return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
}
}
//sun.nio.ch.Net#getSocketOption
static Object getSocketOption(FileDescriptor fd, ProtocolFamily family,
SocketOption<?> name)
throws IOException
{
Class<?> type = name.type();
if (extendedOptions.isOptionSupported(name)) {
return extendedOptions.getOption(fd, name);
}
//只支持整形和布尔型，否则抛出断言错误
// only simple values supported by this method
if (type != Integer.class && type != Boolean.class)
throw new AssertionError("Should not reach here");
// map option name to platform level/name
OptionKey key = SocketOptionRegistry.findOption(name, family);
if (key == null)
throw new AssertionError("Option not found");
boolean mayNeedConversion = (family == UNSPEC);
//获取文件描述的选项配置
int value = getIntOption0(fd, mayNeedConversion, key.level(), key.name());
if (type == Integer.class) {
return Integer.valueOf(value);
} else {
//我们要看到前面支持配置处的源码其支持的类型要么是Boolean，要么是Integer
//所以，返回值为Boolean.FALSE 或 Boolean.TRUE也就不足为奇了
return (value == 0) ? Boolean.FALSE : Boolean.TRUE;
}
}
Сходства и различия между ServerSocketChannel и ServerSocket при привязке
существуетNet.bindВ разделе мы, наконец, упомянули примечание.При каждом подключении будет создан сокет для работы соединения.Это можно увидеть в методе accept.После получения соединения он будетnew SocketChannelImpl(provider(), newfd, isa)этот объект. Здесь поднимается тема. Когда мы используем метод привязки, должны ли мы также привязываться к сокету? Как раньше работало био, давайте сначала рассмотрим его.
мы звонилиjava.net.ServerSocket#ServerSocket(int, int, java.net.InetAddress)метод, естьsetImpl():
//java.net.ServerSocket
public ServerSocket(int port, int backlog, InetAddress bindAddr) throws IOException {
setImpl();
if (port < 0 || port > 0xFFFF)
throw new IllegalArgumentException(
"Port value out of range: " + port);
if (backlog < 1)
backlog = 50;
try {
bind(new InetSocketAddress(bindAddr, port), backlog);
} catch(SecurityException e) {
close();
throw e;
} catch(IOException e) {
close();
throw e;
}
}
//java.net.ServerSocket#setImpl
private void setImpl() {
if (factory != null) {
impl = factory.createSocketImpl();
checkOldImpl();
} else {
// No need to do a checkOldImpl() here, we know it's an up to date
// SocketImpl!
impl = new SocksSocketImpl();
}
if (impl != null)
impl.setServerSocket(this);
}
Однако в центре нашего внимания здесьbind(new InetSocketAddress(bindAddr, port), backlog);, код здесь следующий:
//java.net.ServerSocket
public void bind(SocketAddress endpoint, int backlog) throws IOException {
if (isClosed())
throw new SocketException("Socket is closed");
if (!oldImpl && isBound())
throw new SocketException("Already bound");
if (endpoint == null)
endpoint = new InetSocketAddress(0);
if (!(endpoint instanceof InetSocketAddress))
throw new IllegalArgumentException("Unsupported address type");
InetSocketAddress epoint = (InetSocketAddress) endpoint;
if (epoint.isUnresolved())
throw new SocketException("Unresolved address");
if (backlog < 1)
backlog = 50;
try {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkListen(epoint.getPort());
//重点！！
getImpl().bind(epoint.getAddress(), epoint.getPort());
getImpl().listen(backlog);
bound = true;
} catch(SecurityException e) {
bound = false;
throw e;
} catch(IOException e) {
bound = false;
throw e;
}
}
мы виделиgetImpl()Я отметил основные моменты, что здесь происходит, и мы входим:
//java.net.ServerSocket#getImpl
SocketImpl getImpl() throws SocketException {
if (!created)
createImpl();
return impl;
}
во всем процессеcreatedИли начальное значение, когда объект только создавался, если оно ложно, то обязательно войдетcreateImpl()В методе:
//java.net.ServerSocket#createImpl
void createImpl() throws SocketException {
if (impl == null)
setImpl();
try {
impl.create(true);
created = true;
} catch (IOException e) {
throw new SocketException(e.getMessage());
}
}
А здесь, потому что предыдущийimplОн назначен, так что он будет идтиimpl.create(true), а затемcreatedустановить какtrue. А теперь, наконец, к тому, о чем я хочу поговорить:
//java.net.AbstractPlainSocketImpl#create
protected synchronized void create(boolean stream) throws IOException {
this.stream = stream;
if (!stream) {
ResourceManager.beforeUdpCreate();
// only create the fd after we know we will be able to create the socket
fd = new FileDescriptor();
try {
socketCreate(false);
SocketCleanable.register(fd);
} catch (IOException ioe) {
ResourceManager.afterUdpClose();
fd = null;
throw ioe;
}
} else {
fd = new FileDescriptor();
socketCreate(true);
SocketCleanable.register(fd);
}
if (socket != null)
socket.setCreated();
if (serverSocket != null)
serverSocket.setCreated();
}
можно увидеть,socketCreate(true);, который реализован следующим образом:
@Override
void socketCreate(boolean stream) throws IOException {
if (fd == null)
throw new SocketException("Socket closed");
int newfd = socket0(stream);
fdAccess.set(fd, newfd);
}
нативным методомsocket0(stream)Получается файловый дескриптор, таким образом, создается Socket, после чего выполняется соответствующая привязка.
Обратим наше внимание обратно наsun.nio.ch.ServerSocketChannelImpl#accept(), вот новыйSocketChannelImplОбъект - это то, что делается после получения соединения. Для сервера какой сокет используется при привязке? Здесь мы используемServerSocketChannelКогда нам нужно использовать унифицированный метод, предоставленный JDK для меняopen, а также для уменьшения сложности, которую мы используем, вотjava.nio.channels.ServerSocketChannel#open:
//java.nio.channels.ServerSocketChannel#open
public static ServerSocketChannel open() throws IOException {
return SelectorProvider.provider().openServerSocketChannel();
}
//sun.nio.ch.SelectorProviderImpl#openServerSocketChannel
public ServerSocketChannel openServerSocketChannel() throws IOException {
return new ServerSocketChannelImpl(this);
}
//sun.nio.ch.ServerSocketChannelImpl#ServerSocketChannelImpl(SelectorProvider)
ServerSocketChannelImpl(SelectorProvider sp) throws IOException {
super(sp);
this.fd = Net.serverSocket(true);
this.fdVal = IOUtil.fdVal(fd);
}
//sun.nio.ch.Net#serverSocket
static FileDescriptor serverSocket(boolean stream) {
return IOUtil.newFD(socket0(isIPv6Available(), stream, true, fastLoopback));
}
Видно, что пока создается новый объект ServerSocketChannelImpl, это эквивалентно получению нового.socketЗатем связывание также имеет падение. Тем не менее, мы должны обратить внимание на деталиServerSocketChannel#openполучить этоServerSocketChannelТипы. После того, как мы примем соединение от клиента, между клиентом и сервером должен быть создан канал Socket, чтобы они могли общаться, поэтомуsun.nio.ch.ServerSocketChannelImpl#accept()что сделано вSocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);, что получаетсяSocketChannelТип объекта, таким образом, метод чтения и записи данных Socket может быть определен в этом классе.
Расширен методом сокета ServerSocketChannel
оServerSocketChannel, у нас все еще есть методы, которые нужно затронуть, такие как socket():
//sun.nio.ch.ServerSocketChannelImpl#socket
@Override
public ServerSocket socket() {
synchronized (stateLock) {
if (socket == null)
socket = ServerSocketAdaptor.create(this);
return socket;
}
}
мы виделиServerSocketAdaptor, мы знаем из аннотаций этого класса, что этоServerSocketВызов тот же, но используется нижний слойServerSocketChannelImplЧтобы реализовать класс, его адаптация должна адаптироваться к нашему использованию.ServerSocketкстати, такServerSocketAdaptorнаследоватьServerSocketИ переписать его методы по порядку, чтобы у нас появились новые возможности при написании этого кода.
InterruptibleChannel и прерываемый ввод-выводуже описано в этой статьеjava.nio.channels.spi.AbstractInterruptibleChannel#closeРеализация здесь, давайте рассмотрим некоторые детали и, кстати, перейдем к нашей новой теме:
//java.nio.channels.spi.AbstractInterruptibleChannel#close
public final void close() throws IOException {
synchronized (closeLock) {
if (closed)
return;
closed = true;
implCloseChannel();
}
}
//java.nio.channels.spi.AbstractSelectableChannel#implCloseChannel
protected final void implCloseChannel() throws IOException {
implCloseSelectableChannel();
// clone keys to avoid calling cancel when holding keyLock
SelectionKey[] copyOfKeys = null;
synchronized (keyLock) {
if (keys != null) {
copyOfKeys = keys.clone();
}
}
if (copyOfKeys != null) {
for (SelectionKey k : copyOfKeys) {
if (k != null) {
k.cancel(); // invalidate and adds key to cancelledKey set
}
}
}
}
//sun.nio.ch.ServerSocketChannelImpl#implCloseSelectableChannel
@Override
protected void implCloseSelectableChannel() throws IOException {
assert !isOpen();
boolean interrupted = false;
boolean blocking;
// set state to ST_CLOSING
synchronized (stateLock) {
assert state < ST_CLOSING;
state = ST_CLOSING;
blocking = isBlocking();
}
// wait for any outstanding accept to complete
if (blocking) {
synchronized (stateLock) {
assert state == ST_CLOSING;
long th = thread;
if (th != 0) {
//本地线程不为null，则本地Socket预先关闭
//并通知线程通知关闭
nd.preClose(fd);
NativeThread.signal(th);
// wait for accept operation to end
while (thread != 0) {
try {
stateLock.wait();
} catch (InterruptedException e) {
interrupted = true;
}
}
}
}
} else {
// non-blocking mode: wait for accept to complete
acceptLock.lock();
acceptLock.unlock();
}
// set state to ST_KILLPENDING
synchronized (stateLock) {
assert state == ST_CLOSING;
state = ST_KILLPENDING;
}
// close socket if not registered with Selector
//如果未在Selector上注册，直接kill掉
//即关闭文件描述
if (!isRegistered())
kill();
// restore interrupt status
//印证了我们上一篇中在异步打断中若是通过线程的中断方法中断线程的话
//最后要设定该线程状态是interrupt
if (interrupted)
Thread.currentThread().interrupt();
}
@Override
public void kill() throws IOException {
synchronized (stateLock) {
if (state == ST_KILLPENDING) {
state = ST_KILLED;
nd.close(fd);
}
}
}
закрыть() приложение канала
и потому, чтоclose()не вInterruptibleChannel и прерываемый ввод-выводВ этой статье подробно описано приложение.Здесь его применение более подробноSocketChannelЗдесь это больше связано с установлением соединения между клиентом и сервером для обмена данными, поэтому нормально закрывать неиспользуемый Канал после отключения.
здесь, вsun.nio.ch.ServerSocketChannelImpl#accept()В исходном коде в:
@Override
public SocketChannel accept() throws IOException {
...
// newly accepted socket is initially in blocking mode
IOUtil.configureBlocking(newfd, true);
InetSocketAddress isa = isaa[0];
SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa);
// check permitted to accept connections from the remote address
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
try {
sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
} catch (SecurityException x) {
sc.close();
throw x;
}
}
return sc;
} finally {
acceptLock.unlock();
}
}
Здесь судят о легальности удаленного адреса полученного соединения.Если в проверке есть аномалия, то созданный выше будет закрыт.SocketChannel.
Существует также практическое использование close(), когда клиент устанавливает соединение, если соединение ненормальное, созданный Socket также закрывается:
//java.nio.channels.SocketChannel#open(java.net.SocketAddress)
public static SocketChannel open(SocketAddress remote)
throws IOException
{
SocketChannel sc = open();
try {
sc.connect(remote);
} catch (Throwable x) {
try {
sc.close();
} catch (Throwable suppressed) {
x.addSuppressed(suppressed);
}
throw x;
}
assert sc.isConnected();
return sc;
}
Далее мыimplCloseSelectableChannelбудет найдено вnd.preClose(fd);а такжеnd.close(fd);, это вSocketChannelImplа такжеServerSocketChannelImplкак дляimplCloseSelectableChannelВы можете увидеть в реализации, что это такое, здесь мы беремServerSocketChannelImplНапример, в конце этого класса есть статический блок кода (SocketChannelImplАналогично), то есть будет выполняться при загрузке этого класса:
//C:/Program Files/Java/jdk-11.0.1/lib/src.zip!/java.base/sun/nio/ch/ServerSocketChannelImpl.java:550
static {
//加载nio，net资源库
IOUtil.load();
initIDs();
nd = new SocketDispatcher();
}
То есть вServerSocketChannelImplКогда этот байт-код класса загружен, он создастSocketDispatcherобъект. пройти черезSocketDispatcherПозволяет вызывать разные собственные методы на разных платформах для операций чтения и записи, а затем на основе этого класса мы можемsun.nio.ch.SocketChannelImplВыполнять операции ввода-вывода сокетов.
//sun.nio.ch.SocketDispatcher
class SocketDispatcher extends NativeDispatcher
{
static {
IOUtil.load();
}
//读操作
int read(FileDescriptor fd, long address, int len) throws IOException {
return read0(fd, address, len);
}
long readv(FileDescriptor fd, long address, int len) throws IOException {
return readv0(fd, address, len);
}
//写操作
int write(FileDescriptor fd, long address, int len) throws IOException {
return write0(fd, address, len);
}
long writev(FileDescriptor fd, long address, int len) throws IOException {
return writev0(fd, address, len);
}
//预关闭文件描述符
void preClose(FileDescriptor fd) throws IOException {
preClose0(fd);
}
//关闭文件描述
void close(FileDescriptor fd) throws IOException {
close0(fd);
}
//-- Native methods
static native int read0(FileDescriptor fd, long address, int len)
throws IOException;
static native long readv0(FileDescriptor fd, long address, int len)
throws IOException;
static native int write0(FileDescriptor fd, long address, int len)
throws IOException;
static native long writev0(FileDescriptor fd, long address, int len)
throws IOException;
static native void preClose0(FileDescriptor fd) throws IOException;
static native void close0(FileDescriptor fd) throws IOException;
}
FileDescriptor
мы виделиFileDescriptorВ предыдущем коде много вхождений, здесь мы представим их специально. Экземпляры класса FileDescriptor действуют как непрозрачная обработка базовых машинно-специфичных структур, представляющих открытые файлы, открытые сокеты или другие источники или приемники байтов.
Основное использование файлового дескриптора — создание FileInputStream или FileOutputStream для его содержания.
Примечание. Приложения не должны создавать свои собственные файловые дескрипторы.
Давайте посмотрим на часть его исходного кода:
public final class FileDescriptor {
private int fd;
private long handle;
private Closeable parent;
private List<Closeable> otherParents;
private boolean closed;
/**
* true, if file is opened for appending.
*/
private boolean append;
static {
initIDs();
}
/**
* 在未明确关闭FileDescriptor的情况下进行清理.
*/
private PhantomCleanable<FileDescriptor> cleanup;
/**
* 构造一个无效的FileDescriptor对象，fd或handle会在之后进行设定
*/
public FileDescriptor() {
fd = -1;
handle = -1;
}
/**
* Used for standard input, output, and error only.
* For Windows the corresponding handle is initialized.
* For Unix the append mode is cached.
* 仅用于标准输入，输出和错误。
* 对于Windows，初始化相应的句柄。
* 对于Unix，缓存附加模式。
* @param fd the raw fd number (0, 1, 2)
*/
private FileDescriptor(int fd) {
this.fd = fd;
this.handle = getHandle(fd);
this.append = getAppend(fd);
}
...
}
Дескрипторы стандартных потоков ввода, вывода и ошибок, которые мы обычно используем, могут быть следующими: обычно мы не используем их напрямую, а используемjava.lang.System.in,java.lang.System#out,java.lang.System#err:
public static final FileDescriptor in = new FileDescriptor(0);
public static final FileDescriptor out = new FileDescriptor(1);
public static final FileDescriptor err = new FileDescriptor(2);
Чтобы проверить, является ли дескриптор файла допустимым, используйте следующие методы:
//java.io.FileDescriptor#valid
public boolean valid() {
return (handle != -1) || (fd != -1);
}
Если возвращаемое значение истинно, то объект файлового дескриптора, представленныйsocket 文件操作или другие активные сетевые подключения допустимы, в противном случае false недопустимо.
Для получения дополнительной информации читатели могут самостоятельно углубиться в исходный код, поэтому я не буду здесь слишком много объяснять. Чтобы дать вам лучшее понимание вышеуказанного содержания, мы рассмотрим его подробнее в следующих разделах.
Интерпретация SocketChannel в пакете NIO
Ранее мы коснулисьSocketChannel, здесь, чтобы связаться с деталями.
Точно так же мы можем также назвать этот классopenметод созданияsocket channel. Обратите внимание:
- нельзя использовать для каких-либо ранее существовавших
socketСоздайтеchannel. - Вновь созданный
socket channelОткрыт, но еще не подключен. - попробуй без подключения
channelзвонитьI/Oоперация приведет к броскуNotYetConnectedException. - позвонив
connectметод подключенияsocket channel; - После подключения
socket channelбудет оставаться подключенным до тех пор, пока он не будет закрыт. - Есть ли связь
socket channelЕго можно вызвать, определивisConnectedметод.
socket channelПоддержка неблокирующих соединений:
- можно сначала создать
socket channel, к которому затем можно получить доступ черезconnectметод сборки на удаленныйsocketСвязь. - позвонив
finishConnectметод завершения соединения. - Чтобы определить, выполняется ли операция подключения, вы можете вызвать
isConnectionPendingметод определить.
socket channelПоддерживается асинхронное отключение, аналогичноChannelАсинхронная операция закрытия в классе.
- если
socketВход закрыт одним потоком, пока другой поток находится здесьsocket channelзаблокирован из-за операции чтения, то операция чтения в заблокированном потоке не будет читать никаких байтов и вернет-1. - если
socketВыход закрывается одним потоком, в то время как другой поток находится вsocket channelзаблокирован из-за операции записи, заблокированный поток получитAsynchronousCloseException.
Далее, давайте посмотрим на его конкретный метод реализации.
Open() для ServerSocketChannel и SocketChannel
//java.nio.channels.SocketChannel#open()
public static SocketChannel open() throws IOException {
return SelectorProvider.provider().openSocketChannel();
}
//java.nio.channels.SocketChannel#open(java.net.SocketAddress)
//这个方法省的我们再次调用connect了
public static SocketChannel open(SocketAddress remote)
throws IOException
{
//默认是堵塞的，这个在AbstractSelectableChannel处讨论过了
SocketChannel sc = open();
try {
sc.connect(remote);
} catch (Throwable x) {
try {
sc.close();
} catch (Throwable suppressed) {
x.addSuppressed(suppressed);
}
throw x;
}
assert sc.isConnected();
return sc;
}
//sun.nio.ch.SelectorProviderImpl#openSocketChannel
public SocketChannel openSocketChannel() throws IOException {
return new SocketChannelImpl(this);
}
//sun.nio.ch.SocketChannelImpl#SocketChannelImpl(java.nio.channels.spi.SelectorProvider)
SocketChannelImpl(SelectorProvider sp) throws IOException {
super(sp);
//调用socket函数，true表示TCP
this.fd = Net.socket(true);
this.fdVal = IOUtil.fdVal(fd);
}
//sun.nio.ch.Net#socket(boolean)
static FileDescriptor socket(boolean stream) throws IOException {
return socket(UNSPEC, stream);
}
//sun.nio.ch.Net#socket(java.net.ProtocolFamily, boolean)
static FileDescriptor socket(ProtocolFamily family, boolean stream)
throws IOException {
boolean preferIPv6 = isIPv6Available() &&
(family != StandardProtocolFamily.INET);
return IOUtil.newFD(socket0(preferIPv6, stream, false, fastLoopback));
}
//sun.nio.ch.IOUtil#newFD
public static FileDescriptor newFD(int i) {
FileDescriptor fd = new FileDescriptor();
setfdVal(fd, i);
return fd;
}
static native void setfdVal(FileDescriptor fd, int value);
оNet.socket(true), мы упоминали ранее, здесь, с помощью базового исходного кода, чтобы снова настроить его (вы можете пропустить его, если не хотите видеть его здесь):
JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_socket0(JNIEnv *env, jclass cl, jboolean preferIPv6,
jboolean stream, jboolean reuse, jboolean ignored)
{
int fd;
//字节流还是数据报,TCP对应SOCK_STREAM,UDP对应SOCK_DGRAM,此处传入的stream=true;
int type = (stream ? SOCK_STREAM : SOCK_DGRAM);
//判断是IPV6还是IPV4
int domain = (ipv6_available() && preferIPv6) ? AF_INET6 : AF_INET;
//调用Linux的socket函数,domain为代表协议;
//type为套接字类型，protocol设置为0来表示使用默认的传输协议
fd = socket(domain, type, 0);
//出错
if (fd < 0) {
return handleSocketError(env, errno);
}
/* Disable IPV6_V6ONLY to ensure dual-socket support */
if (domain == AF_INET6) {
int arg = 0;
//arg=1设置ipv6的socket只接收ipv6地址的报文,arg=0表示也可接受ipv4的请求
if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (char*)&arg,
sizeof(int)) < 0) {
JNU_ThrowByNameWithLastError(env,
JNU_JAVANETPKG "SocketException",
"Unable to set IPV6_V6ONLY");
close(fd);
return -1;
}
}
//SO_REUSEADDR有四种用途:
//1.当有一个有相同本地地址和端口的socket1处于TIME_WAIT状态时，而你启动的程序的socket2要占用该地址和端口，你的程序就要用到该选项。
//2.SO_REUSEADDR允许同一port上启动同一服务器的多个实例(多个进程)。但每个实例绑定的IP地址是不能相同的。
//3.SO_REUSEADDR允许单个进程绑定相同的端口到多个socket上，但每个socket绑定的ip地址不同。
//4.SO_REUSEADDR允许完全相同的地址和端口的重复绑定。但这只用于UDP的多播，不用于TCP;
if (reuse) {
int arg = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&arg,
sizeof(arg)) < 0) {
JNU_ThrowByNameWithLastError(env,
JNU_JAVANETPKG "SocketException",
"Unable to set SO_REUSEADDR");
close(fd);
return -1;
}
}
#if defined(__linux__)
if (type == SOCK_DGRAM) {
int arg = 0;
int level = (domain == AF_INET6) ? IPPROTO_IPV6 : IPPROTO_IP;
if ((setsockopt(fd, level, IP_MULTICAST_ALL, (char*)&arg, sizeof(arg)) < 0) &&
(errno != ENOPROTOOPT)) {
JNU_ThrowByNameWithLastError(env,
JNU_JAVANETPKG "SocketException",
"Unable to set IP_MULTICAST_ALL");
close(fd);
return -1;
}
}
//IPV6_MULTICAST_HOPS用于控制多播的范围，
// 1表示只在本地网络转发,
//更多介绍请参考(http://www.ctt.sbras.ru/cgi-bin/www/unix_help/unix-man?ip6+4);
/* By default, Linux uses the route default */
if (domain == AF_INET6 && type == SOCK_DGRAM) {
int arg = 1;
if (setsockopt(fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &arg,
sizeof(arg)) < 0) {
JNU_ThrowByNameWithLastError(env,
JNU_JAVANETPKG "SocketException",
"Unable to set IPV6_MULTICAST_HOPS");
close(fd);
return -1;
}
}
#endif
return fd;
}
Добавлено после Linux 3.9SO_REUSEPORTКонфигурация, эта конфигурация очень мощная, множественнаяsocket(Независимо от того, прослушивает он или не прослушивает, является ли это TCP или UDP), если он установлен перед привязкойSO_REUSEPORTсвойство, то вы можете выполнить привязку к точно такому же адресу и порту.
Для предотвращения «захвата порта» (Port hijacking) имеет специальное ограничение: все сокеты, желающие использовать исходный адрес и порт, должны иметь один и тот же эффективный идентификатор пользователя (effective user ID). Таким образом, один пользователь не может «украсть» порты у другого пользователя. Кроме того, ядро обрабатываетSO_REUSEPORT socketПри использовании «особых приемов», не используемых в других системах:
- Для сокетов UDP ядро пытается одинаково пересылать дейтаграммы;
- Для прослушивающих сокетов TCP ядро пытается равномерно распределить новые клиентские запросы на подключение (возвращенные accept) на сокеты, которые имеют один и тот же адрес и порт (сокеты прослушивания сервера).
Например: несколько экземпляров простой серверной программы могут использоватьSO_REUSEPORT socket, который реализует простую балансировку нагрузки, потому что ядро уже сделало распределение запросов.
Как видно из предыдущего кода, в этомsocketПосле успешного создания вызовитеIOUtil.newFDфайловый дескриптор создан
. Здесь я просто хочу знать, может ли этот сокет быть введен или может быть прочитан, или что-то не так, пожалуйста, обратитесь кFileDescriptorНастройки последних нескольких стандартных состояний в этом разделе на самом деле здесь такие же, потому что мы хотим писать и читать в сокет, а стандартные состояния — это не что иное, как эти три: ввод, вывод и ошибка. И этот сокет привязан кSocketChannelна, то ставьFileDescriptorЕго также можно связать с указанным выше, чтобы мы могли получить его состояние. Поскольку FileDescriptor не предоставляет метода для внешней установки fd, setfdVal реализуется через собственный метод:
JNIEXPORT void JNICALL
Java_sun_nio_ch_IOUtil_setfdVal(JNIEnv *env, jclass clazz, jobject fdo, jint val)
{
(*env)->SetIntField(env, fdo, fd_fdID, val);
}
Если у вас есть понимание программирования оболочки или команд под Linux, мы знаем, что оболочка перенаправляет ошибки с помощью 2>, то есть сообщение об ошибке выводится из канала, на который указывает № 2, где 0 и 1 также указывают на канал. Это также представляет состояние, так что можно управлять состоянием, представляющим Socket, то есть изменятьSelectionKeyизinterest ops, то есть сначалаSelectionKeyКлассифицируется по типу ввода и вывода, и тогда наши операции чтения и записи статуса будут улажены. Здесь мы попали в точку, и мы подробно объясним это в следующей статье.
мы возвращаемся кSocketChannelизopenметод. Мы видим, что,SelectorProvider.provider().openSocketChannel()то, что возвращаетсяSocketChannelImplэкземпляр объекта. существуетSocketChannelImpl(SelectorProvider sp), мы его не видимthis.stateВыполнить операцию со значением, т. е. по умолчанию 0, т. е.ST_UNCONNECTED(неподключенное состояние), а Socket по умолчанию заблокирован.
Поэтому в целом при использовании асинхронного метода чаще используется открытый метод без параметров, таким образом, мы будем вызыватьconfigureBlockingустановить неблокирующий.
Интерпретация соединения SocketChannel
Как видно спереди, мы называемconnectМетод подключается к удаленному серверу, и его исходный код выглядит следующим образом:
//sun.nio.ch.SocketChannelImpl#connect
@Override
public boolean connect(SocketAddress sa) throws IOException {
InetSocketAddress isa = Net.checkAddress(sa);
SecurityManager sm = System.getSecurityManager();
if (sm != null)
sm.checkConnect(isa.getAddress().getHostAddress(), isa.getPort());
InetAddress ia = isa.getAddress();
if (ia.isAnyLocalAddress())
ia = InetAddress.getLocalHost();
try {
readLock.lock();
try {
writeLock.lock();
try {
int n = 0;
boolean blocking = isBlocking();
try {
//支持线程中断，通过设置当前线程的Interruptible blocker属性实现
beginConnect(blocking, isa);
do {
//调用connect函数实现，如果采用堵塞模式，会一直等待，直到成功或出//现异常
n = Net.connect(fd, ia, isa.getPort());
} while (n == IOStatus.INTERRUPTED && isOpen());
} finally {
endConnect(blocking, (n > 0));
}
assert IOStatus.check(n);
//连接成功
return n > 0;
} finally {
writeLock.unlock();
}
} finally {
readLock.unlock();
}
} catch (IOException ioe) {
// connect failed, close the channel
close();
throw SocketExceptions.of(ioe, isa);
}
}
оbeginConnectа такжеendConnect, направлен наAbstractInterruptibleChannelсерединаbegin()а такжеendУсовершенствование метода. Что нам нужно знать здесь, так это то, что если это неблокирующий канал, нам не нужно заботиться о прерывании процесса соединения. Как следует из названия, необходимо учитывать только ожидание блокировки, чтобы прервать появление этой сцены. Остальные детали полностью прокомментированы в коде, и читатели могут проверить их самостоятельно.
//sun.nio.ch.SocketChannelImpl#beginConnect
private void beginConnect(boolean blocking, InetSocketAddress isa)
throws IOException
{ //只有阻塞的时候才会进入begin
if (blocking) {
// set hook for Thread.interrupt
//支持线程中断，通过设置当前线程的Interruptible blocker属性实现
begin();
}
synchronized (stateLock) {
//默认为open, 除非调用了close方法
ensureOpen();
//检查连接状态
int state = this.state;
if (state == ST_CONNECTED)
throw new AlreadyConnectedException();
if (state == ST_CONNECTIONPENDING)
throw new ConnectionPendingException();
//断言当前的状态是否是未连接状态，如果是，赋值表示正在连接中
assert state == ST_UNCONNECTED;
//表示正在连接中
this.state = ST_CONNECTIONPENDING;
//只有未绑定本地地址也就是说未调用bind方法才执行,
//该方法在ServerSocketChannel中也见过
if (localAddress == null)
NetHooks.beforeTcpConnect(fd, isa.getAddress(), isa.getPort());
remoteAddress = isa;
if (blocking) {
// record thread so it can be signalled if needed
readerThread = NativeThread.current();
}
}
}
В процессе подключения нам нужно обратить внимание на статус нескольких подключений:ST_UNCONNECTED,ST_CONNECTED,ST_CONNECTIONPENDING,ST_CLOSING,ST_KILLPENDING,ST_KILLED, но также, поскольку это общедоступное состояние, его может соединять несколько потоков. так,stateопределяется какvolatileпеременная, эта переменная должна иметьstateLockэтот объект какsynchronizedБлокируйте объекты для управления синхронными операциями.
//sun.nio.ch.SocketChannelImpl#endConnect
private void endConnect(boolean blocking, boolean completed)
throws IOException
{
endRead(blocking, completed);
//当上面代码中n>0，说明连接成功，更新状态为ST_CONNECTED
if (completed) {
synchronized (stateLock) {
if (state == ST_CONNECTIONPENDING) {
localAddress = Net.localAddress(fd);
state = ST_CONNECTED;
}
}
}
}
//sun.nio.ch.SocketChannelImpl#endRead
private void endRead(boolean blocking, boolean completed)
throws AsynchronousCloseException
{ //当阻塞状态下的话，才进入
if (blocking) {
synchronized (stateLock) {
readerThread = 0;
// notify any thread waiting in implCloseSelectableChannel
if (state == ST_CLOSING) {
stateLock.notifyAll();
}
}
//和begin成对出现，当线程中断时，抛出ClosedByInterruptException
// remove hook for Thread.interrupt
end(completed);
}
}
давайте следоватьconnectсерединаNet.connect(fd, ia, isa.getPort())метод:
//sun.nio.ch.Net#connect
static int connect(FileDescriptor fd, InetAddress remote, int remotePort)
throws IOException
{
return connect(UNSPEC, fd, remote, remotePort);
}
//sun.nio.ch.Net#connect
static int connect(ProtocolFamily family, FileDescriptor fd, InetAddress remote, int remotePort)
throws IOException
{
boolean preferIPv6 = isIPv6Available() &&
(family != StandardProtocolFamily.INET);
return connect0(preferIPv6, fd, remote, remotePort);
}
Этот метод в конечном итоге вызовет собственный метод, который аннотируется следующим образом:
JNIEXPORT jint JNICALL
Java_sun_nio_ch_Net_connect0(JNIEnv *env, jclass clazz, jboolean preferIPv6,
jobject fdo, jobject iao, jint port)
{
SOCKETADDRESS sa;
int sa_len = 0;
int rv;
//地址转换为struct sockaddr格式
if (NET_InetAddressToSockaddr(env, iao, port, &sa, &sa_len, preferIPv6) != 0) {
return IOS_THROWN;
}
//传入fd和sockaddr,与远程服务器建立连接，一般就是TCP三次握手
//如果设置了configureBlocking(false),不会堵塞，否则会堵塞一直到超时或出现异常
rv = connect(fdval(env, fdo), &sa.sa, sa_len);
//0表示连接成功，失败时通过errno获取具体原因
if (rv != 0) {
//非堵塞，连接还未建立(-2)
if (errno == EINPROGRESS) {
return IOS_UNAVAILABLE;
} else if (errno == EINTR) {
//中断(-3)
return IOS_INTERRUPTED;
}
return handleSocketError(env, errno);
}
//连接建立,一般TCP连接连接都需要时间，因此除非是本地网络，
//一般情况下非堵塞模式返回IOS_UNAVAILABLE比较多；
return 1;
}
Как видно из комментариев выше, если он неблокирующий и соединение не установлено сразу, возвращается -2, то есть соединение не установлено успешно.beginConnectЧасть исходного кода можно увидеть, на данный момент статусST_CONNECTIONPENDING, то в неблокирующих условиях, когда он станетST_CONNECTEDЕсть ли способ запросить статус или дождаться завершения соединения?
Тогда давайте обратим вниманиеsun.nio.ch.SocketChannelImpl#finishConnect
Интерпретация finishConnect в SocketChannelImpl
Во-первых, давайте рассмотрим, что мы рассмотрели ранееsun.nio.ch.ServerSocketAdaptorИспользование удобно для людей, которые только привыкли к программированию Socket.Здесь мы также можем увидеть основную логику реализации ядра, затем естьServerSocketAdaptorимеютSocketAdaptor, здесь он, наконец, вызывается в программировании сокетов BIOconnect(address)работать:
//java.net.Socket#Socket
private Socket(SocketAddress address, SocketAddress localAddr,
boolean stream) throws IOException {
setImpl();
// backward compatibility
if (address == null)
throw new NullPointerException();
try {
createImpl(stream);
if (localAddr != null)
bind(localAddr);
connect(address);
} catch (IOException | IllegalArgumentException | SecurityException e) {
try {
close();
} catch (IOException ce) {
e.addSuppressed(ce);
}
throw e;
}
}
Здесь мы можем позвонитьjava.nio.channels.SocketChannel#open(), а затем вызвать полученныйSocketChannelобъектsocket()метод, вы можете получитьsun.nio.ch.SocketAdaptorэкземпляр объекта. Давайте проверимSocketAdaptorРеализация подключения:
//sun.nio.ch.SocketAdaptor#connect
public void connect(SocketAddress remote) throws IOException {
connect(remote, 0);
}
public void connect(SocketAddress remote, int timeout) throws IOException {
if (remote == null)
throw new IllegalArgumentException("connect: The address can't be null");
if (timeout < 0)
throw new IllegalArgumentException("connect: timeout can't be negative");
synchronized (sc.blockingLock()) {
if (!sc.isBlocking())
throw new IllegalBlockingModeException();
try {
//未设定超时则会一直在此等待直到连接或者出现异常
// no timeout
if (timeout == 0) {
sc.connect(remote);
return;
}
//有超时设定，则会将Socket给设定为非阻塞
// timed connect
sc.configureBlocking(false);
try {
if (sc.connect(remote))
return;
} finally {
try {
sc.configureBlocking(true);
} catch (ClosedChannelException e) { }
}
long timeoutNanos = NANOSECONDS.convert(timeout, MILLISECONDS);
long to = timeout;
for (;;) {
//通过计算超时时间，在允许的时间范围内无限循环来进行连接，
//如果超时，则关闭这个Socket
long startTime = System.nanoTime();
if (sc.pollConnected(to)) {
boolean connected = sc.finishConnect();
//看下文解释
assert connected;
break;
}
timeoutNanos -= System.nanoTime() - startTime;
if (timeoutNanos <= 0) {
try {
sc.close();
} catch (IOException x) { }
throw new SocketTimeoutException();
}
to = MILLISECONDS.convert(timeoutNanos, NANOSECONDS);
}
} catch (Exception x) {
Net.translateException(x, true);
}
}
}
Вот следующее небольшое примечание: в JavaassertКлючевые слова изJAVA SE 1.4вводится во избежание иСтарая версия кода Javaиспользуется вassertключевые слова вызывают ошибки,Java не включает проверку утверждений по умолчанию (в настоящее время все утверждения утверждений будут игнорироваться!). Если вы хотите включить проверку утверждений, вам нужно использовать переключатель -enableassertions или -ea, чтобы включить его.Благодаря приведенным выше комментариям к исходному коду я считаю, что все уже знают общий процесс.sun.nio.ch.SocketChannelImpl#finishConnectЧто именно он сделал, вот, давайте рассмотрим это:
//sun.nio.ch.SocketChannelImpl#finishConnect
@Override
public boolean finishConnect() throws IOException {
try {
readLock.lock();
try {
writeLock.lock();
try {
// no-op if already connected
if (isConnected())
return true;
boolean blocking = isBlocking();
boolean connected = false;
try {
beginFinishConnect(blocking);
int n = 0;
if (blocking) {
do {
//阻塞情况下，第二个参数传入true
n = checkConnect(fd, true);
} while ((n == 0 || n == IOStatus.INTERRUPTED) && isOpen());
} else {
//非阻塞情况下，第二个参数传入false
n = checkConnect(fd, false);
}
connected = (n > 0);
} finally {
endFinishConnect(blocking, connected);
}
assert (blocking && connected) ^ !blocking;
return connected;
} finally {
writeLock.unlock();
}
} finally {
readLock.unlock();
}
} catch (IOException ioe) {
// connect failed, close the channel
close();
throw SocketExceptions.of(ioe, remoteAddress);
}
}
//sun.nio.ch.SocketChannelImpl#checkConnect
private static native int checkConnect(FileDescriptor fd, boolean block)
throws IOException;
оbeginFinishConnectа такжеendFinishConnectи наш предыдущий анализsun.nio.ch.SocketChannelImpl#beginConnectа такжеsun.nio.ch.SocketChannelImpl#endConnectПроцесс почти такой же, читатели, которые не понимают, могут оглянуться назад. Остальное — это основная основная логика, на которой мы сосредоточимся.checkConnect(fd, true), это также локальный метод, и задействованный исходный код выглядит следующим образом:
JNIEXPORT jint JNICALL
Java_sun_nio_ch_SocketChannelImpl_checkConnect(JNIEnv *env, jobject this,
jobject fdo, jboolean block)
{
int error = 0;
socklen_t n = sizeof(int);
//获取FileDescriptor中的fd
jint fd = fdval(env, fdo);
int result = 0;
struct pollfd poller;
//文件描述符
poller.fd = fd;
//请求的事件为写事件
poller.events = POLLOUT;
//返回的事件
poller.revents = 0;
//-1表示阻塞,0表示立即返回，不阻塞进程
result = poll(&poller, 1, block ? -1 : 0);
//小于0表示调用失败
if (result < 0) {
if (errno == EINTR) {
return IOS_INTERRUPTED;
} else {
JNU_ThrowIOExceptionWithLastError(env, "poll failed");
return IOS_THROWN;
}
}
//非堵塞时，0表示没有准备好的连接
if (!block && (result == 0))
return IOS_UNAVAILABLE;
//准备好写或出现错误的socket数量>0
if (result > 0) {
errno = 0;
result = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &n);
//出错
if (result < 0) {
return handleSocketError(env, errno);
//发生错误，处理错误
} else if (error) {
return handleSocketError(env, error);
} else if ((poller.revents & POLLHUP) != 0) {
return handleSocketError(env, ENOTCONN);
}
//socket已经准备好，可写，即连接已经建立好
// connected
return 1;
}
return 0;
}
Конкретный процесс показан в комментариях к исходному коду, блокировать ли нас в исходном коде локального метода и перед ним.sun.nio.ch.SocketChannelImpl#finishConnectповедение соответствует. Кроме того, как видно из исходного кода выше, нижний слойpollЗапросsocketстатус, чтобы определить, успешно ли установлено соединение, потому что в неблокирующем режимеfinishConnectметод вернется немедленно, согласно здесьsun.nio.ch.SocketAdaptor#connectобработки, он использует цикл, чтобы определить, установлено ли соединение.В нашем программировании nio это не рекомендуется, это относится к полуфабрикатам, но рекомендуется прописать вSelector,пройти черезops=OP_CONNECTзавершить соединениеSelectionKey, затем позвонитеfinishConnectЗавершите установление соединения;
ТакfinishConnectМожно ли не звонить? Ответа нет, потому что толькоfinishConnectобновит статус доST_CONNECTED, во время вызоваreadа такжеwriteстатус будет оцениваться.
Здесь мы представляем то, что мы собираемся покрытьSelectorа такжеSelectionKey, мы подробно объясним в следующей статье.