Netty 应用：Socket服务器

Socket服务器

Netty作为Socket服务器，需要编写服务端和客户端，服务器端和客户端收发消息通过自定义的Handler.channelRead0方法来交互，客户端连接上服务器后，需要在active时向服务器发送一条消息来触发服务器的行为。

服务端实现

1/**
2 * Created by fubin on 2019/7/11.
3 */
4public class SocketServer {
5    public static void main(String[] args) {
6        EventLoopGroup bossGroup = new NioEventLoopGroup();
7        EventLoopGroup workerGroup = new NioEventLoopGroup();
8        try{
9            ServerBootstrap serverBootstrap = new ServerBootstrap();
10            serverBootstrap.group(bossGroup,workerGroup)
11                    .channel(NioServerSocketChannel.class)
12                    .childHandler(new SocketServerInitializer());
13
14            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
15            channelFuture.channel().closeFuture().sync();
16        } catch (InterruptedException e) {
17            e.printStackTrace();
18        } finally {
19            bossGroup.shutdownGracefully();
20            workerGroup.shutdownGracefully();
21        }
22    }
23}
24class  SocketServerInitializer extends ChannelInitializer<SocketChannel>{
25    protected void initChannel(SocketChannel socketChannel) throws Exception {
26        ChannelPipeline channelPipeline = socketChannel.pipeline();
27        channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
28        channelPipeline.addLast(new LengthFieldPrepender(4));
29        channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
30        channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
31        channelPipeline.addLast(new SocketServerHandler());
32    }
33}
34class SocketServerHandler extends SimpleChannelInboundHandler<String>{
35    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
36        //远程地址
37        System.out.println(ctx.channel().remoteAddress()+","+msg);
38        //处理完业务，把结果返回给客户端
39        ctx.channel().writeAndFlush("from server:"+ UUID.randomUUID());
40    }
41    @Override
42    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
43        super.exceptionCaught(ctx, cause);
44        //出现异常，关闭连接
45        ctx.close();
46    }
47}

客户端实现

1/**
2 * Created by fubin on 2019/7/11.
3 */
4public class SocketClient {
5    public static void main(String[] args) {
6        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
7
8        try{
9            Bootstrap bootstrap = new Bootstrap();
10            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new SocketClientInitializer());
11
12            ChannelFuture future = bootstrap.connect("localhost",8899).sync();
13            future.channel().closeFuture().sync();
14        } catch (InterruptedException e) {
15            e.printStackTrace();
16        } finally {
17            eventLoopGroup.shutdownGracefully();
18        }
19    }
20}
21class SocketClientInitializer extends ChannelInitializer<SocketChannel>{
22    protected void initChannel(SocketChannel ch) throws Exception {
23        ChannelPipeline channelPipeline = ch.pipeline();
24
25        channelPipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));
26        channelPipeline.addLast(new LengthFieldPrepender(4));
27        channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
28        channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
29        channelPipeline.addLast(new SocketClientHandler());
30    }
31}
32class SocketClientHandler extends SimpleChannelInboundHandler<String>{
33    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
34        System.out.println(ctx.channel().remoteAddress());
35        System.out.println("client output:"+msg);
36        ctx.channel().writeAndFlush("from client:"+ LocalDateTime.now());
37    }
38    @Override
39    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
40        super.exceptionCaught(ctx, cause);
41        ctx.close();
42    }
43    //在连接激活时，客户端向服务器发送一条消息
44    @Override
45    public void channelActive(ChannelHandlerContext ctx) throws Exception {
46        ctx.channel().writeAndFlush("hello world socket");
47    }
48}

Socket实现的聊天程序消息群发

netty实现多客户端消息群发的关键： ChannelGroup

服务端实现

1/**
2 * Created by fubin on 2019/7/11.
3 * 聊天程序，消息群发
4 */
5public class ChatServer {
6    public static void main(String[] args) {
7        EventLoopGroup bossGroup = new NioEventLoopGroup();
8        EventLoopGroup workerGroup = new NioEventLoopGroup();
9        try{
10            ServerBootstrap serverBootstrap = new ServerBootstrap();
11            serverBootstrap.group(bossGroup,workerGroup)
12                    .channel(NioServerSocketChannel.class)
13                    .childHandler(new ChatServerInitializer());
14
15            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
16            channelFuture.channel().closeFuture().sync();
17        } catch (InterruptedException e) {
18            e.printStackTrace();
19        } finally {
20            bossGroup.shutdownGracefully();
21            workerGroup.shutdownGracefully();
22        }
23    }
24}
25class  ChatServerInitializer extends ChannelInitializer<SocketChannel> {
26    protected void initChannel(SocketChannel socketChannel) throws Exception {
27        ChannelPipeline channelPipeline = socketChannel.pipeline();
28        //分隔符编解码器
29        channelPipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
30        channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
31        channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
32        channelPipeline.addLast(new ChatServerHandler());
33    }
34}
35class ChatServerHandler extends SimpleChannelInboundHandler<String> {
36
37    //保存channel对象
38    private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
39
40    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
41        Channel channel = ctx.channel();
42        //广播消息
43        channels.forEach(ch ->{
44            if(channel != ch){
45                ch.writeAndFlush(channel.remoteAddress() + "发送的消息："+ msg + "\n");
46            }else{
47                ch.writeAndFlush(" 【自己】" + msg + "\n");
48            }
49        });
50    }
51    @Override
52    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
53        super.exceptionCaught(ctx, cause);
54        //出现异常，关闭连接
55        ctx.close();
56    }
57    /**
58     * 连接建立，告诉其他客户端，xxx已经上线
59     * @param ctx
60     * @throws Exception
61     */
62    @Override
63    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
64        Channel channel = ctx.channel();
65        //广播
66        channels.writeAndFlush("【服务器】 - " + channel.remoteAddress() + "加入\n");
67
68        channels.add(channel);
69    }
70    @Override
71    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
72        Channel channel = ctx.channel();
73        //广播
74        channels.writeAndFlush("【服务器】 - " + channel.remoteAddress() + "离开\n");
75        //此代码可不要，netty自动移除
76        //channels.remove(channel);
77        //验证
78        System.out.println("channelGroup size :"+channels.size());
79    }
80    @Override
81    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
82        Channel channel = ctx.channel();
83        System.out.println("【服务器】- "+ channel.remoteAddress() + "下线\n");
84    }
85    @Override
86    public void channelActive(ChannelHandlerContext ctx) throws Exception {
87        Channel channel = ctx.channel();
88        System.out.println("【服务器】- "+ channel.remoteAddress() + "上线\n");
89    }
90}

客户端实现

1/**
2 * Created by fubin on 2019/7/11.
3 */
4public class ChatClient {
5    public static void main(String[] args) {
6        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
7
8        try{
9            Bootstrap bootstrap = new Bootstrap();
10            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new ChatClientInitializer());
11            Channel channel = bootstrap.connect("localhost",8899).sync().channel();
12            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
13            //循环监听客户端输入
14            for(;;){
15                channel.writeAndFlush(br.readLine()+"\r\n");
16            }
17        } catch (InterruptedException | IOException e) {
18            e.printStackTrace();
19        } finally {
20            eventLoopGroup.shutdownGracefully();
21        }
22    }
23}
24class ChatClientInitializer extends ChannelInitializer<SocketChannel>{
25    protected void initChannel(SocketChannel ch) throws Exception {
26        ChannelPipeline channelPipeline = ch.pipeline();
27
28        channelPipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
29        channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
30        channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
31        channelPipeline.addLast(new ChatClientHandler());
32    }
33}
34class ChatClientHandler extends SimpleChannelInboundHandler<String>{
35    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
36        System.out.println(msg);
37    }
38}

应用内的推送的思考：服务器端和客户端维持长连接推送给app，如果客户端断线，服务端需要把消息保存在数据库或者mongodb中，当客户端重连后，服务端再再重新发送。

Netty 应用：Socket服务器

Socket服务器

Socket实现的聊天程序消息群发

评论区

相关推荐