Springboot+Netty+Websocket实现消息推送实例
⽬录
前⾔
⼀、引⼊netty依赖
⼆、使⽤步骤
前⾔
WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在 WebSocket API 中，浏览器和服务器只需要完成⼀次握⼿，两者之间就直接可以创建持久性的连接，并进⾏双向数据传输。
Netty框架的优势
1. API使⽤简单，开发门槛低；
2. 功能强⼤，预置了多种编解码功能，⽀持多种主流协议；
3. 定制能⼒强，可以通过ChannelHandler对通信框架进⾏灵活地扩展；
4. 性能⾼，通过与其他业界主流的NIO框架对⽐，Netty的综合性能最优；
5. 成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发⼈员不需要再为NIO的BUG⽽烦恼
提⽰：以下是本篇⽂章正⽂内容，下⾯案例可供参考
⼀、引⼊netty依赖
<dependency>
<groupId>ioty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.48.Final</version>
</dependency>
⼆、使⽤步骤
1.引⼊基础配置类
stty;
public enum Cmd {
START("000", "连接成功"),
WMESSAGE("001", "消息提醒"),
;
private String cmd;
private String desc;
Cmd(String cmd, String desc) {
this.desc = desc;
}
public String getCmd() {
return cmd;
}
public String getDesc() {
return desc;
}
}
2ty服务启动
stty;
import ioty.bootstrap.ServerBootstrap;
import ioty.channel.ChannelFuture;
import ioty.channel.ChannelOption;
import ioty.channel.EventLoopGroup;
import ioty.channel.nio.NioEventLoopGroup;
import ioty.channel.socket.nio.NioServerSocketChannel;
slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationRunner;
import t.annotation.Bean;
import org.springframework.stereotype.Component;
/
**
* @author test
* <p>
* 服务启动
**/
@Slf4j
@Component
public class NettyServer {
@Value("${serverty.port}")
private int port;
@Autowired
private ServerChannelInitializer serverChannelInitializer;
@Bean
ApplicationRunner nettyRunner() {
return args -> {
//new ⼀个主线程组
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
websocket和socket//new ⼀个⼯作线程组
EventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap()
.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.
childHandler(serverChannelInitializer)
//设置队列⼤⼩
.option(ChannelOption.SO_BACKLOG, 1024)
// 两⼩时内没有数据的通信时,TCP会⾃动发送⼀个活动探测数据报⽂
.childOption(ChannelOption.SO_KEEPALIVE, true);
//绑定端⼝，开始接收进来的连接
try {
ChannelFuture future = bootstrap.bind(port).sync();
log.info("服务器启动开始监听端⼝: {}", port);
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//关闭主线程组
bossGroup.shutdownGracefully();
//关闭⼯作线程组
workGroup.shutdownGracefully();
}
};
}
}
3ty服务端处理器
stty;
stmon.util.JsonUtil;
import ioty.channel.Channel;
import ioty.channel.ChannelHandler;
import ioty.channel.ChannelHandlerContext;
import ioty.channel.SimpleChannelInboundHandler;
import dec.http.websocketx.TextWebSocketFrame;
import dec.http.websocketx.WebSocketServerProtocolHandler;
import lombok.Data;
slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.URLDecoder;
import java.util.*;
/**
* @author test
* <p>
* netty服务端处理器
**/
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Autowired
private ServerChannelCache cache;
private static final String dataKey = "test=";
@Data
public static class ChannelCache {
}
/**
* 客户端连接会触发
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
log.info("通道连接已打开，ID->{}......", channel.id().asLongText());
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
Channel channel = ctx.channel();
WebSocketServerProtocolHandler.HandshakeComplete handshakeComplete = (WebSocketServerProtocolHandler.HandshakeComplete) evt;
String requestUri = questUri();
requestUri = URLDecoder.decode(requestUri, "UTF-8");
log.info("HANDSHAKE_COMPLETE，ID->{}，URI->{}", channel.id().asLongText(), requestUri);
String socketKey = requestUri.substring(requestUri.lastIndexOf(dataKey) + dataKey.length());
if (socketKey.length() > 0) {
cache.add(socketKey, channel);
this.send(channel, Cmd.DOWN_START, null);
} else {
channel.disconnect();
ctx.close();
}
}
super.userEventTriggered(ctx, evt);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
log.info("通道连接已断开，ID->{}，⽤户ID->{}......", channel.id().asLongText(), CacheId(channel));
}
/**
* 发⽣异常触发
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
Channel channel = ctx.channel();
<("连接出现异常，ID->{}，⽤户ID->{}，异常->{}......", channel.id().asLongText(), CacheId(channel), Message(), cause); ve(channel);
ctx.close();
}
/**
* 客户端发消息会触发
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
try {
// log.info("接收到客户端发送的消息：{}", ());
ctx.channel().writeAndFlush(new String(Collections.singletonMap("cmd", "100"))));
} catch (Exception e) {
<("消息处理异常：{}", e.getMessage(), e);
}
}
public void send(Cmd cmd, String id, Object obj) {
HashMap<String, Channel> channels = (id);
if (channels == null) {
return;
}
Map<String, Object> data = new LinkedHashMap<>();
data.put("cmd", Cmd());
data.put("data", obj);
String msg = String(data);
log.info("服务器下发消息: {}", msg);
channels.values().forEach(channel -> {
channel.writeAndFlush(new TextWebSocketFrame(msg));
});
}
public void send(Channel channel, Cmd cmd, Object obj) {
Map<String, Object> data = new LinkedHashMap<>();
data.put("cmd", Cmd());
data.put("data", obj);
String msg = String(data);
log.info("服务器下发消息: {}", msg);
channel.writeAndFlush(new TextWebSocketFrame(msg));
}
}
4ty服务端缓存类
stty;
import ioty.channel.Channel;
import ioty.util.AttributeKey;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import urrent.ConcurrentHashMap;
@Component
public class ServerChannelCache {
private static final ConcurrentHashMap<String, HashMap<String, Channel>> CACHE_MAP = new ConcurrentHashMap<>(); private static final AttributeKey<String> CHANNEL_ATTR_KEY = AttributeKey.valueOf("test");
public String getCacheId(Channel channel) {
return channel.attr(CHANNEL_ATTR_KEY).get();
}
public void add(String cacheId, Channel channel) {
channel.attr(CHANNEL_ATTR_KEY).set(cacheId);
HashMap<String, Channel> hashMap = (cacheId);
if (hashMap == null) {
hashMap = new HashMap<>();
}
hashMap.put(channel.id().asShortText(), channel);
CACHE_MAP.put(cacheId, hashMap);
}
public HashMap<String, Channel> get(String cacheId) {
if (cacheId == null) {
return null;
}
return (cacheId);
}
public void remove(Channel channel) {
String cacheId = getCacheId(channel);
if (cacheId == null) {
return;
}
HashMap<String, Channel> hashMap = (cacheId);
if (hashMap == null) {
hashMap = new HashMap<>();
}
CACHE_MAP.put(cacheId, hashMap);
}
}
5ty服务初始化器
stty;
import ioty.channel.ChannelInitializer;
import ioty.channel.ChannelPipeline;
import ioty.channel.socket.SocketChannel;
import dec.http.HttpObjectAggregator;
import dec.http.HttpServerCodec;
import dec.http.websocketx.WebSocketServerProtocolHandler;
import ioty.handler.stream.ChunkedWriteHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author test
* <p>
* netty服务初始化器
**/
@Component
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Autowired
private NettyServerHandler nettyServerHandler;
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(8192));
pipeline.addLast(new WebSocketServerProtocolHandler("/test.io", true, 5000));
pipeline.addLast(nettyServerHandler);
}
}
6.html测试
<!DOCTYPE HTML>
<html>
<head>
<meta charset="utf-8">
<title>test</title>
<script type="text/javascript">
function WebSocketTest()
{
if ("WebSocket" in window)
{
alert("您的浏览器⽀持 WebSocket!");
// 打开⼀个 web socket
var ws = new WebSocket("ws://localhost:port/test.io");
{
// Web Socket 已连接上，使⽤ send() ⽅法发送数据
ws.send("发送数据");
alert("数据发送中...");
};
{
var received_msg = evt.data;
alert("数据已接收...");
};
{
// 关闭 websocket
alert("连接已关闭...");
};
}
else
{
// 浏览器不⽀持 WebSocket
alert("您的浏览器不⽀持 WebSocket!");
}
}
</script>
</head>
<body>
<div id="sse">
<a href="javascript:WebSocketTest()" rel="external nofollow" >运⾏ WebSocket</a>
</div>
</body>
</html>
7.vue测试
mounted() {
this.initWebsocket();
},
methods: {
initWebsocket() {
let websocket = new WebSocket('ws://localhost:port/test.io?test=123456');
let msg = JSON.parse(event.data);
switch (d) {
case "000":
this.$message({
type: 'success',
message: "建⽴实时连接成功！",
duration: 1000
})
setInterval(()=>{websocket.send("heartbeat")},60*1000);
break;
case "001":
this.$message.warning("收到⼀条新的信息，请及时查看！")
break;
}
}
setTimeout(()=>{
this.initWebsocket();
},30*1000);
}
setTimeout(()=>{
this.initWebsocket();
},30*1000);
}
},
},
;
到此这篇关于Springboot+Netty+Websocket实现消息推送实例的⽂章就介绍到这了，更多相关Springboot Websocket消息推送内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持！
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论