JAVA Netty实现聊天室+私聊功能的示例代码

功能介绍

使用Netty框架实现聊天室功能，服务器可监控客户端上下限状态，消息转发。同时实现了点对点私聊功能。技术点我都在代码中做了备注，这里不再重复写了。希望能给想学习netty的同学一点参考。

服务器代码

服务器入口代码

package nio.test.netty.groupChat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;


public class NettyChatServer {
	private int port;
	
	public NettyChatServer(int port){
		this.port = port;
	}
	
	//初始化 netty服务器
	private void init() throws Exception{
		EventLoopGroup boss = new NioEventLoopGroup(1);
		EventLoopGroup work = new NioEventLoopGroup(16);
		try {
			ServerBootstrap boot = new ServerBootstrap();
			boot.group(boss,work);
			boot.channel(NioServerSocketChannel.class);//设置boss selector建立channel使用的对象
			boot.option(ChannelOption.SO_BACKLOG, 128);//boss 等待连接的 队列长度
			boot.childOption(ChannelOption.SO_KEEPALIVE, true); //让客户端保持长期活动状态
			boot.childHandler(new ChannelInitializer() {

				@Override
				protected void initChannel(SocketChannel ch) throws Exception {
					//从channel中获取pipeline 并往里边添加Handler
					ChannelPipeline pipeline = ch.pipeline();
					pipeline.addLast("encoder",new StringEncoder());
					pipeline.addLast("decoder",new StringDecoder());
					pipeline.addLast(new ServerMessageHandler());//自定义Handler来处理消息
				}
			});
			System.out.println("服务器开始启动...");
			//绑定端口 
			ChannelFuture channelFuture = boot.bind(port).sync();
			channelFuture.addListener(new GenericFutureListener>() {

				@Override
				public void operationComplete(Future future)
						throws Exception {
					if(future.isSuccess()){
						System.out.println("服务器正在启动...");
					}
					if(future.isDone()){
						System.out.println("服务器启动成功...OK");
					}
					
				}
			});
			//监听channel关闭
			channelFuture.channel().closeFuture().sync();
			channelFuture.addListener(new GenericFutureListener>() {

				@Override
				public void operationComplete(Future future)
						throws Exception {
					if(future.isCancelled()){
						System.out.println("服务器正在关闭..");
					}
					if(future.isCancellable()){
						System.out.println("服务器已经关闭..OK");
					}
					
				}
			});
			
		}finally{
			boss.shutdownGracefully();
			work.shutdownGracefully();
		}
	}
	
	public static void main(String[] args) throws Exception {
		new NettyChatServer(9090).init();

	}

}

服务器端消息处理Handler

package nio.test.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;
import java.util.HashMap;
import java.util.Map;

public class ServerMessageHandler extends SimpleChannelInboundHandler{
	
	private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	
	private static Map all = new HashMap();
	
	private SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg)
			throws Exception {
		Channel channel = ctx.channel();
		
		if(msg.contains("#")){
			String id = msg.split("#")[0];
			String body = msg.split("#")[1];
			Channel userChannel = all.get(id);
			String key = channel.remoteAddress().toString().split(":")[1];
			userChannel.writeAndFlush(sf.format(new Date())+"n 【用户】 "+key+" 说 : "+body);
			return;
		}
		
		//判断当前消息是不是自己发送的
		for(Channel c : channels){
			String addr = c.remoteAddress().toString();
			if(channel !=c){
				c.writeAndFlush(sf.format(new Date())+"n 【用户】 "+addr+" 说 : "+msg);
			}else{
				c.writeAndFlush(sf.format(new Date())+"n 【自己】 "+addr+" 说 : "+msg);
			}
		}
		
	}
	
	@Override
	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
		Channel channel = ctx.channel();
		String addr = channel.remoteAddress().toString();
		
		channels.writeAndFlush(sf.format(new Date())+"n 【用户】 "+addr+" 加入聊天室 ");
		channels.add(channel);
		String key = channel.remoteAddress().toString().split(":")[1];
		all.put(key, channel);
	}
	
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		String addr = ctx.channel().remoteAddress().toString();
		System.out.println(sf.format(new Date())+" n【用户】 "+addr+" 上线 ");
	}
	
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		String addr = ctx.channel().remoteAddress().toString();
		System.out.println(sf.format(new Date())+" n【用户】 "+addr+" 下线 ");
		//下线移除
		String key = ctx.channel().remoteAddress().toString().split(":")[1];
		all.remove(key);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		//System.out.println("连接发生异常！");
		ctx.close();
	}
	
	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
		Channel channel = ctx.channel();
		String addr = channel.remoteAddress().toString();
		
		channels.writeAndFlush(sf.format(new Date())+"n 【用户】 "+addr+" 离开了 ");
		//打印 ChannelGroup中的人数
		System.out.println("当前在线人数是:"+channels.size());
		System.out.println("all："+all.size());
	}

	
}

客户端主方法代码

package nio.test.netty.groupChat;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.Scanner;

public class NettyChatClient {

	private String ip;
	
	private int port;
	
	public NettyChatClient(String ip,int port){
		this.ip = ip;
		this.port = port;
	}
	
	private void init() throws Exception{
		//创建监听事件的监听器
		EventLoopGroup work = new NioEventLoopGroup();
		try {
			Bootstrap boot = new Bootstrap();
			boot.group(work);
			boot.channel(NioSocketChannel.class);
			boot.handler(new ChannelInitializer() {

				@Override
				protected void initChannel(NioSocketChannel ch)
						throws Exception {
					ChannelPipeline pipeline = ch.pipeline();
					pipeline.addLast("encoder",new StringEncoder());
					pipeline.addLast("decoder",new StringDecoder());
					pipeline.addLast(new ClientMessageHandler());
					
				}
			});
			
			ChannelFuture channelFuture = boot.connect(ip, port).sync();
			channelFuture.addListener(new GenericFutureListener>() {

				@Override
				public void operationComplete(Future future)
						throws Exception {
					if(future.isSuccess()){
						System.out.println("客户端启动中...");
					}
					if(future.isDone()){
						System.out.println("客户端启动成功...OK！");
					}
				}
			});
			System.out.println(channelFuture.channel().localAddress().toString());
			System.out.println("#################################################");
			System.out.println("~~~~~~~~~~~~~~端口号#消息内容~~这样可以给单独一个用户发消息~~~~~~~~~~~~~~~~~~");
			System.out.println("#################################################");
			
			
			Channel channel = channelFuture.channel();
			//获取channel
			Scanner scanner = new Scanner(System.in);
			while(scanner.hasNextLine()){
				String str = scanner.nextLine();
				channel.writeAndFlush(str+"n");
			}
			channelFuture.channel().closeFuture().sync();
			scanner.close();
		} finally {
			work.shutdownGracefully();
		}
	}
	
	
	public static void main(String[] args) throws Exception{

		new NettyChatClient("127.0.0.1",9090).init();
	}

}

客户端消息处理Handler

package nio.test.netty.groupChat;

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

public class ClientMessageHandler extends SimpleChannelInboundHandler {

	
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg)
			throws Exception {
		System.out.println(msg);
		
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		ctx.close();
		
	}
}

测试结果

启动了四个客户端服务器端日志效果如下：

客户端一端日志：

客户端二日志：

客户端三日志：

客户端四日志：

现在在客户端四发送消息：

每个客户端都可以收到消息：

软化关闭客户端客户端三：

服务器日志：

其他客户端日志：

发送私聊消息：

这个客户端收不到消息

到此这篇关于JAVA Netty实现聊天室+私聊功能的示例代码的文章就介绍到这了,更多相关JAVA Netty聊天室内容请搜索考高分网以前的文章或继续浏览下面的相关文章希望大家以后多多支持考高分网！

JAVA Netty实现聊天室+私聊功能的示例代码

Java相关栏目本月热门文章