程序员经验分享-hwhale

socket(二)--Tcp同步非阻塞式

2023-05-16

socket(二)–Tcp同步非阻塞式

文章目录

  • socket(二)--Tcp同步非阻塞式
      • 一、简介
      • 二、关键类
        • 2.1 ServerSocketChannel
        • 2.2 SocketChannel
        • 2.3 Selector
        • 2.4 SelectionKey
        • 2.5 Buffer
      • 三、示例
        • 3.1 服务端代码
        • 3.2 客户端代码
      • 四、注意

一、简介

同步阻塞式通信，工作线程一次只能处理一个连接请求，服务完成后，才可处理下一个连接请求。针对这种情况，有两种解决方案：

  1. 采用多线程编程，即是：主线程负责接收连接请求，连接成功后，服务的过程（即是数据交互的过程）交由其它多个线程处理;
  2. 采用非阻塞式通信，主要由java.nio包中的类实现。实现方式是，工作线程采用轮询的方式，不停监控连接事件、读事件、写事件，当某事件发生时，就进行相应事件的处理。

这里对第2种方式的tcp同步非阻塞通信方式进行介绍。

二、关键类

2.1 ServerSocketChannel

ServerSocketChannel是ServerSocket的替代类，支持阻塞和非阻塞两种方式，默认是阻塞式的。

2.2 SocketChannel

SocketChannel是Socket的替代类，同样支持阻塞和非阻塞两种式，默认是阻塞式的。

2.3 Selector

监听器，用于监听连接就绪事件、读就绪事件、写就绪事件。方法有：

keys()：注册的所有事件;
selectedKeys()：已经发生的事件;

2.4 SelectionKey

事件的句柄，当某事件的selectionKey对象位于Selector对象的selectedKeys()集合中时，表示事件发生。
具体事件有：

SelectionKey.OP_ACCEPT：接收连接就绪事件，表示至少有一个连接进来;
SelectionKey. OP_CONNECT：连接就绪事件，表示与服务连接成功;
SelectionKey.OP_READ：读就绪事件，表示输入流有数据了，可进行数据读;
SelectionKey.OP_WRITE：写就绪事件，表示可以向输出流写数据了。

2.5 Buffer

由于数据输入输出比较耗时，buffer用于缓冲，buffer有三个属性：

capacity：即容量，表示缓冲区可以保存多少数据;
limit：即极限，表示当前缓冲区的终点，读取操作只有在极限范围内。极限可以修改，可用于重用缓冲区，是非负整数，不能大于容量;
position：即位置，表示下一次读写的位置，是非负整数，不能大于极限。

另外buffer提供了改变上述属性的方法：

clear()：极限设为容量，位置设为0；
flip()：极限设为位置，位置设为0；
rewind()：位置设为0；
compact()：压缩，即删除0到位置间的数据，然后将位置与极限之前的数据移动，使位置再次为0;

三、示例

这里以创建服务端和客户端，两者可自由发送消息给对方为例。解释请查看注释。

3.1 服务端代码

import org.apache.commons.lang.StringUtils;
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.nio.charset.Charset;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;

public class TcpNioServerSocketChannelPaperMain {
    public static void main(String[] args) throws Exception {
        //指定端口
        int port = 7001;
        //指定编码
        Charset charset = Charset.forName("utf-8");

        //创建服务
        ServerSocketChannel server = ServerSocketChannel.open();
        //指定为非阻塞，默认为阻塞式
        server.configureBlocking(false);
        //绑定端口
        server.socket().bind(new InetSocketAddress(port));
        System.out.println("服务启动");

        //读缓冲
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
        //写缓冲

        //创建监听器
        Selector selector = Selector.open();
        //注册连接就绪事件
        server.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            //阻塞等待最多1000毫秒获取事件，有事件时会立即返回,另外还有select()和selectNow().分别表示一直等和立即返回
            if (selector.select(1000) < 1) {
                continue;
            }

            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();
                it.remove();
                try {

                    if (key.isAcceptable()) {  //接收事件
                        //获取key关联的ServerSocketChannel
                        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                        //获取key连接的SocketChannel
                        SocketChannel sc = ssc.accept();
                        System.out.println("receive connection,  address:" + sc.socket().getRemoteSocketAddress());

                        //指定为非阻塞，默认为阻塞式
                        sc.configureBlocking(false);
                        //添加附件缓冲，接收数据
                        ByteBuffer attachBuffer = ByteBuffer.allocateDirect(1024);
                        //注册读写事件
                        sc.register(selector, SelectionKey.OP_READ, attachBuffer);
                        new WriteThread(sc).start();
                    } else if (key.isReadable()) {  //读事件
                        //获取key关联的buffer
                        ByteBuffer attachBuffer = (ByteBuffer) key.attachment();
                        //获取key连接的SocketChannel
                        SocketChannel sc = (SocketChannel) key.channel();
                        try {
                            sc.read(readBuffer);
                            readBuffer.flip();
                            String msg = charset.decode(readBuffer).toString();
                            System.out.println("server receive msg : " + msg);
                            readBuffer.clear();

                            //收到bye或空串，则注销读事件并关闭当前连接
                            if ("bye".equals(msg) || StringUtils.isBlank(msg)) {
                                key.interestOps(key.interestOps() & SelectionKey.OP_READ);
                                sc.close();
                            }
                        } catch (Exception e) {
                            sc.close();
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /
     * 发送消息线程
     */
    public static class WriteThread extends Thread {
        private SocketChannel sc;
        public WriteThread(SocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void run() {
            //写缓冲
            ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            while (true) {
                try {
                    //处理于未连接状态，则退出
                    if(!sc.isConnected()){
                        break;
                    }

                    System.out.print("server send msg:");
                    String msg = scanner.nextLine();
                    //发送bye,并关闭当前链接
                    if ("bye".equals(msg)) {
                        sc.close();
                        break;
                    }

                    //输入为空，则不发送消息
                    if (StringUtils.isBlank(msg)) {
                        continue;
                    }

                    //消息写入缓冲区，并发送消息
                    writeBuffer.put(msg.getBytes());
                    writeBuffer.flip();
                    sc.write(writeBuffer);

                    //发送完后，清空缓冲区
                    writeBuffer.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

3.2 客户端代码

import org.apache.commons.lang.StringUtils;

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.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;

public class TcpNioSocketChannelPaperMain {
    public static void main(String[] args) throws Exception {
        //指定编码
        Charset charset = Charset.forName("utf-8");
        //指定host
        String host = "127.0.0.1";
        //指定端口
        int port = 7001;

        //创建SocketChanel
        SocketChannel sc = SocketChannel.open();
        InetSocketAddress isa = new InetSocketAddress(host, port);
        //非阻塞方式
        sc.configureBlocking(false);
        //连接
        sc.connect(isa);

        //读缓冲
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);

        //创建监听器
        Selector selector = Selector.open();
        //注册连接事件
        sc.register(selector, SelectionKey.OP_CONNECT);
        while (true) {
            //等事件出现
            if (selector.select() < 1) {
                continue;
            }

            //获取发生的事件
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                //获取事件，移除正在处理的事件
                SelectionKey key = it.next();
                it.remove();
                try {
                    if (key.isConnectable()) { //连接成功事件
                        System.out.println("连接服务器成功");

                        //获取key关联的socketChannel，这里和上面的sc是相同的，也就是socketChannel==sc为true
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        //注册读事件
                        socketChannel.register(selector, SelectionKey.OP_READ);

                        //完成连接
                        if (socketChannel.isConnectionPending()) {
                            socketChannel.finishConnect();
                        }

                        //启动单独的发送消息线程
                        new WriteThread(socketChannel).start();
                    } else if (key.isReadable()) { //读事件
                        //获取key关联的socketChannel
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        try {
                            //读取数据
                            socketChannel.read(readBuffer);
                            readBuffer.flip();
                            String msg = charset.decode(readBuffer).toString();
                            //空消息不输出
                            if (StringUtils.isBlank(msg)) {
                                continue;
                            }
                            System.out.println("client receive msg : " + msg);
                            readBuffer.clear();

                            //收到bye或空串，则注销读事件并关闭当前连接
                            if ("bye".equals(msg) || StringUtils.isBlank(msg)) {
                                key.interestOps(key.interestOps() & SelectionKey.OP_READ);
                                socketChannel.close();
                            }

                        } catch (Exception e) {
                            socketChannel.close();
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /
     * 发送消息线程
     */
    public static class WriteThread extends Thread {
        //连接的socket通道
        private SocketChannel sc;

        public WriteThread(SocketChannel sc) {
            this.sc = sc;
        }

        @Override
        public void run() {
            //写缓冲
            ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            while (true) {
                try {
                    //处理于未连接状态，则退出
                    if (!sc.isConnected()) {
                        break;
                    }
                    System.out.print("client send msg:");
                    String msg = scanner.nextLine();
                    //发送bye,并关闭当前链接
                    if ("bye".equals(msg)) {
                        sc.close();
                        break;
                    }

                    //输入为空，则不发送消息
                    if (StringUtils.isBlank(msg)) {
                        continue;
                    }

                    //消息写入缓冲区，并发送消息
                    writeBuffer.put(msg.getBytes());
                    writeBuffer.flip();
                    sc.write(writeBuffer);

                    //发送完后，清空缓冲区
                    writeBuffer.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                    try {
                        sc.close();
                    } catch (IOException e1) {
                        e1.printStackTrace();
                    }
                }
            }
        }
    }
}

四、注意

1、通常写事件不进行注册，可直接调socketChannel.write()直接发送。
写就绪事件，监控的是内核写缓冲区是否可写，当内核写缓冲区有空闲空间时，就会发生写就缓绪事件，而一般情况下内核写缓冲区是空闲的，这会导致cpu空转。所以一般不注册写就绪事件，或者写时再注册，写完就取消。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

socket(二)--Tcp同步非阻塞式 的相关文章

随机推荐

  • 光流传感器 定位精度_光流传感器其它方面的应用

    光流传感器可以通过在一定的时间内拍摄两张不同的照片 进而计算出物体运动的速度 光流是一种简单实用的图像运动表达方式 通常定义为一个图像序列中的图像亮度模式的表观运动 光流法检测运动物体的基本原理是 xff1a 给图像的每一个像素点赋予一个速

  • 6.28-机器人模拟器Gazebo基础

    gazebo基础学习 前言 在算法人员开发出可以真机使用的算法之前进行仿真学习机器人物理仿真器的基本使用 xff0c 包括创建场景 xff0c 制作ROS控制接口等 目录 gazebo基础学习 前言 目录 参考 学习记录 基础 安装gaze

  • Poco C++库网络模块例子解析2-------HttpServer

    下面程序取自 Poco 库的Net模块例子 HTTPServer 下面开始解析代码 include 34 Poco Net HTTPServer h 34 继承自TCPServer 实现了一个完整的HTTP多线程服务器 include 34

  • 【0928 | Day 39】事务（精讲）

    目录 一 事务 1 mysql如何控制事务 xff1f 2 默认事务开启的作用是什么 xff1f 3 事务的其他打开方式 xff1f 二 事物的四大特性 一 事务 在mysql中 xff0c 事务其实是一个最小的不可分割的工作单元 xff0

  • Unity项目 - DeathtrapDungeon死亡地牢

    目录 游戏原型项目演示绘图资源代码实现注意事项技术探讨参考来源 游戏原型 死亡地牢是一款 2D Roguelike 的地牢冒险游戏 手握利刃 xff0c 斩杀怪物 xff0c 在凶险的地牢内生存下去 但注意 xff0c 敌人也并非善茬 xf

  • Unity - 存读档机制简析

    本文旨在于简要分析Unity中的两种存档机制 xff0c 即 xff1a PlayerPrefs数据持久化方法及Serialization数据序列化方法 较比于源项目 xff0c 我另加了JSON方法 XML方法等及一些Unity设置 xf

  • Windows 无法安装到所选位置。错误：0x80300001

    Windows 无法安装到所选位置 错误 xff1a 0x80300001 这里遇到的情况是这样的 xff0c iDrac安装windows 2008 R2 xff0c 一开始映射 windows 2008 R2系统镜像 xff0c 后来

  • 【udacity】机器学习-2模型验证

    Evernote Export 1 模型的评估与验证简介 机器学习通常是大量传入数据 xff0c 然后会有一些关于数据的决策 想法和摘要 2 模型评估 评估模型使用的是各种数据分析的方法 xff0c 至少需要使用python编程和一些统计学

  • C++编程（五）--- Cmake详解&Makefile详解

    C C 43 43 程序员肯定离不开Makefile和Cmake xff0c 因为如果对这两个工具不熟悉 xff0c 那么你就不是一个合格的C C 43 43 程序员 本文对Makefile和Cmake xff0c 及它们的使用进行了详细的

  • 【统计学】第四章

    Evernote Export 一组数据的分布特征可以从那几个方面进行测度 xff1f 数据的分布特征可以从三个方面进行测度和描述 xff0c 一是分布的集中趋势 xff0c 反映各数据向其中心值靠拢或聚集的程度 xff1b 二是分布的离散

  • UG NX安装包大集合（包括UG目前发布的所有版本）

    UG NX安装包大集合 xff08 包括UG目前发布的所有版本 xff09 UG爱好者官方交流群 216953883 有了这个你就不怕找UG安装包麻烦了 xff0c 现在所有安装包全在这里了 所有版本的补丁包也在年后陆续更新 提醒 xff1

  • HTML常用字体代码

    HTML常用字体代码 常用字体 lt FONT style 61 34 FONT SIZE 40pt FILTER shadow color 61 green WIDTH 100 COLOR white LINE HEIGHT 150 FO

  • 线性链式存储结构c语言建立,线性表的链式存储结构（C语言版）

    上一篇博文我对数据结构中线性表的顺序存储结构顺序表 http 12172969 blog 51cto com 12162969 1916336 按照我的理解做了总结 xff0c 今天我继续对顺序表的另一种存储结构 xff0c 链表谈一下我看

  • stract oracle,ORACLE 字符串聚合函数 strCat

    源码如下 xff1a create or replace type strcat type as object currentstr varchar2 4000 currentseprator varchar2 8 static funct

  • ARM存储格式之 大端小端

    开头讲个有关大端小端的故事 xff1a 端模式 xff08 Endian xff09 的这个词出自Jonathan Swift书写的 格列佛游记 这本书根据将鸡蛋敲开的方法不同将所有的人分为两类 xff0c 从圆头开始将鸡蛋敲开的人被归为B

  • 多版本opencv管理; find_package()的原理解析

    近期用cmake编译程序时 xff0c 报错找不到opencv2 由于我电脑里安装了多个版本的opencv xff0c 管理不善 xff0c 借此机会梳理一下思路 1 Cmake find package Opencv REQUIRED x

  • 解决 Flask 项目无法用 .env 文件中解析的参数设置环境变量的错误

    在 Windows 上启动 Flask 项目时 xff0c 工作目录有 UTF 8 编码的 env 文件 xff0c 里面配置的环境变量在 Python2 中识别为 Unicode 类型 xff0c 导致下述错误 xff1a Serving

  • cordova环境搭建

    一 步骤列表 准备依赖环境 安装cordova 创建app xff0c 并build 二 准备依赖环境 1 需要准备的安装包 说明 xff1a gradle下载后 xff0c 解压到硬盘某个目录即可 xff1b 安装步骤 xff1a jav

  • JavaWeb_(Struts2框架)Ognl小案例查询帖子

    创建paste帖子表 CREATE TABLE 96 strutstest 96 96 paste 96 96 id 96 VARCHAR 50 NOT NULL 96 answer 96 INT NULL 96 offer 96 INT

  • socket(二)--Tcp同步非阻塞式

    socket 二 Tcp同步非阻塞式 文章目录 socket 二 Tcp同步非阻塞式一 简介二 关键类2 1 ServerSocketChannel2 2 SocketChannel2 3 Selector2 4 SelectionKey2

热门标签