java基于netty NIO的简单聊天室的实现

package nettyserverv1;
 
import java.util.ArrayList;
 
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.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
 
/**
 * Created by linguolong on 2019/05/08.
 * Chatroom server built using netty framework
 */
 
public class SubreqServer {
	//保存已注册的用户信息
	public static ArrayList<UserInfo> userlist = new ArrayList<UserInfo>();
	//自动生成注册用户
	static{
		for(int i=0;i<10;i++){
			UserInfo user=new UserInfo();
			user.setUserID("123"+i);
			user.setUserName("user"+i);
			user.setPassword("pwd"+i);
			userlist.add(user);
		}
	}
	
 public void bind(int port) throws Exception{
  //配置服务端NIO 线程组
  EventLoopGroup boss = new NioEventLoopGroup();
  EventLoopGroup worker = new NioEventLoopGroup();
 
  ServerBootstrap server = new ServerBootstrap();
 
  try {
   server.group(boss, worker)
     .channel(NioServerSocketChannel.class)
     .option(ChannelOption.SO_BACKLOG, 100)
     .childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
       /**
        * 解码器: 构造器传入了两个参数: 
        * #1 单个对象序列化后最大字节长度,这是设置是1M;
        * #2 类解析器: weakCachingConcurrentResolver创建线程安全的WeakReferenceMa对类加载器进行缓存,
        * 支持多线程并发访问,当虚拟机内存不足时,会释放缓存中的内存,防止内存泄漏.
        */
       ch.pipeline().addLast(new ObjectDecoder(1024*1024,
         ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())) );
       ch.pipeline().addLast(new ObjectEncoder());
       ch.pipeline().addLast(new SubreqServerHandler());
      }
     });
 
   System.out.println("Start the server success");
   //绑定端口, 同步等待成功
   ChannelFuture future = server.bind(port).sync();
 
   //等待服务端监听端口关闭
   future.channel().closeFuture().sync();
  } finally {
   //优雅关闭 线程组
   boss.shutdownGracefully();
   worker.shutdownGracefully();
  }
 }
 
 public static void main(String[] args) {
  SubreqServer server = new SubreqServer();
  try {
   server.bind(18888);
  } catch (Exception e) {
   e.printStackTrace();
  }
 }
}

package nettyserverv1;
 
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
 
/**
 * Created by linguolong on 2019/05/08.
 * Chatroom client built using netty framework
 */
 
public class SubreqServerHandler extends SimpleChannelInboundHandler<String>{
	//新建一个channelGroup，用于存放连接的channel
	public static ChannelGroup online_channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	
	@Override
	public void handlerRemoved(ChannelHandlerContext ctx){
  Channel leave_channel = ctx.channel();
  for (Channel channel : online_channels) {
   if (channel != leave_channel){
    channel.writeAndFlush("[用户 " + leave_channel.remoteAddress() + "]下线了!\n");
   }
  }
  System.out.println(ctx.channel().id()+"下线了");
		//把刚下线的channel移除出在线用户队列
		online_channels.remove(leave_channel);
	}
 
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 	//判断用户是否已经登录
 	if(!check_login(ctx)){
 		//未登录则进行登录验证
 		check_identity(ctx,msg);
 	}else{
 		//已登录则进行消息转发
   Channel coming_channel = ctx.channel();
   for (Channel channel : online_channels) {
    if (channel != coming_channel){
     channel.writeAndFlush("[用户 " + coming_channel.remoteAddress() + "]: " + msg );
    } else {
     channel.writeAndFlush("[我]: " + msg);
    }
   }
 	}
 	
 }
 
 //用户信息验证，检查用户ID和密码是否正确
 public void check_identity(ChannelHandlerContext ctx, Object msg){
 	UserInfo req = (UserInfo) msg;
  System.out.println("service receive client login req :{"+ req.toString() +"}");
  boolean login_flag = false;
  for(int i=0;i<SubreqServer.userlist.size();i++){
  	 if( SubreqServer.userlist.get(i).getUserID().equalsIgnoreCase(req.getUserID())&&(SubreqServer.userlist.get(i).getPassword().equals(req.getPassword()))){
     login_flag=true;
    }
  }
  if(login_flag){
  	System.out.println("账号"+req.getUserID()+"登录成功");
   ctx.writeAndFlush("您已登录成功~\n");
   //将当前的通道加入在线队列中
   online_channels.add(ctx.channel());
  }
  else{
  	System.out.println("账号"+req.getUserID()+"登录失败");
   ctx.writeAndFlush("登录失败!");
   //关闭连接
  	ctx.close();
  	online_channels.remove(ctx.channel());
  }	
 }
 
 //判断用户是否已经在线
 public boolean check_login(ChannelHandlerContext ctx){
 	boolean online_flag = false;
 	for(int i=0;i<online_channels.size();i++){
 		if(online_channels.contains(ctx.channel())){
 			online_flag = true;
 		}
 	}
		return online_flag;
 }
 
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  //释放资源
  ctx.close();
 }
	
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		
	}
 
	//ChannelRead0()只能处理类型为String的消息，因此我们这里不能用ChannelRead0()这个方法,这里的第二个参数类型使用了泛型
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
		
	}
}

package nettyserverv1;
 
import java.io.Serializable;
 
/**
 * Created by linguolong on 2019/05/08.
 * Chatroom User Infomation
 */
 
public class UserInfo implements Serializable{
 private String userID;
 private String userName;
 private String password;
 
 public String getUserID() {
  return userID;
 }
 
 public void setUserID(String userID) {
  this.userID = userID;
 }
 
 public String getPassword() {
  return password;
 }
 
 public void setPassword(String password) {
  this.password = password;
 }
 
 public String getUserName() {
  return userName;
 }
 
 public void setUserName(String userName) {
  this.userName = userName;
 }
 
 @Override
 public String toString() {
  return "SubscribeReq: [messageID]:"+ userID + " [userName]:" +userName
    + " [password]:" +password;
 }
 
}

package nettyclientv1;
 
import java.io.BufferedReader;
import java.io.InputStreamReader;
 
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
 
/**
 * Created by linguolong on 2019/05/08.
 * Chatroom client built using netty framework
 */
 
public class SubreqClient {
 
 public void connect(int port, String host) throws Exception{
  //配置客户端NIO 线程组
  EventLoopGroup group = new NioEventLoopGroup();
  Bootstrap client = new Bootstrap();
 
  try {
   client.group(group)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.TCP_NODELAY, true)
     .handler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
      	//添加解码器
       ch.pipeline().addLast(new ObjectDecoder(1024,
         ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())) );
       //添加编码器
       ch.pipeline().addLast(new ObjectEncoder());
       ch.pipeline().addLast(new SubreqClientHandler());
      }
     });
 
   //异步获取当前已连接的channel
   Channel now_channel = client.connect(host,port).sync().channel();
   BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
   
   //异步等待客户端连接端口关闭
//   now_channel.closeFuture().sync();
   
   //使客户端一直处于输入状态，直到读取到"bye"
   String message = " ";
   while (true) {
   	//读到bye时退出
   	if(message.equals("bye")) break;
   	message = reader.readLine();
    now_channel.writeAndFlush(message+"\n");
   }
   
   //读到了"bye"字符串，主动断开连接
   now_channel.close();
  } finally {
   //优雅关闭 线程组
   group.shutdownGracefully();
  }
 }
 
 
 public static void main(String[] args) {
  SubreqClient client = new SubreqClient();
  try {
   client.connect(18888, "127.0.0.1");
  } catch (Exception e) {
   e.printStackTrace();
  }
 }
}

package nettyclientv1;
 
import nettyserverv1.UserInfo;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
 
/**
 * Created by linguolong on 2019/05/08.
 * Chatroom client built using netty framework
 */
 
public class SubreqClientHandler extends ChannelInboundHandlerAdapter{
 
 public SubreqClientHandler() {
 }
 
 /**
  * @param ctx
  * @throws Exception
  */
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 	ctx.writeAndFlush(subReq("1231","user1","pwd1"));
  ctx.flush();
 }
 
	private UserInfo subReq(String id,String userName,String password){
  UserInfo req = new UserInfo();
  req.setUserID(id);
  req.setUserName(userName);
  req.setPassword(password);
 
  return req;
 }
 
 
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 	System.out.print(msg.toString());
 
 }
 
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  ctx.close();
 }
}