Java 类java.nio.channels.ServerSocketChannel 实例源码

protected void
bind(
    ServerSocketChannel ssc,
    InetAddress         address,
    int                 port )

    throws IOException
{
    if ( address == null ){

        ssc.socket().bind( new InetSocketAddress( port ), 1024 );

    }else{

        ssc.socket().bind( new InetSocketAddress( address, port ), 1024 );
    }
}

private synchronized void OpenSockets() {
    if (mode == CLIENT) {
        try
        {
            attemptClientConnection();
            receiving = true;
        } catch (IOException e) {
            log.error(e);
        }
    } else if (mode == SERVER) {
        try {

            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            receiving = true;
        } catch (IOException ex) {
            selector = null;
        }

    }
}

private final void acceptConnection(final SelectionKey key, MMOConnection<T> con)
{
    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
    SocketChannel sc;

    try
    {
        while ((sc = ssc.accept()) != null)
        {
            if (_acceptFilter == null || _acceptFilter.accept(sc))
            {
                sc.configureBlocking(false);
                SelectionKey clientKey = sc.register(_selector, SelectionKey.OP_READ);
                con = new MMOConnection<T>(this, sc.socket(), clientKey);
                con.setClient(_clientFactory.create(con));
                clientKey.attach(con);
            }
            else
                sc.socket().close();
        }
    }
    catch (IOException e)
    {
        e.printStackTrace();
    }
}

/**
 * Creates a NIO ServerSocketChannel, and gets the ServerSocket from
 * there. Then binds the obtained socket.
 * This fails on Windows with Oracle JDK1.6.0u33, if the passed InetAddress is a
 * IPv6 address. Works on Oracle JDK 1.7.
 */
private void bindNIOServerSocket(InetAddress inetAddr) throws IOException {
  while (true) {
    int port = HBaseTestingUtility.randomFreePort();
    InetSocketAddress addr = new InetSocketAddress(inetAddr, port);
    ServerSocketChannel channel = null;
    ServerSocket serverSocket = null;
    try {
      channel = ServerSocketChannel.open();
      serverSocket = channel.socket();
      serverSocket.bind(addr); // This does not work
      break;
    } catch (BindException ex) {
      //continue
    } finally {
      if (serverSocket != null) {
        serverSocket.close();
      }
      if (channel != null) {
        channel.close();
      }
    }  
  }
}

/**
 * {@inheritDoc}
 */
@Override
protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {

    SelectionKey key = handle.keyFor(selector);

    if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {
        return null;
    }

    // accept the connection from the client
    SocketChannel ch = handle.accept();

    if (ch == null) {
        return null;
    }

    return new NioSocketSession(this, processor, ch);
}

@Override
public void configure(InetSocketAddress addr, int maxcc) throws IOException {
    configureSaslLogin();
    /**
     * thread是ServerCnxnFactory的主线程先启动；然后再启动NIO服务器
     * @see ServerCnxnFactory#run也已经被执行了
     * 此时客户端可以访问zk的2181端口；但是还无法处理客户端的请求
     */
    thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr);
    //线程后台运行
    thread.setDaemon(true);
    maxClientCnxns = maxcc;
    //??
    this.ss = ServerSocketChannel.open();
    //??
    ss.socket().setReuseAddress(true);
    LOG.info("binding to port " + addr);
    ss.socket().bind(addr);
    //非阻塞
    ss.configureBlocking(false);
    //??
    ss.register(selector, SelectionKey.OP_ACCEPT);
}

public TunnelServer(int port, Selector selector) throws IOException {
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.configureBlocking(false);
    // ServerSocketChannel.bind() requires API 24
    serverSocketChannel.socket().bind(new InetSocketAddress(Inet4Address.getLoopbackAddress(), port));

    SelectionHandler socketChannelHandler = (selectionKey) -> {
        try {
            ServerSocketChannel channel = (ServerSocketChannel) selectionKey.channel();
            acceptClient(selector, channel);
        } catch (IOException e) {
            Log.e(TAG, "Cannot accept client", e);
        }
    };
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, socketChannelHandler);
}

public void
 selectFailure(
VirtualAbstractSelectorListener     listener,
AbstractSelectableChannel           sc,
Object                              attachment,
Throwable                           msg)
 {
  if ( op == OP_ACCEPT ){

      ((VirtualAcceptSelectorListener)listener).selectFailure( VirtualChannelSelector.this, (ServerSocketChannel)sc, attachment, msg );
  }else{

      ((VirtualSelectorListener)listener).selectFailure( VirtualChannelSelector.this, (SocketChannel)sc, attachment, msg );

  }
 }

public static void main(String args[]) throws Exception {
    InetAddress iaddr = InetAddress.getLocalHost();

    try ( ServerSocket ss = new ServerSocket(0);
          Socket s1 = new Socket(iaddr, ss.getLocalPort());
          Socket s2 = ss.accept() ) {

        test(s1, s2, "Testing NET");
    }

    // check the NIO socket adapter
    try (ServerSocketChannel sc = ServerSocketChannel.open().bind(null);
         SocketChannel s1 = SocketChannel.open(
                 new InetSocketAddress(iaddr, sc.socket().getLocalPort()));
         SocketChannel s2 = sc.accept() ) {

        test(s1.socket(), s2.socket(), "Testing NIO");
    }

    if (failed) {
        throw new RuntimeException("Failed: check output");
    }
}

public void run() {
    try {
        java.nio.channels.Selector acceptSelector = java.nio.channels.Selector.open();
        serverSocketChannel.register(acceptSelector, SelectionKey.OP_ACCEPT);
        while (serverSocketChannel.isOpen()) {
            if (acceptSelector.select(1000) > 0) {
                Iterator<SelectionKey> it = acceptSelector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    if (key.isAcceptable()) {
                        SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
                        socketChannel.configureBlocking(false);
                        newChannels.add(socketChannel);
                        selector.wakeup();
                    }
                    it.remove();
                }
            }
        }
    } catch (IOException e) {
        // ignore
    }
}

public final void openServerSocket(InetAddress address, int tcpPort) throws IOException
{
    final ServerSocketChannel selectable = ServerSocketChannel.open();
    selectable.configureBlocking(false);

    final ServerSocket ss = selectable.socket();

    if (address != null)
    {
        ss.bind(new InetSocketAddress(address, tcpPort));
    }
    else
    {
        ss.bind(new InetSocketAddress(tcpPort));
    }

    selectable.register(_selector, SelectionKey.OP_ACCEPT);
}

public static void main(String[] args) throws Exception {

        boolean keepAlive = false;
        String prop = System.getProperty("com.sun.CORBA.transport.enableTcpKeepAlive");
        if (prop != null)
            keepAlive = !"false".equalsIgnoreCase(prop);

        DefaultSocketFactoryImpl sfImpl = new DefaultSocketFactoryImpl();
        ORBImpl orb = new ORBImpl();
        orb.set_parameters(null);
        sfImpl.setORB(orb);

        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(0));

        InetSocketAddress isa = new InetSocketAddress("localhost", ssc.socket().getLocalPort());
        Socket s = sfImpl.createSocket("ignore", isa);
        System.out.println("Received factory socket" + s);
        if (keepAlive != s.getKeepAlive())
            throw new RuntimeException("KeepAlive value not honoured in CORBA socket");
    }

private void openSocketForTransferPort(int port) throws IOException {

        executor = Utils.getStandardExecService("[ Acceptable ServersThreadPool ] ",
                2,
                10,
                new ArrayBlockingQueue<Runnable>(65500),
                Thread.NORM_PRIORITY - 2);
        ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        FDTSession sess = FDTSessionManager.getInstance().getSession(sessionID);
        ss = ssc.socket();
        String listenIP = config.getListenAddress();
        if (listenIP == null) {
            ss.bind(new InetSocketAddress(port));
        }
        else
        {
            ss.bind(new InetSocketAddress(InetAddress.getByName(listenIP), port));
        }

        sel = Selector.open();
        ssc.register(sel, SelectionKey.OP_ACCEPT);
        sc = ssc.accept();
        config.setSessionSocket(ssc, ss, sc, s, port);
    }

public static void main(String[] args) throws Exception {
//            创建ServerSocketChannel，监听8080端口
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(8080));
//            设置为非阻塞模式
        ssc.configureBlocking(false);
//            为ssc注册选择器
        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
//            创建处理器
        Handler handler = new Handler(1024);
        while (true) {
//            等待请求，每次等待阻塞3s，超过3s后线程继续向下运行，如果传入0或者不传参数将一直阻塞
            if (selector.select(3000) == 0) {
                System.out.println("等待请求超时……");
                continue;
            }
            System.out.println("处理请求……");
//            获取待处理的SelectionKey
            Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
            while (keyIter.hasNext()) {
                SelectionKey key = keyIter.next();
                try {
                    // 接收到连接请求时
                    if (key.isAcceptable()) {
                        handler.handleAccept(key);
                    }
//            读数据
                    if (key.isReadable()) {
                        handler.handleRead(key);
                    }
                } catch (IOException ex) {
                    keyIter.remove();
                    continue;
                }
//            处理完后，从待处理的SelectionKey迭代器中移除当前所使用的key
                keyIter.remove();
            }
        }
    }

public static void main(String[] args) throws Exception {
//    创建ServerSocketChannel，监听8080端口
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(8080));
//    设置为非阻塞模式
        ssc.configureBlocking(false);
//    为ssc注册选择器
        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
//    创建处理器
        while (true) {
//    等待请求，每次等待阻塞3s，超过3s后线程继续向下运行，
//    如果传入0或者不传参数将一直阻塞
            if (selector.select(3000) == 0) {
                continue;
            }
//    获取待处理的SelectionKey
            Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
            while (keyIter.hasNext()) {
                SelectionKey key = keyIter.next();
//    启动新线程处理SelectionKey
                new Thread(new HttpHandler(key)).run();
//    处理完后，从待处理的SelectionKey迭代器中移除当前所使用的key
                keyIter.remove();
            }
        }
    }

public ConnectionAcceptor accept(Action<ConnectCompletion> action, boolean allowRemote) {
    final ServerSocketChannel serverSocket;
    int localPort;
    try {
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(addressFactory.getLocalBindingAddress(), 0));
        localPort = serverSocket.socket().getLocalPort();
    } catch (Exception e) {
        throw UncheckedException.throwAsUncheckedException(e);
    }

    UUID id = idGenerator.generateId();
    List<InetAddress> addresses = addressFactory.getCommunicationAddresses();
    final Address address = new MultiChoiceAddress(id, localPort, addresses);
    LOGGER.debug("Listening on {}.", address);

    final StoppableExecutor executor = executorFactory.create("Incoming " + (allowRemote ? "remote" : "local")+ " TCP Connector on port " + localPort);
    executor.execute(new Receiver(serverSocket, action, allowRemote));

    return new ConnectionAcceptor() {
        public Address getAddress() {
            return address;
        }

        public void requestStop() {
            CompositeStoppable.stoppable(serverSocket).stop();
        }

        public void stop() {
            requestStop();
            executor.stop();
        }
    };
}

public TcpProxyServer(int port) throws IOException {
    m_Selector = Selector.open();
    m_ServerSocketChannel = ServerSocketChannel.open();
    m_ServerSocketChannel.configureBlocking(false);
    m_ServerSocketChannel.socket().bind(new InetSocketAddress(port));
    m_ServerSocketChannel.register(m_Selector, SelectionKey.OP_ACCEPT);
    this.Port = (short) m_ServerSocketChannel.socket().getLocalPort();
    System.out.printf("AsyncTcpServer listen on %d success.\n", this.Port & 0xFFFF);
}

public void restart() throws IOException {
    server.close();
    selector.selectNow();
    server = ServerSocketChannel.open();
    server.configureBlocking(false);
    server.socket().setReuseAddress(true);
    server.socket().bind(srcAddr);
    server.register(selector, SelectionKey.OP_ACCEPT);
    bridge.clear();
}

protected void bind() throws IOException {
    // allocate an unbound server socket channel
    serverChannel = ServerSocketChannel.open();
    // Get the associated ServerSocket to bind it with
    ServerSocket serverSocket = serverChannel.socket();
    // create a new Selector for use below
    synchronized (Selector.class) {
        // Selector.open() isn't thread safe
        // http://bugs.sun.com/view_bug.do?bug_id=6427854
        // Affects 1.6.0_29, fixed in 1.7.0_01
        this.selector.set(Selector.open());
    }
    // set the port the server channel will listen to
    // serverSocket.bind(new InetSocketAddress(getBind(),
    // getTcpListenPort()));
    bind(serverSocket, getPort(), getAutoBind());
    // set non-blocking mode for the listening socket
    serverChannel.configureBlocking(false);
    // register the ServerSocketChannel with the Selector
    serverChannel.register(this.selector.get(), SelectionKey.OP_ACCEPT);

    // set up the datagram channel
    if (this.getUdpPort() > 0) {
        datagramChannel = DatagramChannel.open();
        configureDatagraChannel();
        // bind to the address to avoid security checks
        bindUdp(datagramChannel.socket(), getUdpPort(), getAutoBind());
    }
}

ClientAcceptor(int port, VoltNetwork network) {
    m_network = network;
    m_port = port;
    ServerSocketChannel socket;
    try {
        socket = ServerSocketChannel.open();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
    m_serverSocket = socket;
}

public void bind(int port) {
    try {
        serverSocket = ServerSocketChannel.open();
        // Avoid TIME_WAIT when killing the server
        serverSocket.socket().setReuseAddress(true);
        // Mac OS X: bind() before calling Selector.register or you don't get accept() events
        serverSocket.socket().bind(new InetSocketAddress(port));
        eventLoop.registerAccept(serverSocket, this);
    } catch (IOException e) { throw new RuntimeException("Failed to bind socket on port #" + port, e); }
}

/**
 * Start the server and begin accepting incoming connections.
 *
 */
@Override
public void start() {
    try{
        this_mon.enter();

   if( !isRunning() ) {
     try {
       server_channel = ServerSocketChannel.open();

       server_channel.socket().setReuseAddress( true );
       if( receive_buffer_size > 0 )  server_channel.socket().setReceiveBufferSize( receive_buffer_size );

       server_channel.socket().bind( bind_address, 1024 );

       if (Logger.isEnabled())  Logger.log(new LogEvent(LOGID, "TCP incoming server socket "    + bind_address));

       AEThread accept_thread = new AEThread( "VServerSelector:port" + bind_address.getPort() ) {
         @Override
         public void runSupport() {
           accept_loop();
         }
       };
       accept_thread.setDaemon( true );
       accept_thread.start();
     }
     catch( Throwable t ) {
        Debug.out( t );
        Logger.log(new LogAlert(LogAlert.UNREPEATABLE,  "ERROR, unable to bind TCP incoming server socket to " +bind_address.getPort(), t));
     }

     last_accept_time = SystemTime.getCurrentTime();  //init to now
   }
    }finally{

        this_mon.exit();
    }
}

@Before
public void setUp() throws IOException {
    eventLoop = new NIOEventLoop();
    serverHandler = new TestHandler();
    acceptSocket = ServerSocketChannel.open();
    // Mac OS X: Must bind() before calling Selector.register, or you don't get accept() events
    acceptSocket.socket().bind(null);
    serverPort = acceptSocket.socket().getLocalPort();
}

MockVolt(int port) {
    try {
        network = new VoltNetwork();
        network.start();
        socket = ServerSocketChannel.open();
        socket.configureBlocking(false);
        socket.socket().bind(new InetSocketAddress(port));
    } catch (IOException e) {
        e.printStackTrace();
    }
}

public TcpProxyServer(int port) throws IOException {
    m_Selector = Selector.open();
    m_ServerSocketChannel = ServerSocketChannel.open();
    m_ServerSocketChannel.configureBlocking(false);
    m_ServerSocketChannel.socket().bind(new InetSocketAddress(port));
    m_ServerSocketChannel.register(m_Selector, SelectionKey.OP_ACCEPT);
    this.Port=(short) m_ServerSocketChannel.socket().getLocalPort();
    System.out.printf("AsyncTcpServer listen on %d success.\n", this.Port&0xFFFF);
}

private static ServerSocketChannel newSocket() {
    try {
        ServerSocketChannel ch = ServerSocketChannel.open();
        ch.configureBlocking(false);
        return ch;
    } catch (IOException e) {
        throw new NioException("Open a server socket error:" + e.getMessage(), e);
    }
}

@Test
public void shouldUseFetchedIpIfSuccessful() throws IOException {
    final ServerSocketChannel channel = createMockedServerSocketChannel(49152);
    final ConnectionHandler handler = Mockito.spy(new ConnectionHandler());
    Mockito.doReturn(channel).when(handler).bindToPort();
    Mockito.doReturn(Optional.of(InetAddress.getByName("168.168.168.168"))).when(handler).tryToFetchFromProviders();

    handler.init();

    assertThat(handler.getIpAddress().getHostAddress()).isEqualTo("168.168.168.168");
}

@Override
public InetAddress getBoundToAddress() {
 if ( server_channels.size() == 0 ){
  return( null);
 }
 ServerSocketChannel    server_channel = (ServerSocketChannel)server_channels.get(0);

 return server_channel.socket().getInetAddress();
}

/**
 * bind address
 * 
 * @param serverSocketChannel
 * serverSocketChannel
 * @param serverParam
 * serverParam
 */
private void bindAddress(ServerSocketChannel serverSocketChannel, ServerParam serverParam) {
    do {
        try {
            serverSocketChannel.socket().bind(new InetSocketAddress(serverParam.getHost(), serverParam.getPort()),
                    serverParam.getBacklog());
            break;
        } catch (IOException e) {
            this.logger.warning("[Accepter] bind address : " + e.toString());
        }
    } while (true);
}

/**
 * Starts listening to slave connections.
 *
 * @throws Exception If fails.
 */
public void beginAccept() throws Exception {
    this.serverSocketChannel = ServerSocketChannel.open();
    this.selector = RemotingUtil.openSelector();
    this.serverSocketChannel.socket().setReuseAddress(true);
    this.serverSocketChannel.socket().bind(this.socketAddressListen);
    this.serverSocketChannel.configureBlocking(false);
    //注册读事件
    this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}

/**
 * {@inheritDoc}
 */
@Override
protected ServerSocketChannel open(SocketAddress localAddress) throws Exception {
    // Creates the listening ServerSocket
    ServerSocketChannel channel = ServerSocketChannel.open();

    boolean success = false;

    try {
        // This is a non blocking socket channel
        channel.configureBlocking(false);

        // Configure the server socket,
        ServerSocket socket = channel.socket();

        // Set the reuseAddress flag accordingly with the setting
        socket.setReuseAddress(isReuseAddress());

        // and bind.
        socket.bind(localAddress, getBacklog());

        // Register the channel within the selector for ACCEPT event
        channel.register(selector, SelectionKey.OP_ACCEPT);
        success = true;
    } finally {
        if (!success) {
            close(channel);
        }
    }
    return channel;
}

static void test4() {
    for (int i=0; i<11000; i++) {
        try {
            ServerSocketChannel sc = ServerSocketChannel.open();
        } catch (Exception e) {
            // Presumably "Too many open files"
        }
    }
}

/**
 * {@inheritDoc}
 */
@Override
protected void close(ServerSocketChannel handle) throws Exception {
    SelectionKey key = handle.keyFor(selector);

    if (key != null) {
        key.cancel();
    }

    handle.close();
}

/**
 * Get the next SocketChannel in the operator we have built from
 * the selected-key et for this selector.
 * 
 * @return The next SocketChannel in the iterator
 */
public ServerSocketChannel next() {
    SelectionKey key = iterator.next();

    if (key.isValid() && key.isAcceptable()) {
        return (ServerSocketChannel) key.channel();
    }

    return null;
}

public static void main(String[] args) throws Exception {

        boolean keepAlive = false;
        String prop = System.getProperty("com.sun.CORBA.transport.enableTcpKeepAlive");
        if (prop != null)
            keepAlive = !"false".equalsIgnoreCase(prop);

        DefaultSocketFactoryImpl sfImpl = new DefaultSocketFactoryImpl();
        ORBImpl orb = new ORBImpl();
        orb.set_parameters(null);
        sfImpl.setORB(orb);

        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(0));

        InetSocketAddress isa = new InetSocketAddress("localhost", ssc.socket().getLocalPort());
        Socket s = sfImpl.createSocket("ignore", isa);
        System.out.println("Received factory socket" + s);
        if (keepAlive != s.getKeepAlive())
            throw new RuntimeException("KeepAlive value not honoured in CORBA socket");
    }

/**
 * Port in use.
 */
@Override
public int getLocalPort() {
    ServerSocketChannel ssc = serverSock;
    if (ssc == null) {
        return -1;
    } else {
        ServerSocket s = ssc.socket();
        if (s == null) {
            return -1;
        } else {
            return s.getLocalPort();
        }
    }
}

public void run() {
    try {
        SelectionKey key;
        log.debug("运行连接请求处理与服务请求接收线程，线程ID：" + Thread.currentThread().getId());
        while (true) {
            selector.select();
            // 首先处理连接请求，并注册连接后的Channel注册到选择器
            // 这种处理连接的方式，可能造成某一个选择器注册的Channel或选择键比其他线程多，导致线程内的繁忙程度不一致
            while ((key = connQueue.poll()) != null) {
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                // 接受一个连接
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                sc.register(selector, SelectionKey.OP_READ);
                if (log.isDebugEnabled()) {
                    log.debug("接受一个客户端连接，处理连接请求的线程ID：" + Thread.currentThread().getId());
                }
            }

            // 再处理服务请求的选择键，并将选择键放入待处理服务队列中
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                SelectionKey keytmp = it.next();
                it.remove();
                if (keytmp.isReadable()) {
                    reqQueue.add(keytmp);
                }
            }
        }
    } catch (IOException e) {
        log.error("处理连接请求，并接收服务请求处理异常", e);
    }
}

private void prepareSocket(int port) throws IOException {
    server = ServerSocketChannel.open();
    server.configureBlocking(false);
    server.socket().bind(new InetSocketAddress(port));

    selector = Selector.open();
    server.register(selector, SelectionKey.OP_ACCEPT);
}

/**
 * Wraps the {@link ServerSocketChannel} into this {@link RapidServerConnection}
 * @param serversocketchannel The {@link ServerSocketChannel} that will be wrapped by this {@link RapidServerConnection}
 */
public void Bind(ServerSocketChannel serversocketchannel)
{
    if(isReady)
    {
        reset();
    }

    Link(serversocketchannel);
}

private void Link(ServerSocketChannel serversocketchannel)
{
    if(serversocketchannel!=null)
    {
        channel = serversocketchannel;
        isReady = true;
    }
}