Netty实现客户端和服务端通信简单例子

Netty是建立在NIO基础之上，Netty在NIO之上又提供了更高层次的抽象。

在Netty里面，Accept连接可以使用单独的线程池去处理，读写操作又是另外的线程池来处理。

Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理，也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装，构造出满足系统需求的并发模型。

Netty提供了内置的常用编解码器，包括行编解码器［一行一个请求］，前缀长度编解码器［前N个字节定义请求的字节长度］，可重放解码器［记录半包消息的状态］，HTTP编解码器，WebSocket消息编解码器等等

Netty提供了一些列生命周期回调接口，当一个完整的请求到达时，当一个连接关闭时，当一个连接建立时，用户都会收到回调事件，然后进行逻辑处理。

Netty可以同时管理多个端口，可以使用NIO客户端模型，这些对于RPC服务是很有必要的。

Netty除了可以处理TCP Socket之外，还可以处理UDP Socket。

在消息读写过程中，需要大量使用ByteBuffer，Netty对ByteBuffer在性能和使用的便捷性上都进行了优化和抽象。

代码：

服务端：

package com.kinson.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * descripiton:服务端
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:37
 * @modifier:
 * @since:
 */
public class NettyServer {
    /**
     * 端口
     */
    private int port;

    public NettyServer(int port) {
        this.port = port;
    }

    public void run() {
        //EventLoopGroup是用来处理IO操作的多线程事件循环器
        //负责接收客户端连接线程
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //负责处理客户端i/o事件、task任务、监听任务组
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //启动 NIO 服务的辅助启动类
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        //配置 Channel
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ServerIniterHandler());
        //BACKLOG用于构造服务端套接字ServerSocket对象，
        // 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        //是否启用心跳保活机制
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        try {
            //绑定服务端口监听
            Channel channel = bootstrap.bind(port).sync().channel();
            System.out.println("server run in port " + port);
            //服务器关闭监听
            /*channel.closeFuture().sync()实际是如何工作:
            channel.closeFuture()不做任何操作，只是简单的返回channel对象中的closeFuture对象，对于每个Channel对象，都会有唯一的一个CloseFuture，用来表示关闭的Future，
            所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法，从上面的解释可以知道，这步是会将当前线程阻塞在CloseFuture上*/
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //关闭事件流组
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        new NettyServer(8899).run();
    }
}

服务端业务逻辑处理：

package com.kinson.netty.server;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * descripiton: 服务器的处理逻辑
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:50
 * @modifier:
 * @since:
 */
public class ServerHandler extends SimpleChannelInboundHandler<String> {

    /**
     * 所有的活动用户
     */
    public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 读取消息通道
     *
     * @param context
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext context, String s)
            throws Exception {
        Channel channel = context.channel();
        //当有用户发送消息的时候，对其他的用户发送消息
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[you]: " + s + "\n");
            } else {
                ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");
            }
        }
        System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");
    }

    /**
     * 处理新加的消息通道
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");
            }
        }
        group.add(channel);
    }

    /**
     * 处理退出消息通道
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");
            }
        }
        group.remove(channel);
    }

    /**
     * 在建立连接时发送消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        boolean active = channel.isActive();
        if (active) {
            System.out.println("[" + channel.remoteAddress() + "] is online");
        } else {
            System.out.println("[" + channel.remoteAddress() + "] is offline");
        }
        ctx.writeAndFlush("[server]: welcome");
    }

    /**
     * 退出时发送消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        if (!channel.isActive()) {
            System.out.println("[" + channel.remoteAddress() + "] is offline");
        } else {
            System.out.println("[" + channel.remoteAddress() + "] is online");
        }
    }

    /**
     * 异常捕获
     *
     * @param ctx
     * @param e
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("[" + channel.remoteAddress() + "] leave the room");
        ctx.close().sync();
    }

}

服务端处理器注册：

package com.kinson.netty.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * descripiton: 服务器初始化
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:46
 * @modifier:
 * @since:
 */
public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //管道注册handler
        ChannelPipeline pipeline = socketChannel.pipeline();
        //编码通道处理
        pipeline.addLast("decode", new StringDecoder());
        //转码通道处理
        pipeline.addLast("encode", new StringEncoder());
        //聊天服务通道处理
        pipeline.addLast("chat", new ServerHandler());
    }
}

客户端：

package com.kinson.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * descripiton: 客户端
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:40
 * @modifier:
 * @since:
 */
public class NettyClient {

    private String ip;

    private int port;

    private boolean stop = false;

    public NettyClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }

    public void run() throws IOException {
        //设置一个多线程循环器
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //启动附注类
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup);
        //指定所使用的NIO传输channel
        bootstrap.channel(NioSocketChannel.class);
        //指定客户端初始化处理
        bootstrap.handler(new ClientIniterHandler());
        try {
            //连接服务
            Channel channel = bootstrap.connect(ip, port).sync().channel();
            while (true) {
                //向服务端发送内容
                BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
                String content = reader.readLine();
                if (StringUtils.isNotEmpty(content)) {
                    if (StringUtils.equalsIgnoreCase(content, "q")) {
                        System.exit(1);
                    }
                    channel.writeAndFlush(content);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new NettyClient("127.0.0.1", 8899).run();
    }
}

客户端逻辑处理：

package com.kinson.netty.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * descripiton: 客户端逻辑处理
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:50
 * @modifier:
 * @since:
 */
public class ClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
        //打印服务端的发送数据
        System.out.println(s);
    }
}

客户端处理器注册：

package com.kinson.netty.client;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * descripiton: 客户端处理初始化
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:55
 * @modifier:
 * @since:
 */
public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //注册管道
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        pipeline.addLast("http", new HttpClientCodec());
        pipeline.addLast("chat", new ClientHandler());
    }
}

测试时先启动服务端，再启动客户端。。。

原文地址：https://www.cnblogs.com/kingsonfu/p/Java.html

时间： 2025-01-13 22:53:57

Netty实现客户端和服务端通信简单例子的相关文章

WCF客户端与服务端通信简单入门教程

服务端 1.新建空白解决方案,然后再空白解决方案中新建:WCF服务应用程序.建完后如图: 2.删掉自动生成的IService1.cs和Service.svc并添加WCF服务文件StudentService.svc,VS会自动生成IStudentService.cs 在添加一个Student类,如图: Student.cs: /// <summary> /// DataContract数据约定,保证student类在WCF调用中被序列化 /// DataMember 在被序列化的成员变量中必须加

Netty入门之客户端与服务端通信(二)

Netty入门之客户端与服务端通信(二) 一.简介 在上一篇博文中笔者写了关于Netty入门级的Hello World程序.书接上回,本博文是关于客户端与服务端的通信,感觉也没什么好说的了,直接上代码吧. 二.客户端与服务端的通信 2.1 服务端启动程序 public class MyServer { public static void main(String[] args) throws InterruptedException { EventLoopGroup bossGroup = ne

警察与小偷的实现之一客户端与服务端通信

来源于ISCC 2012 破解关第四题 目的是通过逆向police,实现一个thief,能够与police进行通信 实际上就是一个RSA加密通信的例子,我们通过自己编写客户端和服务端来实现上面的thief和police的功能．． 要通信,这们这次先通过python写出可以进行网络连接的客户端与服务端．． 服务端代码 #!/usr/bin/env python import SocketServer from time import ctime HOST = '127.0.0.1' PORT =

[Java]命令行模拟TCP客户端与服务端的简单小程序遇到的问题（基础不牢！扎实和亲手实践比什么都重要！）

简单得不能再简单的需求: 简单模拟TCP客户端与服务端的一次连接和通信,客户端发出一个消息,服务端回馈一个消息 自己第一次编写的代码: Client: class TcpClient1 { public static void main(String[] args) throws Exception { Socket s=new Socket("127.0.0.1",10010); OutputStream out=s.getOutputStream(); out.write(&quo

使用多线程实现多个客户端与服务端通信1

Server.java package socket; import java.io.BufferedReader;import java.io.IOException;import java.io.InputStream;import java.io.InputStreamReader;import java.net.InetAddress;import java.net.ServerSocket;import java.net.Socket; /** * 聊天室服务端 * @author 小

Python socket编程客户端与服务端通信

目标:实现客户端与服务端的socket通信,消息传输. 客户端 客户端代码: from socket import socket,AF_INET,SOCK_STREAM #服务端的ip地址 server_ip = '127.0.0.1' #服务端socket绑定的端口号 server_port = 20000 if __name__ == '__main__': while True: str_msg = input("请输入要发送信息:") if str_msg != "&

二、网络编程-socket之TCP协议开发客户端和服务端通信

知识点:之前讲的udp协议传输数据是不安全的,不可靠不稳定的,tcp协议传输数据安全可靠,因为它们的通讯机制是不一样的.udp是用户数据报传输,也就是直接丢一个数据包给另外一个程序,就好比寄信给别人,信丢了你也不知道,tcp传输需要先和服务端建立连接,当客户端与服务器连接时,服务器会给出应答,我俩连上了,而且数据传过来还会进行一个数据包数量验证,不一致会重新发送,还有其他种种验证,总之保证了数据传输安全可靠   这一章主要介绍使用套接字,编写一个tcp协议客户端和服务端.同样要用到上一章节提到小

使用socket实现简单的客户端和服务端通信(C#语言)

1.主要思路: (1) 服务端开启监听线程,等待客户端的连接. 每个socket连接放到独立线程中处理. (2) 服务端和客户端使用约定的消息格式通信.对于比较复杂的消息(如向服务端传递一个实例),可以使用json封装传输. (3) 每个连接的客户端,注册唯一的ClientID,在服务端以此来区分消息的来源. 2.代码构成 实现该样例包含两个cs的客户端程序. 分别为 服务端 和 客户端. 3.主要代码: (1) 服务端 xaml <Window x:Class="WpfApplicati

ssl客户端与服务端通信的demo

服务端程序流程 1 #include <stdio.h> 2 #include <stdlib.h> 3 #include <errno.h> 4 #include <string.h> 5 #include <sys/types.h> 6 #include <netinet/in.h> 7 #include <sys/socket.h> 8 #include <sys/wait.h> 9 #include &