一、服务端
1、服务端处理器
首先我们下一个服务端部分的处理器
/**
* 类说明：
* 编写NIO的服务端部分的处理器，主要是为了启动服务，收到请求的报文，然后返回返回的报文
*/
public class NioServerHandle implements Runnable{
private Selector selector;//选择器
private ServerSocketChannel serverChannel;//通道，用来沟通系统和程序之间的交互事件
private volatile boolean started; //端口监听是否启动状态
/**
* 构造方法
* 初始化选择器 selector
* 初始化渠道 serverChannel
* 设置渠道的阻塞方式 ，true为阻塞式编程，false为非阻塞式
* 这里主要是基本的API使用方式
* @param port 服务开发的端口号
*/
public NioServerHandle(int port) {
try {
selector = Selector.open();
serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
started = true;
serverChannel.socket().bind( new InetSocketAddress( port ) );
serverChannel.register( selector,SelectionKey.OP_CONNECT );
started = true;
System.out.println("服务端口已经启动");
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
/**
* 停止服务
*/
public void stop(){
started = false;
}
@Override
public void run() {
while (started){//这里是判断服务是否启动着，如果为false就关闭接口
try {
selector.select();//选择器选择中，这是哥阻塞方法，除非有事件触发，否则一般都是阻塞了
Set<SelectionKey> keys = selector.selectedKeys();//获取到所有触发的事件
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key = null;
while (it.hasNext()){//遍历所有的事件
key = it.next();
it.remove();
handleInput(key);//依次处理事件
}
} catch (IOException e) {
e.printStackTrace();
}
}
if(selector!=null){
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 处理事件的方法
* 如果事件为可读事件，就读取缓存中的所有东西，经过处理，然后返回。
* 这个方法有个欠缺，如果请求的报文长度大于1024字节的时候，就会出现报文不完全的情况。
* @param key 操作类型
* @throws IOException
*/
private void handleInput(SelectionKey key) throws IOException {
if(key.isValid()){//先判断事件是不是有效事件
if(key.isAcceptable()){
ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
SocketChannel sc = ssc.accept();
System.out.println("=======建立连接===");
sc.configureBlocking(false);
sc.register(selector,SelectionKey.OP_READ);
}
//处理新接入的请求信息
if(key.isReadable()){
/**isReadbale()官网的解释，判断这个key的同时是否完成
* Tests whether this key's channel is ready for reading.
*/
System.out.println("=====socket channel 数据准备完成，可以去读 读取===");
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(buffer);//从sc中读取1024的数据放到buffer里边
if(readBytes>0){
buffer.flip();//读写转换
byte[] bytes = new byte[buffer.remaining()];////buffer.remaining :buffer可以读取的数据的长度
/**
* buffer.get(bytes)方法的官方解释，将buffer里边的byte传入到参数的数组中
* This method transfers bytes from this buffer into the given
* destination array.
*/
buffer.get(bytes);
String message = new String(bytes,"utf-8");
System.out.println("服务器收到消息，"+message);
String result = response(message);//这个是调用业务逻辑的方法
doWrite(sc,result);//将返回的报文写入到channer中
}
else if(readBytes<0){
key.cancel();
sc.close();
}
}
if (key.isWritable()){
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer attr = (ByteBuffer) key.attachment();
if(attr.hasRemaining()){
int count = sc.write( attr );
System.out.println("write "+ count +
"byte and has R"+attr.hasRemaining());
}else {
sc.register( selector,SelectionKey.OP_READ );
}
}
}
}
/**
* 开始处理返回报文，传入的参数是通道和返回报文，将返回报文写入到渠道中去
* @param sc 传入的渠道信息
* @param response 传入的返回报文
* @throws IOException
*/
private void doWrite(SocketChannel sc, String response) throws IOException {
byte[] bytes = response.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
// sc.write(writeBuffer);
serverChannel.register( selector,SelectionKey.OP_READ|SelectionKey.OP_WRITE,writeBuffer );
}
/**
* 业务逻辑处理
* @param msg 传入的参数是请求报文。
* @return
*/
public static String response(String msg){
return "Hello,"+msg+",Now is "+new java.util.Date(
System.currentTimeMillis()).toString() ;
}
}
2、服务端启用函数
private static NioServerHandle nioServerHandle;
public static void main(String[] args) {
start();
}
private static void start(){
if(nioServerHandle!=null){
nioServerHandle.stop();//如果服务已经起起来了就关了重新启动一下
}
nioServerHandle = new NioServerHandle(12000);//表示监听的端口号
new Thread(nioServerHandle,"server").start();//开始服务，run()方法开始调用
}
二、客户端代码的编写
1、客户端的框架部分
/**
* 类说明：客户端调用的处理器
*/
public class NioClientHandle implements Runnable {
/**
* 需要定义要链接哪个服务端，要连接哪个端口，是否要启动这个连接，渠道是多少
*/
private String host;
private int port;
private volatile boolean started;
private Selector selector;
private SocketChannel socketChannel;
/**
* 构造函数，一个服务端用一个实例
* @param host ip
* @param port 端口
*/
public NioClientHandle(String host,int port){
this.host = host;
this.port = port;
try {
//创建选择器
this.selector = Selector.open();
//打开监听通道
socketChannel = SocketChannel.open();
//设置是否为阻塞模式
socketChannel.configureBlocking(false);
started = true;
} catch (IOException e) {
e.printStackTrace();
System.exit(-1);
}
}
public void stop(){
started = false;
}
@Override
public void run() {
try {
// 先连接
doConnect();
} catch (IOException e) {
e.printStackTrace();
System.exit(-1);
}
while (started){
try {
selector.select();//阻塞方法，至少一个注册事件发生的时候，就会继续
Set<SelectionKey> keys = selector.selectedKeys();//获取到事件
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key =null;
//遍历所有的
while(it.hasNext()){
key = it.next();
it.remove();//现将处理过的selectionKey从选定的集合中删除，如果没有删除，仍然在事件集合中以一个激活的建出现，这会是我们再次处理
try {
handleInput(key);//具体处理事件的方法
} catch (Exception e) {
if(key!=null){
key.cancel();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 处理的逻辑
* @param key 传入的操作类型
* @throws Exception
*/
private void handleInput(SelectionKey key) throws Exception {
if(key.isValid()){
SocketChannel sc = (SocketChannel) key.channel();//
if (key.isConnectable()){//如果是连接事件的话
if(sc.finishConnect()){//确认链接已经建立
socketChannel.register(selector, SelectionKey.OP_READ);//连接建立就去订阅阅读事件，然后有返回的时候，就会进入下边的readable事件处理里边
}else {
System.exit(-1);
}
}
if(key.isReadable()){
ByteBuffer buffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(buffer);
if(readBytes>0){
//读到的数据，进行具体的业务处理//这里应该是返回的报文
}else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
/**
* 进行连接的具体操作
* @throws IOException
*/
private void doConnect() throws IOException {
if(socketChannel.connect(new InetSocketAddress(host,port))){}//如果连接的时候已经连接上了就直接处理
else{
socketChannel.register(selector,SelectionKey.OP_CONNECT);//如果没有连接上就订阅事件，进入到Connectable里边
}
}
/**
* 进行连接的具体操作
* @throws IOException
*/
private void doConnect() throws IOException {
if(socketChannel.connect(new InetSocketAddress(host,port))){}//如果连接的时候已经连接上了就直接处理
else{
socketChannel.register(selector,SelectionKey.OP_CONNECT);//如果没有连接上就订阅事件，进入到Connectable里边
}
}
//发送消息
public void sendMsg(String msg){
//socketChannel，一开始初始化的
doWrite(socketChannel,msg);
}
/**
* 用来发送信息
* @param sc 渠道
* @param msg 写入的信息
* @return
*/
private void doWrite(SocketChannel sc,String msg){
byte[] bytes = msg.getBytes();
//定义buffer
ByteBuffer byteBuffer = ByteBuffer.allocate( bytes.length );
//将bytes放入buffer
byteBuffer.put( bytes );
//改变buffer的读写状态
byteBuffer.flip();
//缓存进入渠道
try {
sc.write( byteBuffer );
} catch (IOException e) {
e.printStackTrace();
}
}
}
2、启用框架
public class NioClient {
private static NioClientHandle nioClientHandle;
public static void main(String[] args) {
start();
NioClient.sendMsg("1123123");
}
private static void start() {
if(nioClientHandle!=null){
nioClientHandle.stop();
nioClientHandle = new NioClientHandle("",1);
new Thread(nioClientHandle,"server").start();
}
}
private static boolean sendMsg(String msg) {
nioClientHandle.sendMsg(msg);
return false;
}
}
网友评论