基于 Netty 搭建 WebSocket 集群实现服务器消息推送

<dependencies>
<dependency>
<groupId>io.nettygroupId>
<artifactId>netty-allartifactId>
<version>4.1.36.Finalversion>
dependency>
dependencies>

websocket:
 port: 7000 #端口
 url: /msg #访问url

package com.wander.netty.websocket.yeelight;
import lombok.Data;
import java.io.Serializable;

@Data
public class MessageRequest implements Serializable {

private Long unionId;

private Integer current = 1;

private Integer size = 10;
}

/**
 * @Author WDYin
 * @Date 2021/6/10
 * @Description websocket初始化器
 **/
@Slf4j
@Component
public class WebsocketInitialization {

@Resource
private WebsocketChannelInitializer websocketChannelInitializer;

@Value("${websocket.port}")
private Integer port;

@Async
public void init() throws InterruptedException {

//bossGroup连接线程组，主要负责接受客户端连接，一般一个线程足矣
 EventLoopGroup bossGroup = new NioEventLoopGroup(1);
//workerGroup工作线程组，主要负责网络IO读写
 EventLoopGroup workerGroup = new NioEventLoopGroup();
try {

//启动辅助类
 ServerBootstrap serverBootstrap = new ServerBootstrap();
//bootstrap绑定两个线程组
 serverBootstrap.group(bossGroup, workerGroup);
//设置通道为NioChannel
 serverBootstrap.channel(NioServerSocketChannel.class);
//可以对入站\出站事件进行日志记录，从而方便我们进行问题排查。
 serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
//设置自定义的通道初始化器，用于入站操作
 serverBootstrap.childHandler(websocketChannelInitializer);
//启动服务器,本质是Java程序发起系统调用，然后内核底层起了一个处于监听状态的服务，生成一个文件描述符FD
 ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
//异步
 channelFuture.channel().closeFuture().sync();

 } finally {
 bossGroup.shutdownGracefully();
 workerGroup.shutdownGracefully();
 }
 }

}

/**
 * @Author WDYin
 * @Date 2021/6/10
 * @Description websocket通道初始化器
 **/
@Component
public class WebsocketChannelInitializer extends ChannelInitializer<SocketChannel> {

@Autowired
private WebSocketHandler webSocketHandler;

@Value("${websocket.url}")
private String websocketUrl;

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {

//获取pipeline通道
 ChannelPipeline pipeline = socketChannel.pipeline();
//因为基于http协议，使用http的编码和解码器
 pipeline.addLast(new HttpServerCodec());
//是以块方式写，添加ChunkedWriteHandler处理器
 pipeline.addLast(new ChunkedWriteHandler());
/*
 说明
 1. http数据在传输过程中是分段, HttpObjectAggregator ，就是可以将多个段聚合
 2. 这就就是为什么，当浏览器发送大量数据时，就会发出多次http请求
 */
 pipeline.addLast(new HttpObjectAggregator(8192));
/* 说明
 1. 对应websocket ，它的数据是以 帧(frame) 形式传递
 2. 可以看到WebSocketFrame 下面有六个子类
 3. 浏览器请求时 ws://localhost:7000/msg 表示请求的uri
 4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 保持长连接
 5. 是通过一个 状态码 101
 */
 pipeline.addLast(new WebSocketServerProtocolHandler(websocketUrl));
//自定义的handler ，处理业务逻辑
 pipeline.addLast(webSocketHandler);
 }
}

/**
 * @Author WDYin
 * @Date 2021/6/10
 * @Description websocket处理器
 **/
@Slf4j
@Component
@ChannelHandler.Sharable//保证处理器，在整个生命周期中就是以单例的形式存在，方便统计客户端的在线数量
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

//通道map，存储channel，用于群发消息，以及统计客户端的在线数量，解决问题问题三，如果是集群环境使用redis的hash数据类型存储即可
private static Map channelMap = new ConcurrentHashMap<>();
//任务map，存储future，用于停止队列任务
private static Map futureMap = new ConcurrentHashMap<>();
//存储channel的id和用户主键的映射，客户端保证用户主键传入的是唯一值，解决问题四，如果是集群中需要换成redis的hash数据类型存储即可
private static Map clientMap = new ConcurrentHashMap<>();

/**
 * 客户端发送给服务端的消息
 *
 * @param ctx
 * @param msg
 * @throws Exception
 */
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {

try {
//接受客户端发送的消息
 MessageRequest messageRequest = JSON.parseObject(msg.text(), MessageRequest.class);

//每个channel都有id，asLongText是全局channel唯一id
 String key = ctx.channel().id().asLongText();
//存储channel的id和用户的主键
 clientMap.put(key, messageRequest.getUnionId());
 log.info("接受客户端的消息......" + ctx.channel().remoteAddress() + "-参数[" + messageRequest.getUnionId() + "]");

if (!channelMap.containsKey(key)) {
//使用channel中的任务队列，做周期循环推送客户端消息，解决问题二和问题五
 Future future = ctx.channel().eventLoop().scheduleAtFixedRate(new WebsocketRunnable(ctx, messageRequest), 0, 10, TimeUnit.SECONDS);
//存储客户端和服务的通信的Chanel
 channelMap.put(key, ctx.channel());
//存储每个channel中的future，保证每个channel中有一个定时任务在执行
 futureMap.put(key, future);
 } else {
//每次客户端和服务的主动通信，和服务端周期向客户端推送消息互不影响 解决问题一
 ctx.channel().writeAndFlush(new TextWebSocketFrame(Thread.currentThread().getName() + "服务器时间" + LocalDateTime.now() + "wdy"));
 }

 } catch (Exception e) {

 log.error("websocket服务器推送消息发生错误：", e);

 }
 }

/**
 * 客户端连接时候的操作
 *
 * @param ctx
 * @throws Exception
 */
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 log.info("一个客户端连接......" + ctx.channel().remoteAddress() + Thread.currentThread().getName());
 }

/**
 * 客户端掉线时的操作
 *
 * @param ctx
 * @throws Exception
 */
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {

 String key = ctx.channel().id().asLongText();
//移除通信过的channel
 channelMap.remove(key);
//移除和用户绑定的channel
 clientMap.remove(key);
//关闭掉线客户端的future
 Optional.ofable(futureMap.get(key)).ifPresent(future -> {
 future.cancel(true);
 futureMap.remove(key);
 });
 log.info("一个客户端移除......" + ctx.channel().remoteAddress());
 ctx.close(); //关闭连接
 }

/**
 * 发生异常时执行的操作
 *
 * @param ctx
 * @param cause
 * @throws Exception
 */
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 String key = ctx.channel().id().asLongText();
//移除通信过的channel
 channelMap.remove(key);
//移除和用户绑定的channel
 clientMap.remove(key);
//移除定时任务
 Optional.ofable(futureMap.get(key)).ifPresent(future -> {
 future.cancel(true);
 futureMap.remove(key);
 });
//关闭长连接
 ctx.close();
 log.info("异常发生 " + cause.getMessage());
 }

public static Map getChannelMap() {
return channelMap;
 }

public static Map getFutureMap() {
return futureMap;
 }

public static Map getClientMap() {
return clientMap;
 }
}

/**
 * @Author WDYin
 * @Date 2021/8/10
 * @Description websocket程序
 **/
@Slf4j
@Component
public class WebsocketApplication {

@Resource
private WebsocketInitialization websocketInitialization;

@PostConstruct
public void start() {
try {
 log.info(Thread.currentThread().getName() + ":websocket启动中......");
 websocketInitialization.init();
 log.info(Thread.currentThread().getName() + ":websocket启动成功！！！");
 } catch (Exception e) {
 log.error("websocket发生错误：",e);
 }
 }
}

/**
 * @Author WDYin
 * @Date 2021/8/10
 * @Description websocket程序
 **/
@Slf4j
@Component
public class WebsocketApplication {

@Resource
private WebsocketInitialization websocketInitialization;

@PostConstruct
public void start() {
try {
 log.info(Thread.currentThread().getName() + ":websocket启动中......");
 websocketInitialization.init();
 log.info(Thread.currentThread().getName() + ":websocket启动成功！！！");
 } catch (Exception e) {
 log.error("websocket发生错误：",e);
 }
 }
}

/**
 * @Author WDYin
 * @Date 2021/9/12
 * @Description
 **/
@RequestMapping("index")
@Controller
public class WebsocketController {

/**
 *
 * @param id 用户主键
 * @param idList 要把消息发送给其他用户的主键
 */
@RequestMapping("hello1")
private void hello(Long id, List idList){
//获取所有连接的客户端,如果是集群环境使用redis的hash数据类型存储即可
 Map channelMap = WebSocketHandler.getChannelMap();
//获取与用户主键绑定的channel,如果是集群环境使用redis的hash数据类型存储即可
 Map clientMap = WebSocketHandler.getClientMap();
//解决问题六,websocket集群中一个客户端向其他客户端主动发送消息，如何实现？
 clientMap.forEach((k,v)->{
if (idList.contains(v)){
 Channel channel = channelMap.get(k);
 channel.eventLoop().execute(() -> channel.writeAndFlush(new TextWebSocketFrame(Thread.currentThread().getName()+"服务器时间" + LocalDateTime.now() + "wdy")));
 }
 });
 }
}

<html lang="en">
<head>
<meta charset="UTF-8">
<title>Titletitle>
head>
<body>
<script>
var socket;
//判断当前浏览器是否支持websocket
if(window.WebSocket) {
 socket = new WebSocket("ws://localhost:7000/msg");
//相当于channelReado, ev 收到服务器端回送的消息
 socket.onmessage = function (ev) {
var rt = document.getElementById("responseText");
 rt.value = rt.value + "\n" + ev.data;
 }

//相当于连接开启(感知到连接开启)
 socket.onopen = function (ev) {
var rt = document.getElementById("responseText");
 rt.value = "连接开启了.."
 }

//相当于连接关闭(感知到连接关闭)
 socket.onclose = function (ev) {

var rt = document.getElementById("responseText");
 rt.value = rt.value + "\n" + "连接关闭了.."
 }
 } else {
 alert("当前浏览器不支持websocket")
 }

//发送消息到服务器
function send(websocketMessage) {
if(!window.socket) { //先判断socket是否创建好
return;
 }
if(socket.readyState == WebSocket.OPEN) {
//通过socket 发送消息
 socket.send(websocketMessage)
 } else {
 alert("连接没有开启");
 }
 }
script>
<form onsubmit="return false">
<textarea name="websocketMessage" style="height: 300px; width: 300px">textarea>
<input type="button" value="发生消息" onclick="send(this.form.websocketMessage.value)">
<textarea id="responseText" style="height: 300px; width: 300px">textarea>
<input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">
form>
body>
html>

#第一步：
upstream websocket-router {
server 192.168.1.31:7000 max_fails=10 weight=1 fail_timeout=5s;
keepalive 1000;
}
#第二步：
server {
listen 80; #监听80端口
server_name websocket.wdy.com; #域名配置

ssl_session_cache shared:SSL:1m; 
ssl_session_timeout 5m;

ssl_ciphers HIGH:!a:!MD5;
ssl_prefer_server_ciphers on;

location / {
client_max_body_size 100M;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header Upgrade $http_upgrade; #支持wss
proxy_set_header Connection "upgrade"; #支持wssi
proxy_pass http://websocket-router; #代理路由
root html;
index index.html index.htm;
 }
 }

链接：https://blog.csdn.net/qq_41889508/article/details/105953114