网络I/o编程模型16 netty框架实现的群聊系统

Posted 2022-06-06 健康平安的活着

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了网络I/o编程模型16 netty框架实现的群聊系统相关的知识，希望对你有一定的参考价值。

一背景描述

1.编写一下群聊系统：实现服务器端和客户端之间数据通讯（非阻塞模式）

服务端：可以检测用户上线，离线，并实现消息转发功能。

客户端：通过channel可以无阻塞发送消息给其他所用用户，同时可以接受其他用户发送的消息（有服务器转发得到）

2.采用思路：使用netty的非阻塞网络机制

二代码实现

2.1 服务端代码

1.server

package com.ljf.netty.netty.groupchat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioserverSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * @ClassName: NettyGroupchatServer
 * @Description: TODO
 * @Author: liujianfu
 * @Date: 2022/06/03 17:46:02
 * @Version: V1.0
 **/
public class NettyGroupchatServer 

    //监听端口
    private   int port;

    public NettyGroupchatServer(int port) 
        this.port = port;
    
    //处理客户端的请求
    public  void  dealHandler()
        //创建两个线程组
        EventLoopGroup bossGroup=new NioEventLoopGroup(1);
        EventLoopGroup workerGroup=new NioEventLoopGroup();//8个NioEventLoop
        try 
        ServerBootstrap b=new ServerBootstrap();
        b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG,128)
                .childOption(ChannelOption.SO_KEEPALIVE,true)
                .childHandler(new ChannelInitializer<SocketChannel>() 
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception 
                        //获取到pipeline
                        ChannelPipeline pipeline=ch.pipeline();
                        //向pipeline加入解码器
                        pipeline.addLast("decoder",new StringDecoder());
                        //向peipeline加入编码器
                        pipeline.addLast("encoder",new StringEncoder());
                        //加入自己的业务处理handler
                        pipeline.addLast(new NettyGroupchatServerHandler());
                    
                );
            System.out.println("netty 服务器启动成功!!!");
        ChannelFuture channelFuture = b.bind(port).sync();
        //监听关闭
        channelFuture.channel().closeFuture().sync();
         catch (InterruptedException e) 
            e.printStackTrace();
        
        finally 
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        
    

    public static void main(String[] args) 
        new NettyGroupchatServer(6666).dealHandler();

2.自定义服务端

package com.ljf.netty.netty.groupchat;

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;


import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @ClassName: NettyGroupchatServerHandler
 * @Description: TODO
 * @Author: liujianfu
 * @Date: 2022/06/03 18:31:21
 * @Version: V1.0
 **/
public class NettyGroupchatServerHandler extends SimpleChannelInboundHandler<String> 
    //定义一个channel组，管理所有的channel,GlobalEventExecutor.INSTANCE是全局的事件执行器，是一个单例
    private static ChannelGroup channelGroup=new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    //时间格式化器
    SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     //建立连接连接，将当前channel加入到channelGroup
    public void handlerAdded(ChannelHandlerContext ctx)
        Channel channel=ctx.channel();
        channelGroup.writeAndFlush(" 【客户端:】"+channel.remoteAddress()+" 进入聊天"+sdf.format(new Date())+"\\n");
        channelGroup.add(channel);
    
    //断开连接，将xx客户离开信息推送给当前在线客户

    public   void handlerRemoved(ChannelHandlerContext ctx)
        Channel channel=ctx.channel();
        channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+" 离开了");
        System.out.println("");
        System.out.println("channelGroup size"+channelGroup.size());
    
     //表示channel处于活动状态，提示xx上线
    public  void channelActive(ChannelHandlerContext ctx)
        System.out.println(""+ctx.channel().remoteAddress()+" 上线了~~~~");
    
    //表示channel处于不活动状态，提示xx离线了
    public void channelInactive(ChannelHandlerContext cxt)
     System.out.println(""+cxt.channel().remoteAddress()+" 离线了~~~~~");
    
   //读取数据
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception 
       //获取当前的channel
        Channel channel=ctx.channel();
        //这时我们遍历channelGroup，根据不同情况，回送不同的消息
        channelGroup.forEach(ch ->
            if(channel!=ch)//不是当前的chennel,转发消息
                ch.writeAndFlush(" 【客户】"+channel.remoteAddress()+" 发送了消息:["+s+"] \\n");
            else//回显自己发送的消息给自己
                ch.writeAndFlush("[自己] 发送了消息:"+s+"\\n");
            
        );
    
    //异常处理
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause)
        //关闭通道
        ctx.close();

2.2 客户端代码

1.客户端

package com.ljf.netty.netty.groupchat;

import com.ljf.netty.netty.tcp.NettyTcpClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;

/**
 * @ClassName: NettyGroupchatClient
 * @Description: TODO
 * @Author: liujianfu
 * @Date: 2022/06/03 19:05:28
 * @Version: V1.0
 **/
public class NettyGroupchatClient 
    //属性
    private final String host;
    private  final int port;

    public NettyGroupchatClient(String host, int port) 
        this.host = host;
        this.port = port;
    
    public void runDeal()
        EventLoopGroup group=new NioEventLoopGroup();
        try 
        Bootstrap bootstrap=new Bootstrap();
        //设置相关参数
        bootstrap.group(group) //设置线程组
                .channel(NioSocketChannel.class) //设置客户端通道的实现类(反射)
                .handler(new ChannelInitializer<SocketChannel>() 
                    protected void initChannel(SocketChannel ch)
                      ChannelPipeline pipeline= ch.pipeline();
                        pipeline.addLast("decoder",new StringDecoder());
                        pipeline.addLast("encoder",new StringEncoder());
                        pipeline.addLast(new NettyGroupchatClientHandler());//加入自己的处理器
                    
                );
        System.out.println("客户端 is ok.....");
        //启动客户端去链接服务器端，channelfuture，涉及道netty的异步模型
            ChannelFuture channelFuture=bootstrap.connect(host,port).sync();
         //得到channel
            Channel channel=channelFuture.channel();
            System.out.println("------"+channel.localAddress()+"------");
            //客户端需要输入信息，创建一个扫描器
            Scanner scanner=new Scanner(System.in);
            while(scanner.hasNextLine())
                String msg=scanner.nextLine();
                //通过channel发送到服务器端
                channel.writeAndFlush(msg+"\\r\\n");
            

         catch (InterruptedException e) 
            e.printStackTrace();
        
        finally 
            group.shutdownGracefully();
        

    

    public static void main(String[] args) 
        new NettyGroupchatClient("127.0.0.1",6666).runDeal();

2.客户端自定义处理

package com.ljf.netty.netty.groupchat;

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

/**
 * @ClassName: NettyGroupchatClientHandler
 * @Description: TODO
 * @Author: liujianfu
 * @Date: 2022/06/03 19:17:46
 * @Version: V1.0
 **/
public class NettyGroupchatClientHandler extends SimpleChannelInboundHandler<String> 
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception 
        System.out.println("客户单 handler发送的消息s:"+s.trim());