執行個體要求
- 編寫一個 Netty 群聊系統，實作伺服器端和用戶端之間的資料簡單通訊（非阻塞）
- 實作多人群聊
- 伺服器端：可以監測使用者上線，離線，并實作消息轉發功能
- 用戶端：通過channel 可以無阻塞發送消息給其它所有使用者，同時可以接受其它使用者發送的消息(有伺服器轉發得到)
目的：進一步了解Netty非阻塞網絡程式設計機制
服務端
編寫GroupChatServer類
GroupChatServerHandler
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
java複制代碼public class GroupChatServer {
private int port;//監聽端口
public GroupChatServer(int port) {
this.port = port;
}
//編寫run方法，處理用戶端端請求
public void run() throws InterruptedException {
//建立兩個線程組
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.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());
//向pipeline加入編碼器
pipeline.addLast("encoder", new StringEncoder());
//加入自己到業務處理handler
pipeline.addLast(new GroupChatServerHandler());
}
});
System.out.println("Netty 伺服器啟動");
ChannelFuture channelFuture = bootstrap.bind(port).sync();
//監聽關閉
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new GroupChatServer(7000).run();
}
}
java複制代碼public class GroupChatServerHandler 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");
//handlerAdded表示連接配接建立，一旦連接配接，第一個被執行該方法
//将目前channel加入到channelGroup
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
//将該用戶端加入聊天端資訊推送給其他線上端用戶端。該方法會将channelGroup中所有端channel周遊，并發送資訊。是以不需要自己周遊
channelGroup.writeAndFlush("[用戶端]" + channel.remoteAddress() + " 加入聊天" + sdf.format(new Date()) + "\n");
channelGroup.add(channel);
}
//斷開連接配接，将xx客戶離開資訊推送哥目前線上的客戶
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
channelGroup.writeAndFlush("[用戶端]" + channel.remoteAddress() + " 離開了\n");
System.out.println("目前channelGroup大小：" + channelGroup.size());
}
//表示channel處于活動的狀态 提示xxx上線
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println(ctx.channel().remoteAddress() + " 上線了～"+sdf.format(new Date()) + "\n");
}
//表示channel處于不活動的狀态 提示xxx下線
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println(ctx.channel().remoteAddress() + " 離線了～"+sdf.format(new Date()) + "\n");
}
@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 + sdf.format(new Date()) + "\n");
} else {//自己發送的資訊，在自己上如何顯示
ch.writeAndFlush("[自己]" + " 發送了消息：" + msg + "\n");
}
});
}
}
用戶端
GroupChatClient
cdn.jsdelivr.net/gh/kylincw/…
java複制代碼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 InterruptedException {
NioEventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
try {
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+"\r\n");
}
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new GroupChatClient("127.0.0.1",7000).run();
}
}
java複制代碼public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(msg.trim());//取掉兩端空格
}
}
測試
作者：以範特西之名
連結：https://juejin.cn/post/7244172094547083323