netty群聊系统
实例要求
1) 编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2) 实现多人群聊
3) 服务器端：可以监测用户上线，离线，并实现消息转发功能
4) 客户端：通过 channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发
得到)
服务端
1、GroupChatServer
package com.atguigu.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.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class GroupChatServer { private int port; //监听端口 public GroupChatServer(int port) { this.port = port; } //编写run方法，处理客户端的请求 public void run() throws Exception{ //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) //异步的服务器端 TCP Socket 连接 .channel(NioServerSocketChannel.class) //用来给 ServerChannel 添加配置 //SO_BACKLOG对应TCP/IP协议listen函数中的backlog参数，用来初始化服务器可连接队列大小。 .option(ChannelOption.SO_BACKLOG, 128) //用来给接收到的通道添加配置 //SO_KEEPALIVE，一直保持连接活动状态 .childOption(ChannelOption.SO_KEEPALIVE, true) //该方法用来设置业务处理类（自定义的 handler） .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //获取到pipeline ChannelPipeline pipeline = ch.pipeline(); //向pipeline加入解码器 pipeline.addLast("decoder", new StringDecoder()); //向pipeline加入编码器 pipeline.addLast("encoder", new StringEncoder()); //加入自己的业务处理handler pipeline.addLast(new GroupChatServerHandler()); } }); System.out.println("netty 服务器启动"); //该方法用于服务器端，用来设置占用的端口号。等待异步操作执行完毕 ChannelFuture channelFuture = b.bind(port).sync(); //监听关闭 channelFuture.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new GroupChatServer(7000).run(); } }
2、GroupChatServerHandler
package com.atguigu.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.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> { //public static List<Channel> channels = new ArrayList<Channel>(); //使用一个hashmap 管理 //public static Map<String, Channel> channels = new HashMap<String,Channel>(); //定义一个channle 组，管理所有的channel //GlobalEventExecutor.INSTANCE) 是全局的事件执行器，是一个单例 private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); //handlerAdded 表示连接建立，一旦连接，第一个被执行 //将当前channel 加入到 channelGroup @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //将该客户加入聊天的信息推送给其它在线的客户端 /* 该方法会将 channelGroup 中所有的channel 遍历，并发送 消息， 我们不需要自己遍历 */ channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天" + sdf.format(new java.util.Date()) + " "); channelGroup.add(channel); } //断开连接, 将xx客户离开信息推送给当前在线的客户 @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了 "); System.out.println("channelGroup size" + channelGroup.size()); } //表示channel 处于活动状态, 提示 xx上线 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress() + " 上线了~"); } //表示channel 处于不活动状态, 提示 xx离线了 @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress() + " 离线了~"); } //读取数据 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { //获取到当前channel Channel channel = ctx.channel(); //这时我们遍历channelGroup, 根据不同的情况，回送不同的消息 channelGroup.forEach(ch -> { if(channel != ch) { //不是当前的channel,转发消息 ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + " "); }else {//回显自己发送的消息给自己 ch.writeAndFlush("[自己]发送了消息" + msg + " "); } }); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭通道 ctx.close(); } }
客户端
1、GroupChatClient
package com.atguigu.netty.groupchat; 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; public class GroupChatClient { //属性 private final String host; private final int port; public GroupChatClient(String host, int port) { this.host = host; this.port = port; } public void run() throws Exception{ EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //得到pipeline ChannelPipeline pipeline = ch.pipeline(); //加入相关handler pipeline.addLast("decoder", new StringDecoder()); pipeline.addLast("encoder", new StringEncoder()); //加入自定义的handler pipeline.addLast(new GroupChatClientHandler()); } }); 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 + " "); } }finally { group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new GroupChatClient("127.0.0.1", 7000).run(); } }
2、GroupChatClientHandler
package com.atguigu.netty.groupchat; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> { @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg.trim()); } }
运行效果
netty心跳检测应用
实例要求
1) 编写一个 Netty 心跳检测机制案例, 当服务器超过 3 秒没有读时，就提示读空闲
2) 当服务器超过 5 秒没有写操作时，就提示写空闲
3) 实现当服务器超过 7 秒没有读或者写操作时，就提示读写空闲
服务端
1、MyServer
package com.atguigu.netty.heartbeat; import io.netty.bootstrap.ServerBootstrap; 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.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.IdleStateHandler; import java.util.concurrent.TimeUnit; public class MyServer { public static void main(String[] args) throws Exception{ //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup); serverBootstrap.channel(NioServerSocketChannel.class); serverBootstrap.handler(new LoggingHandler(LogLevel.INFO)); serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入一个netty 提供 IdleStateHandler /* 说明 1. IdleStateHandler 是netty 提供的处理空闲状态的处理器 2. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接 3. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接 4. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接 5. 文档说明 triggers an {@link IdleStateEvent} when a {@link Channel} has not performed * read, write, or both operation for a while. * 6. 当 IdleStateEvent 触发后 , 就会传递给管道 的下一个handler去处理 * 通过调用(触发)下一个handler 的 userEventTiggered , 在该方法中去处理 IdleStateEvent(读空闲，写空闲，读写空闲) */ pipeline.addLast(new IdleStateHandler(7000,7000,10, TimeUnit.SECONDS)); //加入一个对空闲检测进一步处理的handler(自定义) pipeline.addLast(new MyServerHandler()); } }); //启动服务器 ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
2、MyServerHandler
package com.atguigu.netty.heartbeat; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.timeout.IdleStateEvent; public class MyServerHandler extends ChannelInboundHandlerAdapter { /** * * @param ctx 上下文 * @param evt 事件 * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if(evt instanceof IdleStateEvent) { //将 evt 向下转型 IdleStateEvent IdleStateEvent event = (IdleStateEvent) evt; String eventType = null; switch (event.state()) { case READER_IDLE: eventType = "读空闲"; break; case WRITER_IDLE: eventType = "写空闲"; break; case ALL_IDLE: eventType = "读写空闲"; break; } System.out.println(ctx.channel().remoteAddress() + "--超时时间--" + eventType); System.out.println("服务器做相应处理.."); //如果发生空闲，我们关闭通道 // ctx.channel().close(); } } }
通过SSL/TLS 保护Netty 应用程序代码实战
目的
实现一个简单的支持https的服务端以及客户端
代码
服务端
1、HttpServer
package cn.enjoyedu.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
public class HttpServer {
public static final int port = 6789; //设置服务端端口
private static EventLoopGroup group = new NioEventLoopGroup();
private static ServerBootstrap b = new ServerBootstrap();
private static final boolean SSL = false;
public static void main(String[] args) throws Exception {
final SslContext sslCtx;
if (SSL) {
//netty为我们提供的ssl加密，缺省
SelfSignedCertificate ssc = new SelfSignedCertificate();
sslCtx = SslContextBuilder.forServer(ssc.certificate(),
ssc.privateKey()).build();
} else {
sslCtx = null;
}
try {
b.group(group);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ServerHandlerInit(sslCtx));
// 服务器绑定端口监听
ChannelFuture f = b.bind(port).sync();
System.out.println("服务端启动成功,端口是:"+port);
// 监听服务器关闭监听
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
2、ServerHandlerInit
package cn.enjoyedu.server;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.ssl.SslContext;
public class ServerHandlerInit extends ChannelInitializer<SocketChannel> {
private final SslContext sslCtx;
public ServerHandlerInit(SslContext sslCtx) {
this.sslCtx = sslCtx;
}
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
if (sslCtx != null) {
ph.addLast(sslCtx.newHandler(ch.alloc()));
}
//http响应编码
ph.addLast("encode",new HttpResponseEncoder());
//http请求编码
ph.addLast("decode",new HttpRequestDecoder());
//聚合http请求
ph.addLast("aggre",
new HttpObjectAggregator(10*1024*1024));
//启用http压缩
ph.addLast("compressor",new HttpContentCompressor());
//自己的业务处理
ph.addLast("busi",new BusiHandler());
}
}
3、BusiHandler
package cn.enjoyedu.server;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
public class BusiHandler extends ChannelInboundHandlerAdapter {
private String result="";
private void send(String content, ChannelHandlerContext ctx,
HttpResponseStatus status){
FullHttpResponse response =
new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,status,
Unpooled.copiedBuffer(content,CharsetUtil.UTF_8));
response.headers().set(HttpHeaderNames.CONTENT_TYPE,
"text/plain;charset=UTF-8");
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
/*
* 收到消息时，返回信息
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
String result="";
//接收到完成的http请求
FullHttpRequest httpRequest = (FullHttpRequest)msg;
try{
String path = httpRequest.uri();
String body = httpRequest.content().toString(CharsetUtil.UTF_8);
HttpMethod method = httpRequest.method();
if(!"/test".equalsIgnoreCase(path)){
result = "非法请求："+path;
send(result,ctx,HttpResponseStatus.BAD_REQUEST);
return;
}
//处理http GET请求
if(HttpMethod.GET.equals(method)){
System.out.println("body:"+body);
result="Get request,Response="+RespConstant.getNews();
send(result,ctx,HttpResponseStatus.OK);
}
//处理http POST请求
if(HttpMethod.POST.equals(method)){
//.....
}
}catch(Exception e){
System.out.println("处理请求失败!");
e.printStackTrace();
}finally{
httpRequest.release();
}
}
/*
* 建立连接时，返回消息
*/
@Override
public void channelActive(ChannelHandlerContext ctx)
throws Exception {
System.out.println("连接的客户端地址:"
+ ctx.channel().remoteAddress());
}
}
4、RespConstant
package cn.enjoyedu.server;
import java.util.Random;
public class RespConstant {
private static final String[] NEWS = {
"她那时候还太年轻，不知道所有命运赠送的礼物，早已在暗中标好了" +
"价格。——斯蒂芬·茨威格《断头皇后》",
"这是一个最好的时代，也是一个最坏的时代；" +
"这是一个智慧的年代，这是一个愚蠢的年代；
" +
"这是一个信任的时期，这是一个怀疑的时期；" +
"这是一个光明的季节，这是一个黑暗的季节；
" +
"这是希望之春，这是失望之冬；" +
"人们面前应有尽有，人们面前一无所有；
" +
"人们正踏上天堂之路，人们正走向地狱之门。 —— 狄更斯《双城记》",
};
private static final Random R = new Random();
public static String getNews(){
return NEWS[R.nextInt(NEWS.length)];
}
}
客户端
1、HttpClient
package cn.enjoyedu.client;
import cn.enjoyedu.server.HttpServer;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
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.http.*;
import java.net.URI;
public class HttpClient {
private static final boolean SSL = false;
public void connect(String host, int port) throws Exception {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ch.pipeline().addLast(new HttpClientCodec());
ch.pipeline().addLast("aggre",
new HttpObjectAggregator(10*1024*1024));
ch.pipeline().addLast("decompressor",new HttpContentDecompressor());
ch.pipeline().addLast("busi",new HttpClientInboundHandler());
}
});
// Start the client.
ChannelFuture f = b.connect(host, port).sync();
URI uri = new URI("/test");
String msg = "Hello";
DefaultFullHttpRequest request =
new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
HttpMethod.GET,
uri.toASCIIString(),
Unpooled.wrappedBuffer(msg.getBytes("UTF-8")));
// 构建http请求
request.headers().set(HttpHeaderNames.HOST, host);
request.headers()
.set(HttpHeaderNames.CONNECTION,
HttpHeaderValues.KEEP_ALIVE);
request.headers()
.set(HttpHeaderNames.CONTENT_LENGTH,
request.content().readableBytes());
// 发送http请求
f.channel().write(request);
f.channel().flush();
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
HttpClient client = new HttpClient();
client.connect("127.0.0.1", HttpServer.port);
}
}
2、HttpClientInboundHandler
package cn.enjoyedu.client;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.util.CharsetUtil;
public class HttpClientInboundHandler extends ChannelInboundHandlerAdapter {
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
//开始对服务器的响应做处理
FullHttpResponse httpResponse = (FullHttpResponse)msg;
System.out.println(httpResponse.headers());
ByteBuf content = httpResponse.content();
System.out.println(content.toString(CharsetUtil.UTF_8));
content.release();
}
}