Java: 网络编程

网络编程

 

socket

在Java中，发送接收数据通过统一的I/O流（InputStream/OutputStream）的方式进行，Java为Socket提供了SocketInputStream/SocketOutputStream进行网络数据读写，这和其他I/O输入输出是一样的，如读取写入文件数据操作是一样的。

 

Java没有为带外（out-of-band）数据提供专门的接收。Java Socket倒是为带外（out-of-band）数据的发送提供了一个专门的#sendUrgentData, sendUrgentData每次只支持一个字节数据的发送。

TCP

 

UDP

 

Java对其他网络协议的支持

 

sun.net.www.protocol.http.Handler

sun.net.www.protocol.https.Handler

 

http

 

https

 

Socket

Java编写Socket程序已经非常简单了，基本上创建一个Socket就完事了。剩下的主要就是处理Socket IO流。

 

Socket IO流

InputStream&OutputStream

 

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;

public class JavaBioEchoServer {
    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket = new ServerSocket(8001);
        while (true) {
            Socket socket = serverSocket.accept();
            new SocketHandler(socket).start();
        }
    }

    private static class SocketHandler extends Thread {
        private Socket socket;
        private boolean closed;

        private String charset;

        public SocketHandler(Socket socket) {
            this.socket = socket;
            this.closed = false;

            this.charset = "GB2312";
        }

        public void run() {
            byte[] bytes = new byte[512];
            int offset = 0;

            try {
                echo("> 连接成功.\r\n");
                while (! closed) {
                    InputStream in = socket.getInputStream();

                    int nbytes = in.read(bytes, offset, bytes.length - offset);
                    if (nbytes > 0) {
                        int i;
                        offset += nbytes;
                        for (i = 0; i < offset; i++) {
                            if (bytes[i] == 10) {
                                break;
                            }
                        }
                        if (i < offset) {
                            handleMessage(new String(bytes, 0, i + 1, charset));
                            System.arraycopy(bytes, i + 1, bytes, 0, bytes.length - i - 1);
                            offset = 0;
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public void handleMessage(String message) throws IOException {
            echo(message);

            message = message.replaceAll("\r\n", "");
            if (message.equals("quit")) {
                socket.close();
                closed = true;
                return;
            }
            System.out.println(message);
        }

        private void echo(String message) throws IOException {
            OutputStream out = socket.getOutputStream();
            out.write(message.getBytes(charset));

//            Writer writer = new BufferedWriter(new OutputStreamWriter(out, charset));
//            writer.write(message);
//            writer.flush();
        }
    }
}

 

阻塞式和非阻塞式

Java中的Socket不像Linux C下创建Socket的时候还可以有阻塞式和非阻塞式Socket的说法。Java中的阻塞式和非阻塞式主要是针对IO的。当然Linux C下创建Socket的时候指定为阻塞式或非阻塞式影响的也是IO的。

 

BIO

就是阻塞式IO。

 

NIO

就是非阻塞式IO。

 

多路复用

写道

多路复用
https://www.iteye.com/blog/lobin-2520040

 

multiplexor

 

Selector

java.nio.channels.Selector

 

Selector selector = Selector.open()

 

 

 

 

Selectors

open

 

public static Selector open() throws IOException {

    return SelectorProvider.provider().openSelector();

}

 

Selectors

SelectorProvider

 

 

 

public static SelectorProvider provider() {

        synchronized (lock) {

                if (provider != null)

                      return provider;

                return (SelectorProvider)AccessController.doPrivileged(new PrivilegedAction() {

                      public Object run() {

                              if (loadProviderFromProperty())

                                    return provider;

                              if (loadProviderAsService())

                                    return provider;

                              provider = sun.nio.ch.DefaultSelectorProvider.create();

                              return provider;

                      }

                });

        }

}

 

默认的Selector提供者

sun.nio.ch.DefaultSelectorProvider，还可通过指定属性java.nio.channels.spi.SelectorProvider进行或者以服务的方式自定义。

通过服务的方式进行自定义：

META-INF/services/SelectorProvider

 

DefaultSelectorProvider在不同平台下都有实现，但具体实现有所不同。

 

 

 

Selectors

Windows平台下，默认的Selector提供者:sun.nio.ch.DefaultSelectorProvider,在jdk\src\windows\classes\sun\nio\ch目录下。

其他实现：

sun.nio.ch.WindowsSelectorProvider，在jdk\src\windows\classes\sun\nio\ch目录下。

 

类Unix平台（包括Linux）下，默认的Selector提供者: sun.nio.ch.DefaultSelectorProvider,在jdk\src\solaris\classes\sun\nio\ch目录下。

其他实现：

sun.nio.ch.EPollSelectorProvider，在jdk\src\solaris\classes\sun\nio\ch目录下。

sun.nio.ch.DevPollSelectorProvider，在jdk\src\share\classes\sun\nio\ch目录下。

sun.nio.ch.PollSelectorProvider，在jdk\src\share\classes\sun\nio\ch目录下。

 

 

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class JavaNioEchoServer {

	private static final String charset = "GB2312";

	public static void main(String[] args) throws Exception {
		new JavaNioEchoServer().start();
	}

	private void start() throws IOException {
		Selector selector = Selector.open();

		ServerSocketChannel channel = ServerSocketChannel.open();
		channel.socket().bind(new InetSocketAddress(8001));

		channel.configureBlocking(false);
		channel.register(selector, SelectionKey.OP_ACCEPT);

		while(true) {
			int readyChannels = selector.select();
			if(readyChannels == 0) {
				continue;
			}

			Set<SelectionKey> selectedKeys = selector.selectedKeys();
			Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
			while(keyIterator.hasNext()) {
				SelectionKey key = keyIterator.next();
				if(key.channel().isOpen() && key.isAcceptable()) {
					ServerSocketChannel server = (ServerSocketChannel) key.channel();
					SocketChannel socketChannel = server.accept();

					socketChannel.configureBlocking(false);
					socketChannel.register(selector, SelectionKey.OP_READ);

					ByteBuffer dst = ByteBuffer.allocate(100);
					socketChannel.keyFor(selector).attach(dst);

					echo(socketChannel, "> 连接成功.\r\n");
				}
				if (key.channel().isOpen() && key.isReadable()) {
					SocketChannel socketChannel = (SocketChannel) key.channel();
					
					ByteBuffer dst = (ByteBuffer) key.attachment();
					int nbytes = socketChannel.read(dst);
					if (nbytes > 0) {
						byte[] bytes = dst.array();

						int i;
						for (i = 0; i < bytes.length; i++) {
							if (bytes[i] == '\n') {
								break;
							}
						}
						if (i < bytes.length) {
							handleMessage(socketChannel, new String(bytes, 0, i + 1, charset));
							dst.get(bytes, 0, i + 1);
							dst.rewind();
						}
					}
				}
				if (key.channel().isOpen() && key.isWritable()) {
					
				}
				keyIterator.remove();
			}
		}
	}

	public void handleMessage(SocketChannel socketChannel, String message) throws IOException {
		echo(socketChannel, message);

		message = message.replaceAll("\r\n", "");
		if (message.equals("quit")) {
			socketChannel.close();
			return;
		}
		System.out.println(message);
	}

	private void echo(SocketChannel socketChannel, String message) throws IOException {
		socketChannel.write(ByteBuffer.wrap(message.getBytes(charset)));
	}
}

 

 

Channel

Channel接口（java.nio.channels.Channel）定义了一个关联I/O操作的接口，它表示一个关联I/O操作的读写通道。所有关联I/O操作的I/O通道都实现了这个接口。

 

通道（Channel）表示一个实体（例如硬件设备、文件、网络套接字、或者一个具有执行一个或多个明确的I/O操作如读或写操作的编程组件，）的开放连接。怎么理解？比如USB设备的读写、文件读写、编程组件如Java标准IO输入输出流。

 

通道要么是打开的要么是关闭的，通道一旦创建就已经打开，并且一旦关闭就只能关闭的。一旦关闭，在通道上执行的io操作将抛出java.nio.channels.ClosedChannelException异常。可以调用isOpen方法判断通道是打开的还是关闭的。

 

 

 

 

java.nio.channels.Channel接口中只定义了2个方法： isOpen， close。同时，java.nio.channels.Channel接口扩展了java.io.Closeable接口，并对close赋予了新的功能定义。

 

 

 

 

isOpen：

用于测试判断通道是打开的还是关闭的.

 

close：

 

关闭通道。通道一旦关闭后，就不能再打开。并且通道关闭后就不能再在这个通道上尝试执行的io操作，否则的话，将抛出java.nio.channels.ClosedChannelException异常。如果通道已经被关闭，再调用close方法将不会有任何影响，也就是说，close方法可调用多次，除了第一次调用该方法关闭通道之外，其他的调用不会有任何影响。并且如果某个线程已经调用了该方法，但这个时候其他的调用将会阻塞直到第一个调用完成，然后才能返回并没有任何影响。

 

Channels

Channels(java.nio.channels.Channels)是Java自带提供的Java NIO Channel读写的工具类。Channels提供了一些工具方法来完成Channel的读写操作。如：

向Channel写入ByteBuffer数据：

private static int write(WritableByteChannel ch, ByteBuffer bb) throws IOException

 

 

通过InputStream流的方式从Channel读：

public static InputStream newInputStream(ReadableByteChannel ch)

 

通过OutputStream流的方式向Channel写入：

public static OutputStream newOutputStream(final WritableByteChannel ch)

 

通过Channel通道方式从InputStream流中读：

public static ReadableByteChannel newChannel(final InputStream in)

 

通过Channel通道方式向OutputStream流中写入：

public static WritableByteChannel newChannel(final OutputStream out)

 

通过Reader方式从Channel读：

public static Reader newReader(ReadableByteChannel ch,

                 CharsetDecoder dec,

                 int minBufferCap)

 

public static Reader newReader(ReadableByteChannel ch,

                 String csName)

 

通过Writer方式向Channel写入：

public static Writer newWriter(final WritableByteChannel ch,

                 final CharsetEncoder enc,

                 final int minBufferCap)

 

public static Writer newWriter(WritableByteChannel ch,

                 String csName)

 

Buffer

Buffer是Java NIO的重要组成部分，它提供了一组用于读写缓冲区操作的实现，这些实现主要分为几大类，包括针对字节的ByteBuffer、MappedByteBuffer、针对字符类型的CharBuffer、针对short类型的ShortBuffer、针对int类型的IntBuffer、针对long类型的LongBuffer、以及针对浮点类型的FloatBuffer和DoubleBuffer，它们都是一些抽象实现，都继承了Buffer抽象类。其中ByteBuffer比较常用，包括MappedByteBuffer，在文件读写、网络IO流读写时经常需要用到，非常重要。当然如果是字符形式的话，如读写字符设备，CharBuffer也很重要。

 

Java NIO提供的这些Buffer都是基于数组的，以及提供一些有用的方法可以很方便的操作缓冲区。

 

包括DirectByteBuffer，它通过MemoryRef在系统内存上分配一块内存作为缓冲区，这块内存称为直接内存，MemoryRef通过一个数组引用（byte[] buffer）引用这块内存，由于DirectByteBuffer继承MappedByteBuffer抽象类。MappedByteBuffer定义为到直接内存等Java内存之外的区域的映射，最后还是在ByteBuffer中通过一个数组引用（byte[] hb）引用这块内存。

 

Java NIO的Buffer主要针对不同类型的数据提供了对应的get、put操作读写缓冲区，以及一些针对Buffer 的有用的方法，如hasRemaining、remaining 、flip、rewind、isReadOnly，以及position、limit、mark、reset、clear、isDirect等方法。

 

还有针对ByteBuffer的slice方法。

 

其中mark、reset这两个方法的使用要非常注意，这两个方法的使用和Java I/O流的InputStream和OutputStream的使用一样，在reset之前一定要确保进行正确的mark，否则将抛出InvalidMarkException异常。

Buffer中capacity、limit以及position是几个非常重要的概念。除此之外，还有mark也很重要，研究过Java I/O流的代码的话，应该对mark有印象。

 

 

 

 

Buffer中的几个重要方法

flip

 

rewind

调用rewind使得Buffer倒回到position等于0，mark等于-1的状态。

ByteBuffer

ByteBuffer除了从Buffer中继承的几个重要的概念，还包括offset、isReadOnly。

 

只读缓存

 

 

ByteBuffer提供了HeapByteBuffer、DirectByteBuffer。其中DirectByteBuffer继承MappedByteBuffer抽象类。MappedByteBuffer定义为到直接内存等Java内存之外的区域的映射。

 

 

HeapByteBuffer实现为final类，我们不能从HeapByteBuffer扩展实现，并且不是public类，这是一个package方法保护的类。DirectByteBuffer倒是一个public类，Android Java SDK的解释是“Not final because it is extended in tests”。

 

ByteBuffer提供了两个方法用来创建HeapByteBuffer和DirectByteBuffer，调用ByteBuffer的allocate方法创建HeapByteBuffer，调用ByteBuffer的allocateDirect方法创建DirectByteBuffer。

    public static ByteBuffer allocate(int capacity) {
        if (capacity < 0)
            throw new IllegalArgumentException();
        return new HeapByteBuffer(capacity, capacity);
    }

以及

    public static ByteBuffer allocateDirect(int capacity) {
        // Android-changed: Android's DirectByteBuffers carry a MemoryRef.
        // return new DirectByteBuffer(capacity);
        DirectByteBuffer.MemoryRef memoryRef = new DirectByteBuffer.MemoryRef(capacity);
        return new DirectByteBuffer(capacity, memoryRef);
    }

 

因为HeapByteBuffer是一个package方法保护的类，不能在java.nio包外面实例化HeapByteBuffer。DirectByteBuffer没有这个限制，可以直接实例化创建DirectByteBuffer。

 

ByteBuffer中的几个重要方法

slice

写道

Creates a new byte buffer whose content is a shared subsequence of this buffer's content.
The content of the new buffer will start at this buffer's current position. Changes to this buffer's content will be visible in the new buffer, and vice versa; the two buffers' position, limit, and mark values will be independent.
The new buffer's position will be zero, its capacity and its limit will be the number of bytes remaining in this buffer, and its mark will be undefined. The new buffer will be direct if, and only if, this buffer is direct, and it will be read-only if, and only if, this buffer is read-only.
Returns:
The new byte buffer

翻译过来的意思是

写道

创建一个新的字节缓冲区，其内容是此缓冲区内容的共享子序列。
新缓冲区的内容将从此缓冲区的当前位置开始。对此缓冲区内容的更改将在新的缓冲区中可见，反之亦然；这两个缓冲区的位置、限制和标记值将是独立的。
新缓冲区的位置将为零，其容量和限制将是此缓冲区中剩余的字节数，并且其标记将未定义。当且仅当此缓冲区是直接的时，新的缓冲区将是直接的，并且当且仅当此缓冲区是只读的时，新的缓冲区将是只读的。
返回：
新字节缓冲区

也就是从一个缓存buffer的当前位置开始，创建一个分片，这个分片也是一个缓存buffer。

 

 

分配一块Buffer

allocate

allocateDirect

 

将一个字节序列封装成一个Buffer.

wrap

 

HeapByteBuffer

HeapByteBuffer在Java Heap(堆)内存上分配一块Buffer。

 

DirectByteBuffer

DirectByteBuffer是直接在系统内存上分配一块Buffer，这块内存称为“Direct Memory”，即直接内存。

 

    final static class MemoryRef {
        byte[] buffer;
        long allocatedAddress;
        final int offset;
        boolean isAccessible;
        boolean isFreed;


        final Object originalBufferObject;

        MemoryRef(int capacity) {
            VMRuntime runtime = VMRuntime.getRuntime();
            buffer = (byte[]) runtime.newNonMovableArray(byte.class, capacity + 7);
            allocatedAddress = runtime.addressOf(buffer);
            // Offset is set to handle the alignment: http://b/16449607
            offset = (int) (((allocatedAddress + 7) & ~(long) 7) - allocatedAddress);
            isAccessible = true;
            isFreed = false;
            originalBufferObject = null;
        }

        MemoryRef(long allocatedAddress, Object originalBufferObject) {
            buffer = null;
            this.allocatedAddress = allocatedAddress;
            this.offset = 0;
            this.originalBufferObject = originalBufferObject;
            isAccessible = true;
        }

        void free() {
            buffer = null;
            allocatedAddress = 0;
            isAccessible = false;
            isFreed = true;
        }
    }

 

 

    java.io.IOException: Broken pipe

        at sun.nio.ch.FileDispatcherImpl.write0(Native Method)

        at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:55)

        at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)

        at sun.nio.ch.IOUtil.write(IOUtil.java:65)

        at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:512)

因服务器Crash导致连接被中断，出现的这个异常。