一.服务端
1.耗能大
package com.learning.notblock;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import static com.learning.bytebuffer.ByteBufferUtil.debugRead;
/**
* @Author wangyouhui
* @Description
**/
@Slf4j
public class Server {
public static void main(String[] args) throws IOException {
// 使用nio来理解阻塞模式
// 0.ByteBuffer
ByteBuffer byteBuffer = ByteBuffer.allocate(16);
// 1.创建服务器
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 非阻塞模式
serverSocketChannel.configureBlocking(false);
// 2.绑定监听端口
serverSocketChannel.bind(new InetSocketAddress(8080));
// 3.连接集合
List<SocketChannel> socketChannelList = new ArrayList<>();
// 与多个客户端连接
while(true){
// 4.accept,建立与客户端的连接
// log.info("connecting...");
// 非阻塞，线程还会继续运行；如果没有连接建立，socketChannel返回的是null
SocketChannel socketChannel = serverSocketChannel.accept();
if(socketChannel != null){
log.info("connected...{}", socketChannel);
// 将socketChannel设置为非阻塞模式
socketChannel.configureBlocking(false);
socketChannelList.add(socketChannel);
}
for(SocketChannel channel : socketChannelList){
// 5.接收客户端发送的数据
// log.info("before read...{}", channel);
// 非阻塞，线程仍然会继续运行。如果没有读到数据，read方法返回0
int read = channel.read(byteBuffer);
if(read > 0){
// 切换到读模式
byteBuffer.flip();
debugRead(byteBuffer);
byteBuffer.clear();
log.debug("after read...{}", channel);
}
}
}
}
}
2.selector处理accept
package com.learning.notblock;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
/**
* @Author wangyouhui
* @Description
**/
@Slf4j
public class SelectorServer {
public static void main(String[] args) throws IOException {
// 1. 创建一个selector,管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 2. 建立selector和channel的联系（注册）
// SelectionKey的作用：将来事件发生后，通过它可以知道和哪个channel的事件
SelectionKey serverSocketChannelSelectionKey = serverSocketChannel.register(selector, 0, null);
log.info("register key:{}", serverSocketChannelSelectionKey);
// 四种事件类型
/*
* accept: 会在有连接请求时触发
* connect: 是客户端，连接建立后触发
* read: 可读事件
* write: 可写事件
*/
// key只关注accept事件
serverSocketChannelSelectionKey.interestOps(SelectionKey.OP_ACCEPT);
serverSocketChannel.bind(new InetSocketAddress(8080));
while(true){
// 3. select 方法，没有事件发生，则线程阻塞；有事件发生，线程才恢复运行
selector.select();
// 4. 处理事件, selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
log.info("key:{}", key);
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
SocketChannel socketChannel = channel.accept();
log.info("{}", socketChannel);
}
}
}
}
3.selector处理cancel
package com.learning.notblock;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
/**
* @Author wangyouhui
* @Description
**/
@Slf4j
public class SelectorServer {
public static void main(String[] args) throws IOException {
// 1. 创建一个selector,管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 2. 建立selector和channel的联系（注册）
// SelectionKey的作用：将来事件发生后，通过它可以知道和哪个channel的事件
SelectionKey serverSocketChannelSelectionKey = serverSocketChannel.register(selector, 0, null);
log.info("register key:{}", serverSocketChannelSelectionKey);
// 四种事件类型
/*
* accept: 会在有连接请求时触发
* connect: 是客户端，连接建立后触发
* read: 可读事件
* write: 可写事件
*/
// key只关注accept事件
serverSocketChannelSelectionKey.interestOps(SelectionKey.OP_ACCEPT);
serverSocketChannel.bind(new InetSocketAddress(8080));
while(true){
// 3. select 方法，没有事件发生，则线程阻塞；有事件发生，线程才恢复运行
// 注意：select在事件未处理时，它不会阻塞
selector.select();
// 4. 处理事件, selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
log.info("key:{}", key);
// ServerSocketChannel channel = (ServerSocketChannel)key.channel();
// SocketChannel socketChannel = channel.accept();
// log.info("{}", socketChannel);
key.cancel();
}
}
}
}
3.客户端正常或异常关闭
package com.learning.notblock;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import static com.learning.bytebuffer.ByteBufferUtil.debugRead;
/**
* @Author wangyouhui
* @Description
**/
@Slf4j
public class SelectorServer {
public static void main(String[] args) throws IOException {
// 1. 创建一个selector,管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 2. 建立selector和channel的联系（注册）
// SelectionKey的作用：将来事件发生后，通过它可以知道和哪个channel的事件
SelectionKey serverSocketChannelSelectionKey = serverSocketChannel.register(selector, 0, null);
log.info("register key:{}", serverSocketChannelSelectionKey);
// 四种事件类型
/*
* accept: 会在有连接请求时触发
* connect: 是客户端，连接建立后触发
* read: 可读事件
* write: 可写事件
*/
// key只关注accept事件
serverSocketChannelSelectionKey.interestOps(SelectionKey.OP_ACCEPT);
serverSocketChannel.bind(new InetSocketAddress(8080));
while(true){
// 3. select 方法，没有事件发生，则线程阻塞；有事件发生，线程才恢复运行
// 注意：select在事件未处理时，它不会阻塞
selector.select();
// 4. 处理事件, selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
// 处理key时要从selectedKeys集合中删除，否则下次处理时会出错。例如channel.accept(),没有连接的时候为null，socketChannel.configureBlocking(false);会报空指针
iterator.remove();
log.info("key:{}", key);
// 5.区分事件类型
if (key.isAcceptable()) {
// 如果是accept
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
SocketChannel socketChannel = channel.accept();
socketChannel.configureBlocking(false);
SelectionKey socketChannelSelectionKey = socketChannel.register(selector,0, null);
socketChannelSelectionKey.interestOps(SelectionKey.OP_READ);
log.info("{}", socketChannel);
} else if(key.isReadable()){
// 如果是read
// 拿到触发事件的channel
try {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
// 如果是正常断开，read方法返回值是-1
int read = channel.read(buffer);
if(read == -1){
key.cancel();
}else{
buffer.flip();
debugRead(buffer);
}
}catch(IOException e){
e.printStackTrace();
// 客户端断开了，需要将key取消（从selector的keys集合中整正删除key)
key.cancel();
}
}
}
}
}
}
4.消息边界超出扩容模式
package com.learning.notblock;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import static com.learning.bytebuffer.ByteBufferUtil.debugAll;
/**
* @Author wangyouhui
* @Description
**/
@Slf4j
public class SelectorServer {
private static void split(ByteBuffer source) {
// 切换为读模式
source.flip();
for(int i=0;i<source.limit();i++){
// 找到一条完整消息
if (source.get(i)=='\n') {
int length = i + 1 -source.position();
// 把这条完整消息存入新的ByteBuffer
ByteBuffer target = ByteBuffer.allocate(length);
// 从source读，向target写
for(int j=0;j<length;j++){
target.put(source.get());
}
debugAll(target);
}
}
// 从上次未读的位置读
source.compact();
}
public static void main(String[] args) throws IOException {
// 1. 创建一个selector,管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
// 2. 建立selector和channel的联系（注册）
// SelectionKey的作用：将来事件发生后，通过它可以知道和哪个channel的事件
SelectionKey serverSocketChannelSelectionKey = serverSocketChannel.register(selector, 0, null);
log.info("register key:{}", serverSocketChannelSelectionKey);
// 四种事件类型
/*
* accept: 会在有连接请求时触发
* connect: 是客户端，连接建立后触发
* read: 可读事件
* write: 可写事件
*/
// key只关注accept事件
serverSocketChannelSelectionKey.interestOps(SelectionKey.OP_ACCEPT);
serverSocketChannel.bind(new InetSocketAddress(8080));
while(true){
// 3. select 方法，没有事件发生，则线程阻塞；有事件发生，线程才恢复运行
// 注意：select在事件未处理时，它不会阻塞
selector.select();
// 4. 处理事件, selectedKeys 内部包含了所有发生的事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
// 处理key时要从selectedKeys集合中删除，否则下次处理时会出错。例如channel.accept(),没有连接的时候为null，socketChannel.configureBlocking(false);会报空指针
iterator.remove();
log.info("key:{}", key);
// 5.区分事件类型
if (key.isAcceptable()) {
// 如果是accept
ServerSocketChannel channel = (ServerSocketChannel)key.channel();
SocketChannel socketChannel = channel.accept();
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 将一个byteBuffer作为附件关联到selectionKey上
SelectionKey socketChannelSelectionKey = socketChannel.register(selector,0, buffer);
socketChannelSelectionKey.interestOps(SelectionKey.OP_READ);
log.info("{}", socketChannel);
} else if(key.isReadable()){
// 如果是read
// 拿到触发事件的channel
try {
SocketChannel channel = (SocketChannel) key.channel();
// ByteBuffer buffer = ByteBuffer.allocate(16);
// 获取selectionKey上关联的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 如果是正常断开，read方法返回值是-1
int read = channel.read(buffer);
if(read == -1){
key.cancel();
}else{
split(buffer);
if(buffer.position() == buffer.limit()) {
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
buffer.flip();
newBuffer.put(buffer);
key.attach(newBuffer);
}
}
}catch(IOException e){
e.printStackTrace();
// 客户端断开了，需要将key取消（从selector的keys集合中整正删除key)
key.cancel();
}
}
}
}
}
}
二.客户端
package com.learning.notblock;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
/**
* @Author wangyouhui
* @Description TODO
**/
public class Client {
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
System.out.println("waiting...");
}
}
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/92323.html
