释放双眼，带上耳机，听听看~！
1.搭建springboot web 项目，参考
https://blog.csdn.net/u013184307/article/details/98057621
2.导入包
2
3
4
5
6
7
2 <groupId>io.netty</groupId>
3 <artifactId>netty-all</artifactId>
4 <version>4.1.36.Final</version>
5</dependency>
6
7
3.启动器中需要new一个NettyServer，并显式调用启动netty
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2
3import com.horse.red.netty.NettyServer;
4import org.springframework.boot.SpringApplication;
5import org.springframework.boot.autoconfigure.SpringBootApplication;
6
7@SpringBootApplication
8public class RedApplication {
9
10 public static void main(String[] args) {
11 SpringApplication.run(RedApplication.class, args);
12
13 try {
14 new NettyServer(12345).start();
15 } catch (Exception e) {
16 e.printStackTrace();
17 }
18 }
19
20}
21
22
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
56
57
58
59
60
61
2
3import io.netty.bootstrap.ServerBootstrap;
4import io.netty.channel.ChannelFuture;
5import io.netty.channel.ChannelInitializer;
6import io.netty.channel.ChannelOption;
7import io.netty.channel.EventLoopGroup;
8import io.netty.channel.nio.NioEventLoopGroup;
9import io.netty.channel.socket.SocketChannel;
10import io.netty.channel.socket.nio.NioServerSocketChannel;
11import io.netty.handler.codec.http.HttpObjectAggregator;
12import io.netty.handler.codec.http.HttpServerCodec;
13import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
14import io.netty.handler.stream.ChunkedWriteHandler;
15
16public class NettyServer {
17
18 private final int port;
19
20 public NettyServer(int port) {
21 this.port = port;
22 }
23
24 public void start() throws Exception {
25 EventLoopGroup bossGroup = new NioEventLoopGroup();
26
27 EventLoopGroup group = new NioEventLoopGroup();
28 try {
29 ServerBootstrap sb = new ServerBootstrap();
30 sb.option(ChannelOption.SO_BACKLOG, 1024);
31 sb.group(group, bossGroup) // 绑定线程池
32 .channel(NioServerSocketChannel.class) // 指定使用的channel
33 .localAddress(this.port)// 绑定监听端口
34 .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
35
36 @Override
37 protected void initChannel(SocketChannel ch) throws Exception {
38 System.out.println("收到新连接");
39 //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
40 ch.pipeline().addLast(new HttpServerCodec());
41 //以块的方式来写的处理器
42 ch.pipeline().addLast(new ChunkedWriteHandler());
43 ch.pipeline().addLast(new HttpObjectAggregator(8192));
44 ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
45 ch.pipeline().addLast(new MyWebSocketHandler());
46 }
47 });
48 ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
49 System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
50 cf.channel().closeFuture().sync(); // 关闭服务器通道
51 } finally {
52 group.shutdownGracefully().sync(); // 释放线程池资源
53 bossGroup.shutdownGracefully().sync();
54 }
55 }
56}
57
58
59
60
61
2
3
4
5
6
7
8
9
10
11
12
13
14
15
2
3import io.netty.channel.group.ChannelGroup;
4import io.netty.channel.group.DefaultChannelGroup;
5import io.netty.util.concurrent.GlobalEventExecutor;
6
7public class MyChannelHandlerPool {
8
9 public MyChannelHandlerPool(){}
10
11 public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
12
13}
14
15
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
2
3import io.netty.channel.ChannelHandlerContext;
4import io.netty.channel.SimpleChannelInboundHandler;
5import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
6
7public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
8
9 @Override
10 public void channelActive(ChannelHandlerContext ctx) throws Exception {
11 System.out.println("与客户端建立连接，通道开启！");
12
13 //添加到channelGroup通道组
14 MyChannelHandlerPool.channelGroup.add(ctx.channel());
15 }
16
17 @Override
18 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
19 System.out.println("与客户端断开连接，通道关闭！");
20 //添加到channelGroup 通道组
21 MyChannelHandlerPool.channelGroup.remove(ctx.channel());
22 }
23
24 @Override
25 protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
26 System.out.println("客户端收到服务器数据：" + msg.text());
27 sendAllMessage(msg.text());
28 }
29
30 private void sendAllMessage(String message){
31 //收到信息后，群发给所有channel
32 MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
33 }
34}
35
36
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
2<html xmlns="http://www.w3.org/1999/xhtml">
3<head>
4 <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
5 <title>Netty-Websocket</title>
6 <script type="text/javascript">
7 // by zhengkai.blog.csdn.net
8 var socket;
9 if(!window.WebSocket){
10 window.WebSocket = window.MozWebSocket;
11 }
12 if(window.WebSocket){
13 socket = new WebSocket("ws://127.0.0.1:12345/ws");
14 socket.onmessage = function(event){
15 var ta = document.getElementById('responseText');
16 ta.value += event.data+"\r\n";
17 };
18 socket.onopen = function(event){
19 var ta = document.getElementById('responseText');
20 ta.value = "Netty-WebSocket服务器。。。。。。连接 \r\n";
21 };
22 socket.onclose = function(event){
23 var ta = document.getElementById('responseText');
24 ta.value = "Netty-WebSocket服务器。。。。。。关闭 \r\n";
25 };
26 }else{
27 alert("您的浏览器不支持WebSocket协议！");
28 }
29 function send(message){
30 if(!window.WebSocket){return;}
31 if(socket.readyState == WebSocket.OPEN){
32 socket.send(message);
33 }else{
34 alert("WebSocket 连接没有建立成功！");
35 }
36
37 }
38
39 </script>
40</head>
41<body>
42<form onSubmit="return false;">
43 <label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br />
44 <label>TEXT</label><input type="text" name="message" value="这里输入消息" /> <br />
45 <br /> <input type="button" value="发送ws消息"
46 onClick="send(this.form.uid.value+':'+this.form.message.value)" />
47 <hr color="black" />
48 <h3>服务端返回的应答消息</h3>
49 <textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
50</form>
51</body>
52</html>
53
运行效果
上面为群发，没有指明发给指定人，以下解决
当有客户端连接时候会被channelActive监听到，当断开时会被channelInactive监听到，一般在这两个方法中去保存/移除客户端的通道信息，而通道信息保存在ChannelSupervise中：
ChannelGroup是netty提供用于管理web于服务器建立的通道channel的，其本质是一个高度封装的set集合，在服务器广播消息时，可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去。但在查找某一个客户端的通道时候比较坑爹，必须通过channelId对象去查找，而channelId不能人为创建，所有必须通过map将channelId的字符串和channel保存起来。
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
2package com.horse.red.netty;
3
4import io.netty.channel.Channel;
5import io.netty.channel.ChannelId;
6import io.netty.channel.group.ChannelGroup;
7import io.netty.channel.group.DefaultChannelGroup;
8import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
9import io.netty.util.concurrent.GlobalEventExecutor;
10
11import java.util.concurrent.ConcurrentHashMap;
12import java.util.concurrent.ConcurrentMap;
13
14public class MyChannelHandlerPool {
15
16 public MyChannelHandlerPool(){}
17
18 public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
19
20 private static ConcurrentMap<String, ChannelId> ChannelMap=new ConcurrentHashMap();
21
22 public static void addChannel(Channel channel){
23 channelGroup.add(channel);
24 ChannelMap.put(channel.id().asShortText(),channel.id());
25 }
26 public static void removeChannel(Channel channel){
27 channelGroup.remove(channel);
28 ChannelMap.remove(channel.id().asShortText());
29 }
30 public static Channel findChannel(String id){
31 return channelGroup.find(ChannelMap.get(id));
32 }
33}
34
35
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
2@Override
3protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
4 System.out.println("客户端收到服务器数据：" + msg.text());
5 ChannelId channelId = ctx.channel().id();
6 //sendAllMessage(msg.text());
7 sendAllMessage2(msg.text(),channelId);
8}
9
10private void sendAllMessage(String message){
11 //收到信息后，群发给所有channel
12 MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
13
14}
15
16private void sendAllMessage2(String message,ChannelId channelId){
17 //收到信息后，发给指定channel
18 MyChannelHandlerPool.findChannel(channelId.asShortText()).writeAndFlush( new TextWebSocketFrame(message));
19
20}
21
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2@Override
3public void channelActive(ChannelHandlerContext ctx) throws Exception {
4 System.out.println("与客户端建立连接，通道开启！");
5
6 //添加到channelGroup通道组
7 MyChannelHandlerPool.addChannel(ctx.channel());
8}
9
10@Override
11public void channelInactive(ChannelHandlerContext ctx) throws Exception {
12 System.out.println("与客户端断开连接，通道关闭！");
13 //添加到channelGroup 通道组
14 MyChannelHandlerPool.removeChannel(ctx.channel());
15}
16
完美解决！！！！！！！！！！
