一、pom.xml依赖配置
|
1
2
3
4
5
6
|
<!-- netty --><dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> <version>4.1.50.Final</version></dependency> |
二、代码
2.1、NettyServer 类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
|
package com.wangjing.socket.server; import com.wangjing.socket.handler.CoordinationSocketHandler;import io.netty.bootstrap.ServerBootstrap;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.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.stream.ChunkedWriteHandler; public class CoordinationNettyServer { private final int port; public CoordinationNettyServer(int port) { this.port = port; } public void start() throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup group = new NioEventLoopGroup(); try { ServerBootstrap sb = new ServerBootstrap(); sb.option(ChannelOption.SO_BACKLOG, 1024); sb.group(group, bossGroup) // 绑定线程池 .channel(NioServerSocketChannel.class) // 指定使用的channel .localAddress(this.port)// 绑定监听端口 .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作 @Override protected void initChannel(SocketChannel ch) throws Exception { //websocket协议本身是基于http协议的，所以这边也要使用http解编码器 ch.pipeline().addLast(new HttpServerCodec()); //以块的方式来写的处理器 ch.pipeline().addLast(new ChunkedWriteHandler()); ch.pipeline().addLast(new HttpObjectAggregator(8192)); ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10)); ch.pipeline().addLast(new CoordinationSocketHandler());//自定义消息处理类 } }); ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定 System.out.println(CoordinationNettyServer.class + "已启动，正在监听： " + cf.channel().localAddress()); cf.channel().closeFuture().sync(); // 关闭服务器通道 } finally { group.shutdownGracefully().sync(); // 释放线程池资源 bossGroup.shutdownGracefully().sync(); } }} |
2.2、SocketHandler 类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
package com.wangjing.socket.handler; import com.wangjing.socket.pool.CoordinationChannelHandlerPool;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; public class CoordinationSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("与客户端建立连接，通道开启！"); //添加到channelGroup通道组 CoordinationChannelHandlerPool.channelGroup.add(ctx.channel()); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println("与客户端断开连接，通道关闭！"); //从channelGroup通道组删除 CoordinationChannelHandlerPool.channelGroup.remove(ctx.channel()); } @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { //接收的消息 System.out.println(String.format("收到客户端%s的数据：%s", ctx.channel().id(), msg.text())); // 单独发消息 // sendMessage(ctx); // 群发消息 sendAllMessage(); } private void sendMessage(ChannelHandlerContext ctx) throws InterruptedException { String message = "我是服务器，你好呀"; ctx.writeAndFlush(new TextWebSocketFrame("hello")); } private void sendAllMessage() { String message = "我是服务器，这是群发消息"; CoordinationChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message)); } } |
2.3、ChannelHandlerPool 类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
package com.wangjing.socket.pool; import io.netty.channel.group.ChannelGroup;import io.netty.channel.group.DefaultChannelGroup;import io.netty.util.concurrent.GlobalEventExecutor; public class CoordinationChannelHandlerPool { public CoordinationChannelHandlerPool() { } //可以存储userId与ChannelId的映射表// public static ConcurrentHashMap<String, ChannelId> channelIdMap = new ConcurrentHashMap<>(); //channelGroup通道组 public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); } |
2.4、Application启动类
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
package com.wangjing.socket; import com.wangjing.socket.server.CoordinationNettyServer;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication(scanBasePackages = "com.wangjing")public class SocketApplication { public static void main(String[] args) { SpringApplication.run(SocketApplication.class, args); try { new CoordinationNettyServer(8804).start(); } catch (Exception e) { System.out.println("NettyServerError:" + e.getMessage()); } } } |
三、测试
websocket 在线测试推荐：在线websocket测试-online tool-postjson
到此这篇关于SpringBoot整合Netty实现WebSocket的示例代码的文章就介绍到这了,更多相关SpringBoot整合Netty实现WebSocket内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！
原文链接：https://blog.csdn.net/wang_jing_jing/article/details/124713009
