netty实现websocket

释放双眼,带上耳机,听听看~!

1.搭建springboot web 项目,参考

https://blog.csdn.net/u013184307/article/details/98057621

2.导入包


1
2
3
4
5
6
7
1<dependency>
2   <groupId>io.netty</groupId>
3   <artifactId>netty-all</artifactId>
4   <version>4.1.36.Final</version>
5</dependency>
6
7

netty实现websocket

3.启动器中需要new一个NettyServer,并显式调用启动netty


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
1package com.horse.red;
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

 


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
56
57
58
59
60
61
1package com.horse.red.netty;
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1package com.horse.red.netty;
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

 


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
1package com.horse.red.netty;
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

 


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
1<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
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

运行效果

netty实现websocket

 

netty实现websocket

上面为群发,没有指明发给指定人,以下解决

 

当有客户端连接时候会被channelActive监听到,当断开时会被channelInactive监听到,一般在这两个方法中去保存/移除客户端的通道信息,而通道信息保存在ChannelSupervise中:

ChannelGroup是netty提供用于管理web于服务器建立的通道channel的,其本质是一个高度封装的set集合,在服务器广播消息时,可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去。但在查找某一个客户端的通道时候比较坑爹,必须通过channelId对象去查找,而channelId不能人为创建,所有必须通过map将channelId的字符串和channel保存起来。


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
1
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

 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
1
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

 

 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
1
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

 

netty实现websocket

 

netty实现websocket

 

完美解决!!!!!!!!!!

给TA打赏
共{{data.count}}人
人已打赏
安全网络

CDN安全市场到2022年价值76.3亿美元

2018-2-1 18:02:50

安全经验

经济危机为部署开源安全解决方案带来曙光

2009-5-6 11:12:22

个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索