服务端
引入Netty依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.50.Final</version>
</dependency>
创建NettyServer
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 NettyServer {
private final int port;
public NettyServer(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<SocketChannel>() { // 绑定客户端连接时候触发操作
@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 MyWebSocketHandler());//自定义消息处理类
}
});
ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
System.out.println(NettyServer.class + "已启动，正在监听： " + cf.channel().localAddress());
cf.channel().closeFuture().sync(); // 关闭服务器通道
} finally {
group.shutdownGracefully().sync(); // 释放线程池资源
bossGroup.shutdownGracefully().sync();
}
}
}
创建MyWebSocketHandler
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("与客户端建立连接，通道开启！");
//添加到channelGroup通道组
MyChannelHandlerPool.channelGroup.add(ctx.channel());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("与客户端断开连接，通道关闭！");
//从channelGroup通道组删除
MyChannelHandlerPool.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){
String message = "消息";
ctx.writeAndFlush(new TextWebSocketFrame(message));
}
private void sendAllMessage(){
String message = "我是服务器，这是群发消息";
MyChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message));
}
}
创建MyChannelHandlerPool
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import java.util.concurrent.ConcurrentHashMap;
public class MyChannelHandlerPool {
public MyChannelHandlerPool() {
}
//可以存储userId与ChannelId的映射表
public static ConcurrentHashMap<String, ChannelId> channelIdMap = new ConcurrentHashMap<>();
//channelGroup通道组
public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}
启动（在SpringBoot启动类中加入以下内容）
import com.gnl.tool.netty.NettyServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ToolApplication {
public static void main(String[] args) {
SpringApplication.run(ToolApplication.class, args);
//在SpringBoot启动类中加入以下内容
try {
new NettyServer(8000).start();
} catch (Exception e) {
System.out.println("NettyServerError:" + e.getMessage());
}
}
}
客户端（微信小程序）
wxml
<view>接收的消息</view>
<textarea style="border: 1px solid black;" value="{{msg}}"></textarea>
<view>发送的消息</view>
<textarea style="border: 1px solid black;" bindinput='inputTitle'></textarea>
<button style="border: 1px solid violet;" bindtap='send'>发送消息</button>
<button style="border: 1px solid red;" bindtap='openWebSocket'>打开WebSocket连接</button>
<button style="border: 1px solid red;" bindtap='closeWebSocket'>关闭WebSocket连接</button>
js
Page({
/**
* 页面的初始数据
*/
data: {
},
/**
* 生命周期函数--监听页面加载
*/
onLoad: function (options) {
},
/**
* 绑定参数
*/
inputTitle: function(e){
var _this = this;
_this.setData({ title: e.detail.value })
},
/**
* 发送信息
*/
send: function(){
var _this = this;
wx.sendSocketMessage({
data: _this.data.title ,
success: (result) => {
console.log(result)
},
fail: (result) => {
console.log(result)
wx.showToast({
title: result.errMsg,
icon:"none",
duration:5000
})
},
})
},
/**
* 打开连接
*/
openWebSocket: function(){
//创建一个 WebSocket 连接
wx.connectSocket({
url: 'ws://10.8.51.124:8000/ws',
success: (result) => {
console.log(result)
},
})
//监听 WebSocket 连接打开事件
wx.onSocketOpen(function (result) {
console.log(result)
})
//监听 WebSocket 错误事件
wx.onSocketError(function (result) {
console.log(result)
wx.showToast({
title: result.errMsg,
})
})
//监听 WebSocket 接受到服务器的消息事件
wx.onSocketMessage((result) => {
console.log(result)
this.setData({
msg: result.data
})
})
//监听 WebSocket 关闭事件
wx.onSocketClose((result) => {
console.log(result)
})
},
/**
* 关闭连接
*/
closeWebSocket: function(){
wx.closeSocket({
success: (result) => {
console.log(result)
},
})
}
})
测试
1.点击打开WebSocket连接，服务端打印连接成功消息
2.在发送的消息框内输入hello world，点击发送消息，服务端接收到消息，同时群发消息给客户端
3.点击关闭WebSocket连接，服务端打印关闭连接成功消息同时关闭连接
网友评论