说明

这是一款使用 Netty 来实现 IM 双向通信的 demo 项目。

通信双方互为发送方、接收方。通信双方均使用 WebSocket 协议。

通信双方的客户端 GUI 界面均是使用 JavaFX 来实现的。在该文本框中，可以点击 发送 按钮来发送消息，也可以使用 Enter，而在文本中另起一行需要使用组合键 Ctrl + Enter 来完成。

通信过程是由其它线程在后台完成，不会阻塞 UI 线程。

运行效果

核心代码

客户端核心代码

package org.wangpai.demo.im.netty;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
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.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.stream.ChunkedWriteHandler;
import java.net.URI;
import java.net.URISyntaxException;/*** @since 2021-12-1*/
public class Client {private String otherServerIp;private int otherServerPort;public Client setIp(String otherServerIp) {this.otherServerIp = otherServerIp;return this;}public Client setPort(int otherServerPort) {this.otherServerPort = otherServerPort;return this;}private Channel channel;private EventLoopGroup workerLoopGroup = new NioEventLoopGroup();public Client start() {var handshaker = this.getWebSocketClientHandshaker();var businessHandler = new WebsocketClientHandler(handshaker);Bootstrap bootstrap = new Bootstrap();bootstrap.group(workerLoopGroup);bootstrap.channel(NioSocketChannel.class);// 设置接收端的 IP 和端口号，但实际上，自己作为发送端也会为自己自动生成一个端口号bootstrap.remoteAddress(otherServerIp, otherServerPort);bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {var pipeline = ch.pipeline();// 定义客户端 HTTP 编解码器pipeline.addLast(new HttpClientCodec());// 定义分段请求聚合时，字节的最大长度pipeline.addLast(new HttpObjectAggregator(65535));// 定义块写处理器pipeline.addLast(new ChunkedWriteHandler());// 定义业务处理器pipeline.addLast("businessHandler", businessHandler);}});ChannelFuture future = bootstrap.connect();future.addListener((ChannelFuture futureListener) -> {if (futureListener.isSuccess()) {System.out.println("客户端连接成功"); // FIXME：日志} else {System.out.println("客户端连接失败"); // FIXME：日志}});try {future.sync();} catch (Exception exception) {exception.printStackTrace(); // FIXME：日志}this.channel = future.channel();handshaker.handshake(channel);try {businessHandler.sync();} catch (InterruptedException exception) {exception.printStackTrace(); // FIXME：日志}return this;}private String generateWebsocketUrl(String ip, int port, String relativePath) {return String.format("ws://%s:%d/%s", ip, port, relativePath);}/*** 进行三报文握手中的第一握手，由客户端发起** @since 2021-12-2*/private WebSocketClientHandshaker getWebSocketClientHandshaker() {URI websocketUri = null;try {// 对于这个 URL，只有 relativePath 是起作用的，ip、port 不起使用。原因不明websocketUri = new URI(this.generateWebsocketUrl(this.otherServerIp, this.otherServerPort,Protocol.WEBSOCKET_PREFIX_PATH));} catch (URISyntaxException exception) {exception.printStackTrace(); // FIXME：日志}var httpHeaders = new DefaultHttpHeaders();var handshaker = WebSocketClientHandshakerFactory.newHandshaker(websocketUri, WebSocketVersion.V13, null, false, httpHeaders);return handshaker;}public void send(String msg) {channel.writeAndFlush(new TextWebSocketFrame(msg));}public void destroy() {this.workerLoopGroup.shutdownGracefully();}private Client() {super();}public static Client getInstance() {return new Client();}
}

客户端处理器核心代码

package org.wangpai.demo.im.netty;import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketHandshakeException;
import io.netty.util.CharsetUtil;/*** @since 2021-12-2*/
public class WebsocketClientHandler extends ChannelInboundHandlerAdapter {private WebSocketClientHandshaker handshaker;private ChannelPromise channelPromise;public WebsocketClientHandler(WebSocketClientHandshaker handshaker) {this.handshaker = handshaker;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object obj) {if (!this.handshaker.isHandshakeComplete()) { // 如果三报文握手的流程还没有走完// 如果现在进行的是三报文握手中的第三握手finishHandshake(ctx, (FullHttpResponse) obj);} else if (obj instanceof FullHttpResponse) {FullHttpResponse response = (FullHttpResponse) obj;var msg = String.format("Unexpected FullHttpResponse, status=%s, content=%s",response.status(), response.content().toString(CharsetUtil.UTF_8));System.out.println(msg); // FIXME：日志} else if (obj instanceof WebSocketFrame) {// TODO：如果需要服务器反馈信息，可在此添加业务handleWebSocketResponse(ctx, (WebSocketFrame) obj);} else {// 此分支不应该发生}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) {this.channelPromise = ctx.newPromise();}public ChannelFuture sync() throws InterruptedException {return this.channelPromise.sync();}private void finishHandshake(ChannelHandlerContext ctx, FullHttpResponse response) {try {this.handshaker.finishHandshake(ctx.channel(), response);//设置成功this.channelPromise.setSuccess();} catch (WebSocketHandshakeException exception) {FullHttpResponse rsp = response;String errorMsg = String.format("WebSocket Client failed to connect, status=%s, reason=%s",rsp.status(), rsp.content().toString(CharsetUtil.UTF_8));this.channelPromise.setFailure(new Exception(errorMsg)); // TODO：日志}}private void handleWebSocketResponse(ChannelHandlerContext ctx, WebSocketFrame frame) {if (frame instanceof TextWebSocketFrame) {TextWebSocketFrame textFrame = (TextWebSocketFrame) frame; // TODO} else if (frame instanceof BinaryWebSocketFrame) {BinaryWebSocketFrame binFrame = (BinaryWebSocketFrame) frame; // TODO} else if (frame instanceof PongWebSocketFrame) {// TODO} else if (frame instanceof CloseWebSocketFrame) {// TODOctx.channel().close();}}
}

服务器端核心代码

package org.wangpai.demo.im.netty;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
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.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import lombok.Setter;
import lombok.experimental.Accessors;
import org.wangpai.demo.im.view.MainFace;/*** @since 2021-12-1*/
@Accessors(chain = true)
public class Server {@Setterprivate int port;@Setterprivate MainFace mainFace;private EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);private EventLoopGroup workerLoopGroup = new NioEventLoopGroup();public Server start() {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(this.bossLoopGroup, this.workerLoopGroup);bootstrap.channel(NioServerSocketChannel.class);bootstrap.localAddress(port);bootstrap.option(ChannelOption.SO_KEEPALIVE, true);bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {var pipeline = ch.pipeline();// 定义服务端 HTTP 编解码器pipeline.addLast(new HttpServerCodec());// 定义分段请求聚合时，字节的最大长度pipeline.addLast(new HttpObjectAggregator(65535));// 定义块写处理器pipeline.addLast(new ChunkedWriteHandler());// 设置数据压缩处理器pipeline.addLast(new WebSocketServerCompressionHandler());// 设置监听的路径pipeline.addLast(new WebSocketServerProtocolHandler("/" + Protocol.WEBSOCKET_PREFIX_PATH));// 定义业务处理器-文本处理器pipeline.addLast(new TextServerHandler(mainFace));}});try {bootstrap.bind().sync().channel().closeFuture().sync();} catch (Exception exception) {exception.printStackTrace(); // FIXME：日志} finally {this.workerLoopGroup.shutdownGracefully();this.bossLoopGroup.shutdownGracefully();}return this;}public void destroy() {this.workerLoopGroup.shutdownGracefully();this.bossLoopGroup.shutdownGracefully();}private Server() {super();}public static Server getInstance() {return new Server();}
}

服务端处理器核心代码

package org.wangpai.demo.im.netty;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.wangpai.demo.im.view.MainFace;/*** @since 2021-12-8*/
public class TextServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {private MainFace mainFace;public TextServerHandler(MainFace mainFace) {super();this.mainFace = mainFace;}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {this.mainFace.receive(msg.text());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof TextWebSocketFrame) {super.channelRead(ctx, msg);} else { // 如果 msg 不是文本，交给流水线上后续的处理器来处理ctx.fireChannelRead(msg);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

完整代码

已上传至 GitCode 中，可免费下载：https://gitcode.net/wangpaiblog/20211221-im_demo-netty_websocket_javafx

参考知识

WebSocket 的通信机制：https://blog.csdn.net/wangpaiblog/article/details/121804329
JavaFX 中使用多线程与保证 UI 线程安全：https://blog.csdn.net/wangpaiblog/article/details/120755930
如何在 JavaFX 的 TextArea 实现回车发送信息而不换行，但组合键 Ctrl + Enter 换行：https://blog.csdn.net/wangpaiblog/article/details/121506912

简易 IM 双向通信电脑端 GUI 应用——基于 Netty、WebSocket、JavaFX 、多线程技术等相关推荐

简易 IM 双向通信电脑端 GUI 应用——基于 Netty、JavaFX、多线程技术等
简易 IM 双向通信电脑端 GUI 应用--基于 Netty.JavaFX.多线程技术等说明运行效果核心代码完整代码参考知识说明这是一个使用 Netty 来实现 IM 双向通信的 d ...
基于netty+websocket实现门户游客实时统计功能
基于netty+websocket实现门户游客实时统计功能基本需求商城门户页面需要实时展示游客访问的数量,商城后台页面需要实时游客访问量.登录用户数量,以及下订单用户数量. 技术选型 1.首先实时 ...
基于安卓的视频遥控小车——电脑端开发
基于安卓的视频遥控小车的电脑端程序采用Java语言编写,Java可以做到一次编译到处运行,因为Java程序是在Java虚拟机中运行的,和平台无关,只要平台上有相应的Java虚拟机. 本设计中安卓手机是 ...
pyaudio：基于pyaudio利用Python编程从电脑端录制音频保存到指定文件夹+将录音上传服务器+录音进行识别并转为文本保存
pyaudio:基于pyaudio利用Python编程从电脑端录制音频保存到指定文件夹+将录音上传服务器+录音进行识别并转为文本保存目录输出结果代码实现输出结果代码实现 # -*- codi ...
基于Spring Boot实现电脑端网页微信扫码授权登录方式一（附带完整源码）
简介电脑端微信网页扫码授权登录有2种方式: 第一种:基于微信公众号,单独获取登录二维码扫码,然后扫码登录,程序控制跳转逻辑,例如CSDN: 第二种:基于微信开放平台,跳转到微信二维码页面进行扫码登录 ...
如何观看局域网内视频-利用个人电脑搭建简易NAS（笔记一）电脑端工作
老鸟飞过.......... 为啥会在这写这么个文章呢,因为纯属笔记,做记录用的. 起因:因为最近很火的一步韩剧<鱿鱼游戏>,下载来以后在电脑上但是又不想用电脑看躺沙发上刷剧多美滋滋,于是 ...
使用VB制作一个简易通信录电话号码查询器电脑端通信录
制作背景: 有手机,为什么还用电脑端的简易通信录? 办公一族,每天上班基本上都是对着电脑,左边一部电话,中间是电脑,右边是文件.如下图所示. 有很多电话是通过内线或者固定电话拨出的.因此,一本纸质的电 ...
苹果微信多开_简易版！微信电脑端多开方法！！！
点击上方"GitHub指南",星标公众号优秀文章,第一时间送达hello,大家好,这里是GitHub指南.有点好奇大家有几个微信号?不少小伙伴应该都有两个甚至多个微信号吧,为了避免 ...
iPad与电脑端文件互传解决方案（基于nPlayer lite）
在APP store中下载nPlayer软件(免费的lite版本即可,但会稍微麻烦一点) 本文先以lite版本为例,后续会给出付费版本的方法电脑端向iPad传输文件: 点击nPlayer左上角的加号 ...

简易 IM 双向通信电脑端 GUI 应用——基于 Netty、WebSocket、JavaFX 、多线程技术等