netty聊天推送消息

转：https://zhuanlan.zhihu.com/p/151000574
https://blog.csdn.net/xishining/article/details/105742797

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.42.Final</version>
</dependency>

PushController.java

@RestController
@RequestMapping(“/push”)
public class PushController {

@Autowired
private PushService pushService;

/**
* 推送给所有用户
* @param __msg
*/
@PostMapping(“/pushAll”)
public void pushToAll(@RequestParam(“msg”) String msg){
pushService.pushMsgToAll(msg);
}
/**
* 推送给指定用户
* @param __userId
* @param __msg
*/
@PostMapping(“/pushOne”)
public void pushMsgToOne(@RequestParam(“userId”) String userId,@RequestParam(“msg”) String msg){
pushService.pushMsgToOne(userId,msg);
}
}

NettyConfig.java

public class NettyConfig {
/**
* 定义一个channel组，管理所有的channel
* GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例
*/
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

/**
* 存放用户与Chanel的对应信息，用于给指定用户发送消息
*/
private static ConcurrentHashMap userChannelMap = new ConcurrentHashMap<>();

private NettyConfig() {
}

/**
* 获取channel组
*
* @return
*/
public static ChannelGroup getChannelGroup() {
return channelGroup;
}

/**
* 获取用户channel map
*
* @return
*/
public static ConcurrentHashMap getUserChannelMap() {
return userChannelMap;
}
}

NettyServer.java

@Component
public class NettyServer {

/**
* webSocket协议名
*/
private static final String WEBSOCKET_PROTOCOL = “WebSocket”;

/**
* 端口号
*/
@Value(“${webSocket.netty.port:58080}”)
private int port;

/**
* webSocket路径
*/
@Value(“${webSocket.netty.path:/webSocket}”)
private String webSocketPath;

@Autowired
private WebSocketHandler webSocketHandler;

private EventLoopGroup bossGroup;
private EventLoopGroup workGroup;

/**
* 启动
*
* @throws InterruptedException
*/
private void start() throws InterruptedException {
bossGroup = new NioEventLoopGroup();
workGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
// bossGroup辅助客户端的tcp连接请求, workGroup负责与客户端之前的读写操作
_bootstrap.group(bossGroup, workGroup);
// 设置NIO类型的channel
bootstrap.channel(NioServerSocketChannel.class);
// 设置监听端口
bootstrap.localAddress(new InetSocketAddress(port));
// 连接到达时会创建一个通道
_bootstrap.childHandler(new ChannelInitializer() {

     @Override<br />         **protected void **initChannel(SocketChannel ch) **throws **Exception {<br />            _// 流水线管理通道中的处理程序（Handler），用来处理业务_<br />_            // webSocket协议本身是基于http协议的，所以这边也要使用http编解码器_<br />_            _ch.pipeline().addLast(**new **HttpServerCodec());<br />            ch.pipeline().addLast(**new **ObjectEncoder());<br />            _// 以块的方式来写的处理器_<br />_            _ch.pipeline().addLast(**new **ChunkedWriteHandler());<br />        _/*_<br />_        说明：_<br />_        1、http数据在传输过程中是分段的，HttpObjectAggregator可以将多个段聚合_<br />_        2、这就是为什么，当浏览器发送大量数据时，就会发送多次http请求_<br />_         */_<br />_            _ch.pipeline().addLast(**new **HttpObjectAggregator(8192));<br />        _/*_<br />_        说明：_<br />_        1、对应webSocket，它的数据是以帧（frame）的形式传递_<br />_        2、浏览器请求时 ws://localhost:58080/xxx 表示请求的uri_<br />_        3、核心功能是将http协议升级为ws协议，保持长连接_<br />_        */_<br />_            _ch.pipeline().addLast(**new **WebSocketServerProtocolHandler(**webSocketPath**, **_WEBSOCKET_PROTOCOL_**, **true**, 65536 * 10));<br />            _// 自定义的handler，处理业务逻辑_<br />_            _ch.pipeline().addLast(**webSocketHandler**);
     }<br />      });<br />      _// 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功_<br />_      _ChannelFuture channelFuture = bootstrap.bind().sync();<br />      MyLogUtil._info_(**"Server started and listen on:" **+ channelFuture.channel().localAddress());<br />      _// 对关闭通道进行监听_<br />_      _channelFuture.channel().closeFuture().sync();<br />   }

/**
* 释放资源
*
* @throws InterruptedException
*/
@PreDestroy
public void destroy() throws InterruptedException {
if (bossGroup != null) {
bossGroup.shutdownGracefully().sync();
}
if (workGroup != null) {
workGroup.shutdownGracefully().sync();
}
}

@PostConstruct()
public void init() {
//需要开启一个新的线程来执行netty server 服务器
new Thread(() -> {
try {
start();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}

PushService.java

public interface PushService {
/**
* 推送给指定用户
*
* @param __userId
* @param __msg
*/
void pushMsgToOne(String userId, String msg);

/**
* 推送给所有用户
*
* @param __msg
*/
void pushMsgToAll(String msg);
}

PushServiceImpl.java

@Service
public class PushServiceImpl implements PushService {

@Override
public void pushMsgToOne(String userId, String msg){
ConcurrentHashMap userChannelMap = NettyConfig.getUserChannelMap();
Channel channel = userChannelMap.get(userId);
channel.writeAndFlush(new TextWebSocketFrame(msg));
}
@Override
public void pushMsgToAll(String msg){
NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketFrame(msg));
}
}

WebSocketHandler.java

/**
* TextWebSocketFrame类型， 表示一个文本帧

*/
@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler {

/**
* 一旦连接，第一个被执行
*
* @param __ctx
* @throws Exception
*/
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
MyLogUtil.info(“handlerAdded 被调用” + ctx.channel().id().asLongText());
// 添加到channelGroup 通道组
_NettyConfig._getChannelGroup().add(ctx.channel());
}

/**
* 读取数据
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
MyLogUtil.info(“服务器收到消息：”+msg.text());

  _// 获取用户ID,关联channel_<br />_      _JSONObject jsonObject = JSONObject._fromObject_(msg.text());<br />      String uid = jsonObject.getString(**"uid"**);<br />      NettyConfig._getUserChannelMap_().put(uid, ctx.channel());
  _// 将用户ID作为自定义属性加入到channel中，方便随时channel中获取用户ID_<br />_      _AttributeKey<String> key = AttributeKey._valueOf_(**"userId"**);<br />      ctx.channel().attr(key).setIfAbsent(uid);
  _// 回复消息_<br />_      _ctx.channel().writeAndFlush(**new **TextWebSocketFrame(**"服务器连接成功！"**));<br />   }

@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
MyLogUtil.info(“handlerRemoved 被调用” + ctx.channel().id().asLongText());
// 删除通道
_NettyConfig._getChannelGroup().remove(ctx.channel());
removeUserId(ctx);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
MyLogUtil.info(“异常：” + cause.getMessage());
// 删除通道
_NettyConfig._getChannelGroup().remove(ctx.channel());
removeUserId(ctx);
ctx.close();
}

/**
* 删除用户与channel的对应关系
*
* @param __ctx
*/
private void removeUserId(ChannelHandlerContext ctx) {
AttributeKey key = AttributeKey.valueOf(“userId”);
String userId = ctx.channel().attr(key).get();
if(userId != null) {
NettyConfig.getUserChannelMap().remove(userId);
}
}
}

testsocket.html

<script>
var socket;
// 判断当前浏览器是否支持webSocket

if(window.WebSocket){
socket = new WebSocket(“ws://127.0.0.1:58080/webSocket”)
// 相当于channel的read事件，ev 收到服务器回送的消息
socket.onmessage = function (ev) {
var rt = document.getElementById(“username”);
rt.value = rt.value + “\n” + ev.data;
}
// 相当于连接开启
socket.onopen = function (ev) {
var rt = document.getElementById(“username”);
rt.value = “连接开启了…”
socket.send(
JSON.stringify({
// 连接成功将，用户ID传给服务端
uid: (new Date()).getTime()
})
);
}
// 相当于连接关闭
socket.onclose = function (ev) {
var rt = document.getElementById(“username”);
rt.value = rt.value + “\n” + “连接关闭了…”;
}
}else{
alert(“当前浏览器不支持webSocket”)
}

</script>