SpringBoot整合Netty实现WebSocket的示例代码

目录一、pom.xml依赖配置二、代码2.1、NettyServer 类2.2、SocketHandler 类2.3、ChannelHandlerPool 类2.4、Application启动类三、测试

一、pom.xml依赖配置

io.netty

netty-all

4.1.50.Final

二、代码

2.1、NettyServer 类

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() { // 绑定客户端连接时候触发操作

@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 类

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 {

@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 类

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 channelIdMap = new ConcurrentHashMap<>();

//channelGroup通道组

public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

2.4、Application启动类

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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。