Java Socket实现NIO通信

news2024/11/19 13:45:16

文章目录

    • 一.简单介绍
      • 通道（Channel）
      • 多路复用器（Selector）
    • 二.代码实现.
      • 客户端
      • 服务端
      • 运行结果

一.简单介绍

      NIO 很多人也称之为 Non-block I/O，即非阻塞 I/O，因为这样叫，更能体现它的特点。

为什么这么说呢？传统的 I/O 存在阻塞问题。因为对 Socket 的输入流进行读取时，读取流会一直阻塞，直到发生以下三种情况的任意一种才会解除阻塞：

  • 有数据可读；
  • 连接释放；
  • 空指针或 I/O 异常。

阻塞问题，就是传统 I/O 最大的弊端。

NIO 发布后，通道和多路复用器这两个基本组件实现了 NIO 的非阻塞，下面我们就一起来了解下这两个组件的优化原理。

通道（Channel）

      传统 I/O 的数据读取和写入是从用户空间到内核空间来回复制，而内核空间的数据是通过操作系统层面的 I/O 接口从磁盘读取或写入。

      最开始，在应用程序调用操作系统 I/O 接口时，是由 CPU 完成分配，这种方式最大的问题是“发生大量 I/O 请求时，非常消耗 CPU“；之后，操作系统引入了 DMA（直接存储器存储），内核空间与磁盘之间的存取完全由 DMA 负责，但这种方式依然需要向 CPU 申请权限，且需要借助 DMA 总线来完成数据的复制操作，如果 DMA 总线过多，就会造成总线冲突。

      通道的出现解决了以上问题，Channel 有自己的处理器，就是在DMA的基础上增加了能执行有限通道指令的I/O控制器，代替CPU管理控制外设.它有自己的指令系统，是一个协处理器，他实质能够执行有限的输入输出指令,并且由专门通讯传输的通道总线完成控制.

      所以,Channel可以完成内核空间和磁盘之间的 I/O 操作。在 NIO 中，我们读取和写入数据都要通过 Channel，由于 Channel 是双向的，所以读、写可以同时进行。

多路复用器（Selector）

      Selector 是 Java NIO 编程的基础。用于检查一个或多个 NIO Channel 的状态是否处于可读、可写。

      Selector 是基于事件驱动实现的，我们可以在 Selector 中注册 accpet、read 监听事件，Selector 会不断轮询注册在其上的 Channel，如果某个 Channel 上面发生监听事件，这个 Channel 就处于就绪状态，然后进行 I/O 操作。

      一个线程使用一个 Selector，通过轮询的方式，可以监听多个 Channel 上的事件。我们可以在注册 Channel 时设置该通道为非阻塞，当 Channel 上没有 I/O 操作时，该线程就不会一直等待了，而是会不断轮询所有 Channel，从而避免发生阻塞。

      目前操作系统的 I/O 多路复用机制都使用了 epoll，相比传统的 select 机制，epoll 没有最大连接句柄 1024 的限制。所以 Selector 在理论上可以轮询成千上万的客户端。

二.代码实现.

总共涉及四个类,以下代码可以直接拿去用

在这里插入图片描述

消息得大概处理流程是这样得:

  1. 创建Channel,监听连接
  2. 创建多路复用器Selector,将Channel注册到Selector中
  3. Selector去轮询所有注册在其上的channel,当发现一个或多个就绪时,返回就绪的监听事件
  4. 程序匹配到监听事件,进行处理

这里扩充一下细节,在将Channel注册到Selector中时,需要指定监听事件,这个监听事件正好对应Socket通信中的 conect、accept、read 、write 几个阻塞操作.
在这里插入图片描述

客户端

Client 类.

import java.util.Scanner;

public class Client {

	private static String DEFAULT_HOST = "127.0.0.1";  
	
    private static int DEFAULT_PORT = 8081;  
	
    private static ClientHandler clientHandler;  
	
    public static void start(){  
        start(DEFAULT_HOST,DEFAULT_PORT);  
    }  
	
    public static synchronized void start(String ip,int port){  
	 
        clientHandler = new ClientHandler(ip,port);  
        new Thread(clientHandler,"Server").start();  
	 
    }  
	
    //向服务器发送消息  
    public static boolean sendMsg(String msg) throws Exception{  
        clientHandler.sendMsg(msg);  
        return true;  
	
    } 
    
	@SuppressWarnings("resource")
	public static void main(String[] args) {

		// 运行客户端
		Client.start();

		try {
			while (Client.sendMsg(new Scanner(System.in).nextLine()));
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}


}

ClientHandler类.

public class ClientHandler implements Runnable {

	private SocketChannel socketChannel;
	private int port;
	private Selector selector;
	private String host;
	private boolean stop;

	public ClientHandler(String host, int port) {

		this.host = host;
		this.port = port;

		try {
			selector = Selector.open();
			socketChannel = SocketChannel.open();
			socketChannel.configureBlocking(false);

		} catch (IOException e) {
			e.printStackTrace();
			System.exit(1);
		}

	}

	public void run() {
		try {
			doConnect();// 连接
		} catch (IOException e) {
			e.printStackTrace();
		}
		while (!stop) {
			try {
				selector.select(1000);
				Set<SelectionKey> selectionKeys = selector.selectedKeys();
				Iterator<SelectionKey> it = selectionKeys.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();
			}
		}
	}

	private void handleInput(SelectionKey key) throws IOException {
		//这个通道已经可以读取了
		if (key.isValid()) {
			SocketChannel sc = (SocketChannel) key.channel();
			if (key.isConnectable()) {
				if (sc.finishConnect()) {
				} else {
					System.exit(1);
				}
			}
			if (key.isReadable()) {// 读取消息
				ByteBuffer readBuffer = ByteBuffer.allocate(1024);
				int read = sc.read(readBuffer);
				if (read > 0) {
					readBuffer.flip();
					byte[] bytes = new byte[readBuffer.remaining()];
					readBuffer.get(bytes);
					String body = new String(bytes, "utf-8");
					System.out.println("现在时间为：" + body);
					this.stop = true;
				} else if (read < 0) {
					key.cancel();
					sc.close();
				} else {
				}
			}

		}
	}

	private void doConnect() throws IOException {
		if (socketChannel.connect(new InetSocketAddress(host, port))) {
			socketChannel.register(selector, SelectionKey.OP_READ);
		} else {
			socketChannel.register(selector, SelectionKey.OP_CONNECT);
		}

	}

	private void doWrite(SocketChannel socketChannel,String request)throws IOException {
		byte[] bytes = request.getBytes();
		ByteBuffer writeBuff = ByteBuffer.allocate(bytes.length);
		writeBuff.put(bytes);
		writeBuff.flip();
		socketChannel.write(writeBuff);
		if (!writeBuff.hasRemaining()) {
			System.out.println("客户端发送命令成功");
		}
	}
	

	public void sendMsg(String msg) throws Exception{  

        doWrite(socketChannel, msg);  
	
    }  


}

服务端

Server类.


public class Server {

	private static int DEFAULT_PORT = 8081;
	private static ServerHandler serverHandler;

	public static void start() {
		start(DEFAULT_PORT);
	}

	public static synchronized void start(int port) {

		if (serverHandler != null) {
			serverHandler.stop();
		}
		
		serverHandler = new ServerHandler(port);
		new Thread(serverHandler, "Server").start();
	}

	/**
	 * 1.创建channel,监听连接
	 * 2.创建多路复用器Selector,将Channel注册到Selector中
	 * 3.Selector去轮询所有注册在其上的channel,当发现一个或多个就绪时,返回就绪的监听事件
	 * 4.程序匹配到监听事件,进行处理
	 * @param args
	 */
	public static void main(String[] args) {

		// 运行服务器
		Server.start();

	}

}

ServerHandler类.

public class ServerHandler implements Runnable {

    private Selector selector = null;
    private ServerSocketChannel serverChannel = null;
    private boolean stop;

	/**
	 * 初始化多路复用器，绑定监听端口
	 *
	 * @param port
	 */
    public ServerHandler(int port) {
        try {
            selector = Selector.open();
            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            serverChannel.socket().bind(new InetSocketAddress(port), 1024);
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器监听" + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }


    }

    public void stop() {
        this.stop = true;
    }

    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectionKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (IOException e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null)
                                key.channel().close();
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
	
    /**
     * 处理事件
     * @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();
                sc.configureBlocking(false);
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuff = ByteBuffer.allocate(1024);
                //非阻塞的
                int read = sc.read(readBuff);
                if (read > 0) {
                    readBuff.flip();
                    byte[] bytes = new byte[readBuff.remaining()];
                    readBuff.get(bytes);
                    String body = new String(bytes, "utf-8");
                    System.out.println("服务收到消息：" + body);
                    String currentTime = new Date(System.currentTimeMillis()).toString();
                    doWrite(sc, currentTime);
                } else if (read < 0) {
                    key.cancel();
                    sc.close();
                } else {
                }
            }
        }
    }

    /**
     * 异步发送应答消息 
     * @param sc
     * @param content
     * @throws IOException
     */
    private void doWrite(SocketChannel sc, String content) throws IOException {
        if (content != null && content.trim().length() > 0) {
            byte[] bytes = content.getBytes();
            ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
            byteBuffer.put(bytes);
            byteBuffer.flip();
            sc.write(byteBuffer);
        }
    }
}

运行结果

在这里插入图片描述
在这里插入图片描述

今天的分享就到这里了,有问题可以在评论区留言,均会及时回复呀.
我是bling,未来不会太差,只要我们不要太懒就行, 咱们下期见.
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/112489.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

kdump功能

kdump功能

kdump功能前言1 kdump流程2 kdump配置2.1 kexec、makedumpfile编译2.2 系统内核2.3 捕获内核3 kdump测试4 kdump的不足前言 kdump 是一种先进的基于 kexec 的内核崩溃转储机制。当系统崩溃时&#xff0c;kdump会将内存导出为vmcore保存到磁盘。 在kernel1运行的时候&#xff…
阅读更多...
RabbitMQ——延迟队列

RabbitMQ——延迟队列

目录 一、延迟队列的应用场景 1. 场景&#xff1a;"订单下单成功后&#xff0c;15分钟未支付自动取消" ① 传统处理超时订单 ② RabbitMQ延时队列方案 二、延迟队列中的消息投递和消息消费 1.TTL 和 DLX ① TTL ② DLX和死信队列 ③ 延迟队列 ④ 开发步骤 …
阅读更多...
spring mvc 通过异常封装 验证 方法

spring mvc 通过异常封装 验证 方法

正常情况 我们先演示一下正常情况下我们验证的方法。 首先定义一个LoginBean Data public class LoginBean {// Blank 不允许保存空格&#xff0c;空格不算内容NotBlank(message "用户名不能为空")String username;// Empty 允许保留空格&#xff0c;是空格也算内容…
阅读更多...
【Spring】核心部分之AOP:通过列举代码例子，从底层刨析，深入源码，轻轻松松理解Spring的核心AOP，AOP有这一篇足以

【Spring】核心部分之AOP:通过列举代码例子，从底层刨析，深入源码，轻轻松松理解Spring的核心AOP，AOP有这一篇足以

AOP基本概念基本原理专业术语案例演示基于注解&#xff08;重点&#xff09;基于配置文件基本概念 面向切面编程&#xff0c;也叫面向方面编程&#xff0c;利用aop可以对业务逻辑的各个部分进行隔离&#xff0c;从而使得业务逻辑各个部分之间降低耦合&#xff0c;提高程序的可…
阅读更多...
AQS 对资源的共享方式

AQS 对资源的共享方式

AQS 定义两种资源共享方式 1) Exclusive&#xff08;独占&#xff09; 只有一个线程能执行&#xff0c;如 ReentrantLock。又可分为公平锁和非公平锁,ReentrantLock 同时支持两种锁,下面以 ReentrantLock 对这两种锁的定义做介绍&#xff1a; 下面来看 ReentrantLock 中相关的…
阅读更多...
概要设计说明书（GB8567——88）基于协同的在线表格forture-sheet

概要设计说明书（GB8567——88）基于协同的在线表格forture-sheet

概要设计说明书 1引言 1.1编写目的 为了帮助用户更好的了解和使用本在线表格&#xff0c;提高用户与软件的亲和度。 用户手册描述配置和使用改在线表格&#xff0c;以及该软件使用过程中应该注意的一下问题。 1.2背景 说明&#xff1a; 本用户手册所描述的软件系统的名称…
阅读更多...
医疗检测数据存储管理系统

医疗检测数据存储管理系统

摘要 医疗信息化的迅速发展导致了医疗数据的指数型增长&#xff0c;医疗检测数据存储管理系统给医院现有信息系统带了巨大的压力。一方面,随着各种非结构化数据的不断涌现&#xff0c;现有的医疗信息系统在存储空间&#xff0c;存储速度、存储结构上达不到医疗检测数据的要求,不…
阅读更多...
Github惊现神作，这份算法宝典让你横扫各大厂算法面试题

Github惊现神作，这份算法宝典让你横扫各大厂算法面试题

时间飞逝&#xff0c;转眼间毕业七年多&#xff0c;从事 Java 开发也六年了。我在想&#xff0c;也是时候将自己的 Java 整理成一套体系。 这一次的知识体系面试题涉及到 Java 知识部分、性能优化、微服务、并发编程、开源框架、分布式等多个方面的知识点。 写这一套 Java 面试…
阅读更多...
使用Docker搭建Nacos的持久化和集群部署

使用Docker搭建Nacos的持久化和集群部署

1. 准备 1.1 mysql安装 下载镜像 docker pull mysql/mysql-server:5.7 在宿主机中相关目录&#xff0c;用于挂载容器的相关数据 mkdir -p /data/mysql/{conf,data} 编写my.cnf配置文件&#xff0c;在/data/mysql/conf目录中 (或下载 直接上传即可) my.cnf.txt - 蓝奏云 / …
阅读更多...
【考研加油】所有上岸的考研人都有一个共同的特点，就是他们都参加考试了。2023考研加油。

【考研加油】所有上岸的考研人都有一个共同的特点，就是他们都参加考试了。2023考研加油。

声明：为 2023考研的朋友加油！ 2023考研加油 今明两天，将是大部分2023考研人，真正“上战场”的时候。 我想，只有经历过的人，才能对这一历程，感同身受吧！ 为你们加油！ 以下是在QQ空间看到的一组图，与你们共勉。 距考研还有____天！ 确定目标院校中…跨考又能如何？…
阅读更多...
阿里人在Github分享的Spring Cloud全栈笔记，你想象不到有多全

阿里人在Github分享的Spring Cloud全栈笔记，你想象不到有多全

微服务到底是什么 微服务到底是什么&#xff0c;一直众说纷纭&#xff0c;我们只知道各大企业纷纷追捧和实践微服务架构&#xff0c;有的项目可能使用了Spring Cloud就算是使用微服务了&#xff0c;然后说微服务就是Spring Cloud&#xff0c;有的系统可能越做越像SOA&#xff…
阅读更多...
RV1126笔记十六：吸烟行为检测及部署＜四＞

RV1126笔记十六：吸烟行为检测及部署＜四＞

若该文为原创文章&#xff0c;转载请注明原文出处。 转换成onnx模型(windows) 一、查看pt文件 准备好训练好的pt文件,可以用Netron打开看看大概长啥样: 二、模型转换 主要的目的是想把训练好的pt文件转成onnx模型&#xff0c;为后面RV1126的部署做铺垫。 我们是在py38的con…
阅读更多...
java之多线程的三种不同创建方式and通过多线程模拟龟兔赛跑

java之多线程的三种不同创建方式and通过多线程模拟龟兔赛跑

Process与Thread&#xff1a; 程序是指令和数据的有序集合&#xff0c;其本身没有任何运行的含义&#xff0c;是一个静态的概念&#xff0c;而进程则是执行程序的一次执行过程&#xff0c;它是一个动态的概念&#xff0c;是系统资源分配的单位&#xff0c;通常在一个进程中可以…
阅读更多...
视频素材网，视频剪辑必备。

视频素材网，视频剪辑必备。

视频剪辑没素材&#xff0c;推荐6个网站帮你解决&#xff0c;免费可商用&#xff0c;建议收藏&#xff01; 1、菜鸟图库 视频素材下载_mp4视频大全 - 菜鸟图库 网站有超多视频素材&#xff0c;全部都是高清无水印&#xff0c;各种类型都有&#xff0c;像自然、城市、动物、科技…
阅读更多...
自动控制原理笔记-线性系统的时域分析与校正

自动控制原理笔记-线性系统的时域分析与校正

目录 时域法的概述&#xff1a; 时域法的作用和特点&#xff1a; 时域法常用的四个时间信号&#xff1a; 线性系统时域性能指标&#xff1a; 五个常用的性能指标&#xff1a; 一阶系统的时间响应及动态性能&#xff1a; 一阶系统动态指标的计算&#xff1a; 一阶系统的典型…
阅读更多...
Github一夜爆火的阿里高并发技术小册究竟有什么魅力

Github一夜爆火的阿里高并发技术小册究竟有什么魅力

阿里在农历2021到来之前却是又搞了一个大动作&#xff01;把阿里这一年在应对高并发流量的技术经验整合成一份技术小册开源分享供大家学习借鉴。我也是昨天才发现这份小册开源至Github上居然一夜爆火&#xff01; 看了小册之后才知道&#xff0c;原来阿里在应对高并发大流量时也…
阅读更多...
python中的json数据和pyecharts模块入门

python中的json数据和pyecharts模块入门

目录 一json数据格式 1.什么是json 2.json有什么用 3.json格式数据转化 4.python数据和json数据的相互转化 5.json总结 二.pyecharts模块入门 1.基础折线图 全局配置选项——set_global_opts方法 一json数据格式 1.什么是json JSON是一种轻量级的数据交互格式。可以按…
阅读更多...
RabbitMQ 第一天 基础 3 RabbitMQ 快速入门 3.2 入门程序【消费者】

RabbitMQ 第一天 基础 3 RabbitMQ 快速入门 3.2 入门程序【消费者】

RabbitMQ 【黑马程序员RabbitMQ全套教程&#xff0c;rabbitmq消息中间件到实战】 文章目录RabbitMQ第一天 基础3 RabbitMQ 快速入门3.2 入门程序3.2.1 消费者3.2.2 小结第一天 基础 3 RabbitMQ 快速入门 3.2 入门程序 3.2.1 消费者 之前我们 已经完成了生产者的基本代码编…
阅读更多...
客快物流大数据项目（九十八）：ClickHouse的SQL函数

客快物流大数据项目（九十八）：ClickHouse的SQL函数

文章目录 ClickHouse的SQL函数 一、​​​​​​​​​​​​​​类型检测函数
阅读更多...
Verilog刷题HDLBits——Exams/review2015 fancytimer

Verilog刷题HDLBits——Exams/review2015 fancytimer

Verilog刷题HDLBits——Exams/review2015 fancytimer题目描述代码结果题目描述 This is the fifth component in a series of five exercises that builds a complex counter out of several smaller circuits. You may wish to do the four previous exercises first (counte…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023