生产上使用自定义线程池控制socket短连接需要注意的事项

    欢迎大家搜索“小猴子的技术笔记”关注我的公众号,有问题可以及时和我交流。

    对于一些平台业务对接除了http还有就是socket用的比较多了,特别软件是跟硬件设备做对接。在并发要求不是很高的时候,可以使用BIO的socket,因为它足够简单,足够方便。如果并发量很高的话,推荐使用NIO的socket,因为它能够支持较高的并发。

    如果使用的是长连接的话,那么需要注意的是长连接的心跳保持,客户端的断线重连,报文格式的定义等。

    如果可以的话,还是建议使用netty这个强大的框架,因为它对各种协议的封装还是很到位的。如果你熟悉它的API的话,那么这里真的强烈推荐netty(后期我会更新学习笔记)。

    言归正传,接下来聊一聊BIO-短连接需要注意的事项:如果短连接作为客户端是比较简单的,只用connect->recv/send->close的流程接收数据发送数据,有业务请求来了,进行处理即可,下面给出了代码范例(以下代码均可以在gitee中找到:https://gitee.com/MonkeyBrothers/study/tree/master/springboot-netty/src/main/java/org/hry/socket/bio/shortlink):

public class SocketClient {
    public static void main(String[] args) throws IOException {
        Socket socket = new Socket("localhost", 8888);
        byte[] bytes = new byte[1024];
        InputStream in = socket.getInputStream();
        StringBuffer sb = new StringBuffer();
        int len;
        while ((len = in.read(bytes)) != -1) {
            sb.append(new String(bytes, 0, len));
        }
        // 接收到信息,拿到接收的信息之后就可以处理了。
        System.out.println("接收到消息:" + sb.toString());
        // 关闭输入流,告诉服务端我读取数据完毕了
        socket.shutdownInput();
        // 如果需要给socketServer返回处理之后的数据,可以拿到输出流进行输出。
        OutputStream out = socket.getOutputStream();
        out.write("这里是处理完毕的数据".getBytes(StandardCharsets.UTF_8));
        // 关闭输出流,告诉服务端我写完数据了
        socket.shutdownOutput();
        System.out.println("数据发送完毕");
        socket.close();
    }
}

    连接上服务端,连接上之后,处理数据,处理完毕之后关闭这个socket进行释放资源就可以了。

    BIO-服务端:如果是服务端那就 需要注意的东西比较多了,第一个就是并发量很大的很大的时候千万不要使用BIO 的socket更建议使用NIO或者netty框架来帮助你解决业务问题。首先来看一个类似一问一答式BIO的socket服务。


public class SocketServer {
    public static void main(String[] args) throws IOException {
        ServerSocket server = new ServerSocket(8888);
        Socket socket = server.accept();
        byte[] bytes = new byte[1024];
        InputStream in = socket.getInputStream();
        OutputStream out = socket.getOutputStream();
        out.write("这里是服务端发送的数据".getBytes(StandardCharsets.UTF_8));
        // 告诉客户端,我已经发送完一部分数据了。如果不调用的话,客户端是接受不到输入信号的
        socket.shutdownOutput();
        StringBuffer sb = new StringBuffer();
        int len;
        while ((len = in.read(bytes)) != -1) {
            sb.append(new String(bytes, 0, len));
        }
        // 接收到信息,拿到接收的信息之后就可以处理了。
        System.out.println("接收到消息:" + sb.toString());
        socket.shutdownInput();
        server.close();
    }
}

    这里是有问题的:就是只能有一个socket client来连接,如果是并发的业务是不能够满足需求的,因此需要为每一个socket启动一个线程。但是这样是很消耗资源的。所以并不是并发量特别大的话,还是可以的。再强调一下,超过一百的并发就不要使用BIO了!切记!!!

    接下来模拟一下低并发的socket请求,看一下客户端和服务端是怎么处理的。首先客户端使用了while循环,这里“TimeUnit.SEONDS.sleep(1)”进行了休眠1秒,也就是每一秒都来一个请求(如果要每秒请求100的话,可以使用CountDownLatch进行控制,文章最后有实例)。

public class SocketClient {
    public static void main(String[] args) throws Exception {
        while (true) {
            TimeUnit.SECONDS.sleep(1);
            Socket socket = new Socket("localhost", 8888);
            byte[] bytes = new byte[1024];
            InputStream in = socket.getInputStream();
            StringBuffer sb = new StringBuffer();
            int len;
            while ((len = in.read(bytes)) != -1) {
                sb.append(new String(bytes, 0, len));
            }
            // 接收到信息,拿到接收的信息之后就可以处理了。
            System.out.println("接收到消息:" + sb.toString());
            // 关闭输入流,告诉服务端我读取数据完毕了
            socket.shutdownInput();
            // 如果需要给socketServer返回处理之后的数据,可以拿到输出流进行输出。
            OutputStream out = socket.getOutputStream();
            out.write("这里是处理完毕的数据".getBytes(StandardCharsets.UTF_8));
            // 关闭输出流,告诉服务端我写完数据了
            socket.shutdownOutput();
            System.out.println("数据发送完毕");
            socket.close();
        }
    }
}

    对应的server服务就需要使用到线程池了,java为我们默认提供了一些线程池。

在这里插入图片描述

    虽然我在刚开始写socket的时候使用了JDK提供的线程池,但是后来对线程池的深入研究之后发现一些潜在问题。因此还是推荐使用自定的线程池,这样一来可以对参数进行把控,还可以对线程池进行自定义设置名称。

public class SocketServer {
    public static void main(String[] args) throws IOException {
        Executor executor = Executors.newCachedThreadPool();
        ServerSocket server = new ServerSocket(8888);
        while(true) {
            Socket socket = server.accept();
            executor.execute(() ->{
                try(InputStream in = socket.getInputStream();OutputStream out = socket.getOutputStream();){
                    byte[] bytes = new byte[1024];
                    out.write("这里是服务端发送的数据".getBytes(StandardCharsets.UTF_8));
                    // 告诉客户端,我已经发送完一部分数据了。如果不调用的话,客户端是接受不到结束信号的
                    socket.shutdownOutput();
                    StringBuffer sb = new StringBuffer();
                    int len;
                    while ((len = in.read(bytes)) != -1) {
                        sb.append(new String(bytes, 0, len));
                    }
                    // 接收到信息,拿到接收的信息之后就可以处理了。
                    System.out.println("接收到消息:" + sb.toString());
                    socket.shutdownInput();
                    socket.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

    自定义的线程池,我的生产上在使用,具体的参数可以根据自己的业务进行设置。

public class SocketServer {
    public static void main(String[] args) throws IOException {
        // 自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                10,
                10,
                5L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(10),
                new MySocketThreadFactory("自定义socket线程池"),
                new ThreadPoolExecutor.DiscardPolicy());
        ServerSocket server = new ServerSocket(8888);
        while(true) {
            Socket socket = server.accept();
            executor.execute(() ->{
                try(InputStream in = socket.getInputStream();OutputStream out = socket.getOutputStream();){
                    byte[] bytes = new byte[1024];
                    out.write("这里是服务端发送的数据".getBytes(StandardCharsets.UTF_8));
                    // 告诉客户端,我已经发送完一部分数据了。如果不调用的话,客户端是接受不到结束信号的
                    socket.shutdownOutput();
                    StringBuffer sb = new StringBuffer();
                    int len;
                    while ((len = in.read(bytes)) != -1) {
                        sb.append(new String(bytes, 0, len));
                    }
                    // 接收到信息,拿到接收的信息之后就可以处理了。
                    System.out.println("接收到消息:" + sb.toString());
                    socket.shutdownInput();
                    socket.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

    自定义线程工厂生产可用。

public class MySocketThreadFactory implements ThreadFactory {
    /**
     * 线程组
     */
    private final ThreadGroup threadGroup;
    /**
     * 线程数量初始值,这里只是展示使用
     */
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    /**
     * 线程名前缀
     */
    public final String namePrefix;

    public MySocketThreadFactory(String name) {
        SecurityManager s = System.getSecurityManager();
        threadGroup = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        if (null == name || "".equals(name.trim())) name = "pool";
        // 线程池数量初始值,这里只是展示使用
        AtomicInteger poolNumber = new AtomicInteger(1);
        namePrefix = name + "-" + poolNumber.getAndIncrement() + "-socket-thread-";
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(threadGroup, r, namePrefix + threadNumber.getAndIncrement(), 0);
        if (t.isDaemon()) t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

    以上的BIO的socket是生产上可以使用的,在并发量不高的情况下可以直接拿去使用。再一次强调,高并发请勿使用线程的socket!

    这里给了并发的测试代码:


public class SocketClient {
    public static void main(String[] args) throws Exception {
        CountDownLatch downLatch = new CountDownLatch(1);
        IntStream.range(0,25).forEach(v -> new Thread(() ->{
            try {
                downLatch.await();
                Socket socket = new Socket("localhost", 8888);
                byte[] bytes = new byte[1024];
                InputStream in = socket.getInputStream();
                StringBuffer sb = new StringBuffer();
                int len;
                while ((len = in.read(bytes)) != -1) {
                    sb.append(new String(bytes, 0, len));
                }
                // 接收到信息,拿到接收的信息之后就可以处理了。
                System.out.println("接收到消息:" + sb.toString());
                // 关闭输入流,告诉服务端我读取数据完毕了
                socket.shutdownInput();
                // 如果需要给socketServer返回处理之后的数据,可以拿到输出流进行输出。
                OutputStream out = socket.getOutputStream();
                out.write("这里是处理完毕的数据".getBytes(StandardCharsets.UTF_8));
                // 关闭输出流,告诉服务端我写完数据了
                socket.shutdownOutput();
                System.out.println("数据发送完毕");
                socket.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start());
        downLatch.countDown();
    }
}

    并发过来就不行,这个时候就需要调整线程池的设置了。由此可见BIO的socket的中等并发使用是很依赖线程池的。

    对于socket阻塞式短连接的使用到此就结束了。需要注意的就是短连接是阻塞式的,因此如果一个业务有问题的话,那么就会阻塞其余的业务。因此开辟了线程池来进行处理这个问题,但是它不适用于高并发的场景,在高并发的场景下建议使用NIO的socket或者netty框架来帮助完成业务!

    最后欢迎大家添加我的微信号“小猴子的技术笔记”,如果有问题可以及时和我交流。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 225,132评论 6 523
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 96,457评论 3 404
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 172,411评论 0 368
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 61,147评论 1 301
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 70,145评论 6 400
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 53,611评论 1 315
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 41,962评论 3 429
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 40,948评论 0 279
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 47,479评论 1 324
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 39,514评论 3 347
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 41,640评论 1 355
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 37,228评论 5 351
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 42,973评论 3 340
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 33,402评论 0 25
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 34,551评论 1 277
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 50,210评论 3 381
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 46,714评论 2 366

推荐阅读更多精彩内容