netty 基于 protobuf 协议 实现 websocket 版本的简易客服系统
2024-06-05 03:14:15
https://segmentfault.com/a/1190000017464313
netty 基于 protobuf 协议 实现 websocket 版本的简易客服系统
结构
- netty 作为服务端
- protobuf 作为序列化数据的协议
- websocket 前端通讯
演示
GitHub 地址
netty 服务端实现
Server.java 启动类
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;import java.net.InetSocketAddress;//websocket长连接示例
public class Server {public static void main(String[] args) throws Exception{// 主线程组EventLoopGroup bossGroup = new NioEventLoopGroup();// 从线程组EventLoopGroup wokerGroup = new NioEventLoopGroup();try{ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,wokerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ServerChannelInitializer());ChannelFuture channelFuture = serverBootstrap.bind(new InetSocketAddress(8899)).sync();channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();wokerGroup.shutdownGracefully();}}
}ServerChannelInitializer.java
import com.example.nettydemo.protobuf.MessageData;
import com.google.protobuf.MessageLite;
import com.google.protobuf.MessageLiteOrBuilder;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.stream.ChunkedWriteHandler;import java.util.List;import static io.netty.buffer.Unpooled.wrappedBuffer;public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// HTTP请求的解码和编码pipeline.addLast(new HttpServerCodec());// 把多个消息转换为一个单一的FullHttpRequest或是FullHttpResponse，// 原因是HTTP解码器会在每个HTTP消息中生成多个消息对象HttpRequest/HttpResponse,HttpContent,LastHttpContentpipeline.addLast(new HttpObjectAggregator(65536));// 主要用于处理大数据流，比如一个1G大小的文件如果你直接传输肯定会撑暴jvm内存的; 增加之后就不用考虑这个问题了pipeline.addLast(new ChunkedWriteHandler());// WebSocket数据压缩pipeline.addLast(new WebSocketServerCompressionHandler());// 协议包长度限制pipeline.addLast(new WebSocketServerProtocolHandler("/ws", null, true));// 协议包解码pipeline.addLast(new MessageToMessageDecoder<WebSocketFrame>() {@Overrideprotected void decode(ChannelHandlerContext ctx, WebSocketFrame frame, List<Object> objs) throws Exception {ByteBuf buf = ((BinaryWebSocketFrame) frame).content();objs.add(buf);buf.retain();}});// 协议包编码pipeline.addLast(new MessageToMessageEncoder<MessageLiteOrBuilder>() {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageLiteOrBuilder msg, List<Object> out) throws Exception {ByteBuf result = null;if (msg instanceof MessageLite) {result = wrappedBuffer(((MessageLite) msg).toByteArray());}if (msg instanceof MessageLite.Builder) {result = wrappedBuffer(((MessageLite.Builder) msg).build().toByteArray());}// ==== 上面代码片段是拷贝自TCP ProtobufEncoder 源码 ====// 然后下面再转成websocket二进制流，因为客户端不能直接解析protobuf编码生成的WebSocketFrame frame = new BinaryWebSocketFrame(result);out.add(frame);}});// 协议包解码时指定Protobuf字节数实例化为CommonProtocol类型pipeline.addLast(new ProtobufDecoder(MessageData.RequestUser.getDefaultInstance()));// websocket定义了传递数据的6中frame类型pipeline.addLast(new ServerFrameHandler());}
}ServerFrameHandler.java
import com.example.nettydemo.protobuf.MessageData;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;import java.util.List;//处理文本协议数据，处理TextWebSocketFrame类型的数据，websocket专门处理文本的frame就是TextWebSocketFrame
public class ServerFrameHandler extends SimpleChannelInboundHandler<MessageData.RequestUser> {private final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);//读到客户端的内容并且向客户端去写内容@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageData.RequestUser msg) throws Exception {// channelGroup.add();Channel channel = ctx.channel();System.out.println(msg.getUserName());System.out.println(msg.getAge());System.out.println(msg.getPassword());MessageData.ResponseUser bank = MessageData.ResponseUser.newBuilder().setUserName("你好,请问有什么可以帮助你!").setAge(18).setPassword("11111").build();channel.writeAndFlush(bank);}//每个channel都有一个唯一的id值@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {//打印出channel唯一值，asLongText方法是channel的id的全名// System.out.println("handlerAdded："+ctx.channel().id().asLongText());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {// System.out.println("handlerRemoved：" + ctx.channel().id().asLongText());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常发生");ctx.close();}}protobuf 文件的使用
proto 文件
syntax ="proto2";package com.example.nettydemo.protobuf;//optimize_for 加快解析的速度
option optimize_for = SPEED;
option java_package = "com.example.nettydemo.protobuf";
option java_outer_classname="MessageData";// 客户端发送过来的消息实体
message RequestUser{optional string user_name = 1;optional int32 age = 2;optional string password = 3;
}// 返回给客户端的消息实体
message ResponseUser{optional string user_name = 1;optional int32 age = 2;optional string password = 3;
}生成 proto 的Java 类
批量生成工具，直接找到这个 bat 或者 sh 文件，在对应的平台执行就可以了具体可以自行百度 protobuf 怎么使用
Windows 版本
set outPath=../../java
set fileArray=(MessageDataProto ATestProto)# 将.proto文件生成java类
for %%i in %fileArray% do (echo generate cli protocol java code: %%i.protoprotoc --java_out=%outPath% ./%%i.proto
)pausesh 版本 地址： https://github.com/lmxdawn/ne...
#!/bin/bashoutPath=../../java
fileArray=(MessageDataProto ATestProto)for i in ${fileArray[@]};
doecho "generate cli protocol java code: ${i}.proto"protoc --java_out=$outPath ./$i.proto
donewebsocket 实现
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>WebSocket客户端</title>
</head>
<body><script src="protobuf.min.js"></script><script type="text/javascript">var socket;//如果浏览器支持WebSocketif (window.WebSocket) {//参数就是与服务器连接的地址socket = new WebSocket("ws://localhost:8899/ws");//客户端收到服务器消息的时候就会执行这个回调方法socket.onmessage = function (event) {var ta = document.getElementById("responseText");// 解码responseUserDecoder({data: event.data,success: function (responseUser) {var content = "客服小姐姐: " + responseUser.userName +", 小姐姐年龄: " + responseUser.age +", 密码: " + responseUser.password;ta.value = ta.value + "\n" + content;},fail: function (err) {console.log(err);},complete: function () {console.log("解码全部完成")}})}//连接建立的回调函数socket.onopen = function (event) {var ta = document.getElementById("responseText");ta.value = "连接开启";}//连接断掉的回调函数socket.onclose = function (event) {var ta = document.getElementById("responseText");ta.value = ta.value + "\n" + "连接关闭";}} else {alert("浏览器不支持WebSocket！");}//发送数据function send(message) {if (!window.WebSocket) {return;}// socket.binaryType = "arraybuffer";// 判断是否开启if (socket.readyState !== WebSocket.OPEN) {alert("连接没有开启");return;}var data = {userName: message,age: 18,password: "11111"};requestUserEncoder({data: data,success: function (buffer) {console.log("编码成功");socket.send(buffer);},fail: function (err) {console.log(err);},complete: function () {console.log("编码全部完成")}});}/*** 发送的消息编码成 protobuf*/function requestUserEncoder(obj) {var data = obj.data;var success = obj.success; // 成功的回调var fail = obj.fail; // 失败的回调var complete = obj.complete; // 成功或者失败都会回调protobuf.load("../proto/MessageDataProto.proto", function (err, root) {if (err) {if (typeof fail === "function") {fail(err)}if (typeof complete === "function") {complete()}return;}// Obtain a message typevar RequestUser = root.lookupType("com.example.nettydemo.protobuf.RequestUser");// Exemplary payloadvar payload = data;// Verify the payload if necessary (i.e. when possibly incomplete or invalid)var errMsg = RequestUser.verify(payload);if (errMsg) {if (typeof fail === "function") {fail(errMsg)}if (typeof complete === "function") {complete()}return;}// Create a new messagevar message = RequestUser.create(payload); // or use .fromObject if conversion is necessary// Encode a message to an Uint8Array (browser) or Buffer (node)var buffer = RequestUser.encode(message).finish();if (typeof success === "function") {success(buffer)}if (typeof complete === "function") {complete()}});}/*** 接收到服务器二进制流的消息进行解码*/function responseUserDecoder(obj) {var data = obj.data;var success = obj.success; // 成功的回调var fail = obj.fail; // 失败的回调var complete = obj.complete; // 成功或者失败都会回调protobuf.load("../proto/MessageDataProto.proto", function (err, root) {if (err) {if (typeof fail === "function") {fail(err)}if (typeof complete === "function") {complete()}return;}// Obtain a message typevar ResponseUser = root.lookupType("com.example.nettydemo.protobuf.ResponseUser");var reader = new FileReader();reader.readAsArrayBuffer(data);reader.onload = function (e) {var buf = new Uint8Array(reader.result);var responseUser = ResponseUser.decode(buf);if (typeof success === "function") {success(responseUser)}if (typeof complete === "function") {complete()}}});}
</script><h1>欢迎访问客服系统</h1><form onsubmit="return false"><textarea name="message" style="width: 400px;height: 200px"></textarea><input type="button" value="发送数据" onclick="send(this.form.message.value);"><h3>回复消息：</h3><textarea id="responseText" style="width: 400px;height: 300px;"></textarea><input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空数据">
</form>
</body>
</html>扩展阅读
spring boot 实现的后台管理系统
vue + element-ui 实现的后台管理界面，接入 spring boot API接口
netty 基于 protobuf 协议 实现 websocket 版本的简易客服系统相关推荐
- 基于springboot+h5+websocket的即时通讯客服系统和百度实时语音转译(语音在线识别)
本文章由本人原创 下载链接:https://download.csdn.net/download/u014191624/51948075 这是一个基于springboot+h5+websocket的即 ...
- Android 4.1.2微信版本,ttkefu在线客服系统
ttkefu在线客服系统是一款智能在线客服,该客服软件可以帮助客服人员去智能解答客户遇到的各种问题,提高的顾客的满意度,所以快来下载试试吧! 软件介绍 TTKEFU是一款免费的即时聊天软件,用户只需将 ...
- netty 对 protobuf 协议的解码与包装探究(2)
netty 默认支持protobuf 的封装与解码,如果通信双方都使用netty则没有什么障碍,但如果客户端是其它语言(C#)则需要自己仿写与netty一致的方式(解码+封装),提前是必须很了解net ...
- 基于stomp协议的websocket
1.依赖 <dependency><groupId>org.springframework.boot</groupId><artifactId>spri ...
- Linux + .net core 开发升讯威在线客服系统：首个经过实际验证的高性能版本
业余时间用 .net core 写了一个在线客服系统.并在博客园写了一个系列的文章,写介绍这个开发过程: .net core 和 WPF 开发升讯威在线客服系统:目录 https://blog.she ...
- 定时采用ajax方式获得数据库,《基于Ajax的在线客服系统的设计与实现》-毕业设计论文（学术).doc...
PAGE 2 西安文理学院 数学与计算机工程学院 本科毕业设计(论文) (2012届) 设计题目 基于Ajax的在线客服系统的设计与实现 Design And Implementation Of On ...
- WEB在线客服系统（websocket+Golang）
真正的大师,永远都怀着一颗学徒的心! 一.项目简介 WEB在线客服系统,项目使用golang开发的,手机和电脑上都是可以自适应的.可以展示在网页页面右下角,只需要一段js代码,就可以实现功能.缩小后以 ...
- ZBLOG即时聊天(客服)插件v1.3.2版本下载，强大的在线客服系统源码
ZBLOG即时聊天(客服)插件v1.3.2版本下载,强大的在线客服系统源码 测试报告:本插件我已按照测试了,正常按照开启,插件配置可正常顺利保存,后台配置项无问题.不过因为本插件要配置的东西太多了,且 ...
- 亲测无限坐席在线客服系统源码，基于ThinkPHP的一款在线客服系统源码
源码简介 东西没问题,和别人换的本来说是多语言带机器人翻译之类的,给了个这... 直接一键安装的,启动两个端口就行了,安装倒是简单 编号:ym270 品牌:无 语言:PHP 大小:34.5MB 类型: ...
最新文章
- R语言基于自定义函数构建xgboost模型并使用LIME解释器进行模型预测结果解释：基于训练数据以及模型构建LIME解释器解释多个iris数据样本的预测结果、使用LIME解释器进行模型预测结果解释
- boostrap 鼠标滚轮滑动图片_BootStrap 轮播插件(carousel)支持左右手势滑动的方法(三种)...
- IOS开发之----远程推送通知
- 解决Tomcat下IntelliJ IDEA报错java.lang.NoClassDefFoundError: javax/servlet/ServletContextListener
- 在linux摸索的过程
- 【滤波器】基于matlab升余弦滤波器【含Matlab源码 993期】
- 揭秘ServerBootstrap神秘面纱（服务端ServerBootstrap）
- 大学学习路线规划建议贴
- 资江小票打印机js实现web打印（web通用打印）
- 阿里云中标“金关工程二期”大数据云项目，总金额8568万！（含标单）
- 软件测试学习之悟空CRM项目测试
- pr剪辑视频转码问题以及子剪辑
- java freemarker导出word时添加或勾选复选框
- 2022年大数据技能大赛国赛（模块A，B）
- MySQL中利用经纬度计算两点之间的距离
- make_unique的使用
- 【知识图谱 赵军 学习笔记】第十章 知识问答与对话
- 数据管理体系之数据质量
- 51单片机（二）.STC89C52单片机的引脚功能
- 如何免费注册好用的电子邮箱呢？