Я уже делал проект обмена мгновенными сообщениями, который включает в себя базовую функцию чата, поэтому обратите внимание, что эта серия статей не является упражнением, она не содержит базовой и пошаговой обучающей части.Если у вас есть энергия, вы можно продолжить серию статей.
Почему первая статья — это чат, чат — это самая простая часть реализации. Это также самая простая часть структуры обмена мгновенными сообщениями, за которой следуют одиночный чат и групповой чат. Бизнес-логика увеличивается слой за слоем, а код комнаты чата полностью удаляется. Расширенный одиночный чат и групповой чат очень просты, а в будущем я запущу комнату для прямых трансляций.
Если вы просто хотите реализовать чат, это очень просто, но я постараюсь пройти весь процесс для удобства понимания.
В основном реализуется двумя функциональными классами: класс инициализации + класс обработки ответов
0. Подготовка
добавить pom.xml
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.2.Final</version>
</dependency>
1. Реализация вспомогательного интерфейса
Вспомогательный интерфейс делает архитектуру сервиса более понятной.Здесь есть два интерфейса, один для обработки запросов Http, а другой для обработки запросов Websocket.
MyHttpService.java
/**
* 处理 http请求
*/
public interface MyHttpService {
void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request);
}
MyWebSocketService.java
/**
* 处理 WebSocket 请求中的frame
*/
public interface MyWebSocketService {
void handleFrame(ChannelHandlerContext ctx, WebSocketFrame frame);
}
Итак, вопрос в том, кто будет реализовывать эти два класса и кто будет заниматься распределением этих двух типов запросов.
Давайте посмотрим на класс обработки ответа службы: WebSocketServerHandler.java.
2. Класс обработки запроса
Наследуйте класс SimpleChannelInboundHandler и реализуйте
channelRead0()handlerAdded()handlerRemoved()exceptionCaught()и другие методы, первый из которых является обязательным, а остальные методы позволяют нам выполнять некоторую маркировку и последующую обработку.
WebSocketServerHandler.java
@Slf4j
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
private MyHttpService httpService;
private MyWebSocketService webSocketService;
public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
public WebSocketServerHandler(MyHttpService httpService, MyWebSocketService webSocketService) {
super();
this.httpService = httpService;
this.webSocketService = webSocketService;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof FullHttpRequest) {
httpService.handleHttpRequest(ctx, (FullHttpRequest) msg);
} else if (msg instanceof WebSocketFrame) {
webSocketService.handleFrame(ctx, (WebSocketFrame) msg);
}
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
channels.add(ctx.channel());
channels.writeAndFlush(new TextWebSocketFrame(ctx.channel() +"上线了"));
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
channels.remove(ctx.channel());
channels.writeAndFlush(new TextWebSocketFrame(ctx.channel() +"下线了"));
}
/**
* 发生异常时处理
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
channels.remove(ctx.channel());
ctx.close();
log.info("异常信息：{}",cause.getMessage());
}
}
- Создайте ChannelGroup для сохранения каждого канала, установившего соединение, в
handlerAdded()методchannels.add(ctx.channel());, соответственно вhandlerRemovedметодremove. - существует
channelRead0()В методе реализована идентификация и раздельная обработка запросов. -
exceptionCaught()Метод заключается в обработке, когда возникает исключение.
3. Инициализация
@Slf4j
public class WebSocketServer implements MyHttpService, MyWebSocketService {
/**
* 握手用的 变量
*/
private static final AttributeKey<WebSocketServerHandshaker> ATTR_HAND_SHAKER = AttributeKey.newInstance("ATTR_KEY_CHANNEL_ID");
private static final int MAX_CONTENT_LENGTH = 65536;
/**
* 请求类型常量
*/
private static final String WEBSOCKET_UPGRADE = "websocket";
private static final String WEBSOCKET_CONNECTION = "Upgrade";
private static final String WEBSOCKET_URI_ROOT_PATTERN = "ws://%s:%d";
/**
* 用户字段
*/
private String host;
private int port;
/**
* 保存 所有的连接
*/
private Map<ChannelId, Channel> channelMap = new HashMap<>();
private final String WEBSOCKET_URI_ROOT;
public WebSocketServer(String host, int port) {
this.host = host;
this.port = port;
// 将 ip 和端口 按照格式 赋值给 uri
WEBSOCKET_URI_ROOT = String.format(WEBSOCKET_URI_ROOT_PATTERN, host, port);
}
public void start(){
// 实例化 nio监听事件池
EventLoopGroup bossGroup = new NioEventLoopGroup();
// 实例化 nio工作线程池
EventLoopGroup workerGroup = new NioEventLoopGroup();
// 启动器
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup,workerGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel channel) throws Exception {
ChannelPipeline pl = channel.pipeline();
// 保存 该channel 到map中
channelMap.put(channel.id(),channel);
log.info("new channel {}",channel);
channel.closeFuture().addListener((ChannelFutureListener) channelFuture -> {
log.info("channel close future {}",channelFuture);
//关闭后 从map中移除
channelMap.remove(channelFuture.channel().id());
});
//添加 http 编解码
pl.addLast(new HttpServerCodec());
// 聚合器
pl.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH));
// 支持大数据流
pl.addLast(new ChunkedWriteHandler());
// 设置 websocket 服务处理方式
pl.addLast(new WebSocketServerHandler(WebSocketServer.this, WebSocketServer.this));
}
});
/**
* 实例化完毕后，需要完成端口绑定
*/
try {
ChannelFuture channelFuture = bootstrap.bind(host,port).addListener((ChannelFutureListener) channelFuture1 -> {
if (channelFuture1.isSuccess()){
log.info("webSocket started");
}
}).sync();
channelFuture.channel().closeFuture().addListener((ChannelFutureListener) channelFuture12 ->
log.info("server channel {} closed.", channelFuture12.channel())).sync();
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("绑定端口失败");
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
log.info("webSocket shutdown");
}
@Override
public void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
//判断是不是 socket 请求
if (isWebSocketUpgrade(request)){
//如果是webSocket请求
log.info("请求是webSocket协议");
// 获取子协议
String subProtocols = request.headers().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL);
//握手工厂 设置 uri+协议+不允许扩展
WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory(WEBSOCKET_URI_ROOT,subProtocols,false);
// 从工厂中实例化一个 握手请求
WebSocketServerHandshaker handshaker = handshakerFactory.newHandshaker(request);
if (handshaker == null){
//握手失败：不支持的协议
WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
}else {
//响应请求:将 握手转交给 channel处理
handshaker.handshake(ctx.channel(),request);
//将 channel 与 handshaker 绑定
ctx.channel().attr(ATTR_HAND_SHAKER).set(handshaker);
}
return;
}else {
// 不处理 HTTP 请求
log.info("不处理 HTTP 请求");
}
}
@Override
public void handleFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
/**
* text frame handler
*/
if (frame instanceof TextWebSocketFrame){
String text = ((TextWebSocketFrame) frame).text();
TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(text);
log.info("receive textWebSocketFrame from channel: {} , 目前一共有{}个在线",ctx.channel(),channelMap.size());
//发给其它的 channel (群聊功能)
for (Channel ch: channelMap.values()){
if (ch.equals(ctx.channel())){
continue;
}
//将 text frame 写出
ch.writeAndFlush(textWebSocketFrame);
log.info("消息已发送给{}",ch);
log.info("write text: {} to channel: {}",textWebSocketFrame,ctx.channel());
}
return;
}
/**
* ping frame , 回复 pong frame
*/
if (frame instanceof PingWebSocketFrame){
log.info("receive pingWebSocket from channel: {}",ctx.channel());
ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
return;
}
/**
* pong frame, do nothing
*/
if (frame instanceof PongWebSocketFrame){
log.info("receive pongWebSocket from channel: {}",ctx.channel());
return;
}
/**
* close frame, close
*/
if (frame instanceof CloseWebSocketFrame){
log.info("receive closeWebSocketFrame from channel: {}", ctx.channel());
//获取到握手信息
WebSocketServerHandshaker handshaker = ctx.channel().attr(ATTR_HAND_SHAKER).get();
if (handshaker == null){
log.error("channel: {} has no handShaker", ctx.channel());
return;
}
handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
return;
}
/**
* 剩下的都是 二进制 frame ，忽略
*/
log.warn("receive binary frame , ignore to handle");
}
/**
* 判断是否是 webSocket 请求
*/
private boolean isWebSocketUpgrade(FullHttpRequest req) {
HttpHeaders headers = req.headers();
return req.method().equals(HttpMethod.GET)
&& headers.get(HttpHeaderNames.UPGRADE).contains(WEBSOCKET_UPGRADE)
&& headers.get(HttpHeaderNames.CONNECTION).contains(WEBSOCKET_CONNECTION);
}
}
-
l.addLast(new WebSocketServerHandler(WebSocketServer.this, WebSocketServer.this));Добавьте свою собственную обработку ответов.WebSocketServerHandlerРеализуется ли второй точкой класс обработки запросов. -
private Map<ChannelId, Channel> channelMap = new HashMap<>();Чтобы сохранить ChannelId и Channel соответственно. Удобно получить соответственно позже. -
bootstrap.bind(host,port)Также можно заменить только портом привязки.
public ChannelFuture bind(String inetHost, int inetPort) {
return bind(new InetSocketAddress(inetHost, inetPort));
}
public synchronized InetAddress anyLocalAddress() {
if (anyLocalAddress == null) {
anyLocalAddress = new Inet4Address(); // {0x00,0x00,0x00,0x00}
anyLocalAddress.holder().hostName = "0.0.0.0";
}
return anyLocalAddress;
}
по умолчанию это даст0.0.0.0Порт открыт.
4.handleHttpRequestиhandleFrameдаMyWebSocketServiceРеализация класса.
5. У каждой детали есть примечания, внимательно читайте примечания.
4. Запустите службу
public class Main {
public static void main(String[] args) {
new WebSocketServer("192.168.1.33",9999).start();
}
}
Как протестировать в локальной сети?
Я использую службу обслуживания npm для работы в локальной сети. Введение на официальном сайте:Woohoo. That Horse Plus.com/package/purse… Моя статья:Рекомендации по упаковке React и создание службы статических файлов
Теперь убедитесь, что ваш мобильный телефон и компьютер находятся в локальной сети, вы можете получить доступ к собственному групповому чату.
5. Первая страница
Отправить комплект.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta http-equiv="X-UA-Compatible" content="ie=edge">
<title>Document</title>
<style type="text/css">
.talk_con{
width:600px;
height:500px;
border:1px solid #666;
margin:50px auto 0;
background:#f9f9f9;
}
.talk_show{
width:580px;
height:420px;
border:1px solid #666;
background:#fff;
margin:10px auto 0;
overflow:auto;
}
.talk_input{
width:580px;
margin:10px auto 0;
}
.whotalk{
width:80px;
height:30px;
float:left;
outline:none;
}
.talk_word{
width:420px;
height:26px;
padding:0px;
float:left;
margin-left:10px;
outline:none;
text-indent:10px;
}
.talk_sub{
width:56px;
height:30px;
float:left;
margin-left:10px;
}
.atalk{
margin:10px;
}
.atalk span{
display:inline-block;
background:#0181cc;
border-radius:10px;
color:#fff;
padding:5px 10px;
}
.btalk{
margin:10px;
text-align:right;
}
.btalk span{
display:inline-block;
background:#ef8201;
border-radius:10px;
color:#fff;
padding:5px 10px;
}
</style>
<script type="text/javascript">
//
document.onkeydown = function (ev) {
if (ev && ev.keyCode == 13){
send();
clear();
}
}
var socket;
if (window.WebSocket) {
socket = new WebSocket("ws://192.168.1.33:9999");
// socket = new WebSocket("ws://127.0.0.1:9999");
// socket = new WebSocket("ws://192.168.43.186:9999");
socket.onmessage = function (ev) {
atalkAppendIn("接收:"+socket.channel + ":" + ev.data)
};
socket.onopen = function () {
btalkAppendIn("连接已建立");
}
socket.onclose = function () {
btalkAppendIn("连接关闭");
};
}else {
alert("浏览器不支持");
}
function send(){
var message = document.getElementById("talkwords");
if (!window.WebSocket){
return
}
if (socket.readyState === WebSocket.OPEN){
socket.send(message.value);
btalkAppendIn("发送:"+ message.value);
clear();
} else {
alert("WebSocket 建立失败");
}
}
function atalkAppendIn(text) {
var append = document.getElementById("words");
append.innerHTML+= '<div class="atalk"><span>'+ text +'</span></div>';
}
function btalkAppendIn(text) {
var append = document.getElementById("words");
append.innerHTML+= '<div class="btalk"><span>'+ text +'</span></div>';
}
function clear () {
var elementById = document.getElementById("talkwords");
elementById.value = "";
}
</script>
</head>
<body>
<div class="talk_con">
<div class="talk_show" id="words">
</div>
<div class="talk_input">
<!--<select class="whotalk" id="who">-->
<!--<option value="0">A说：</option>-->
<!--<option value="1">B说：</option>-->
<!--</select>-->
<input type="text" class="talk_word" id="talkwords">
<input type="button" onclick="send()" value="发送" class="talk_sub" id="talksub">
</div>
</div>
</body>
</html>
-
socket = new WebSocket("ws://192.168.1.33:9999");Обратите внимание, что ip и порт здесь соответствуют службам один к одному. -
socket.onmessage()заключается в получении информации о сокете.socket.onopenзаключается в создании соединения.socket.oncloseзаключается в том, чтобы закрыть соединение.socket.send(message.value);Отправка информации о сокете.
Вывод консоли:
15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - receive textWebSocketFrame from channel: [id: 0x0d08c657, L:/192.168.1.33:9999 - R:/192.168.1.33:50440] , 目前一共有2个在线
15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - 消息已发送给[id: 0xacd5c1ad, L:/192.168.1.33:9999 - R:/192.168.1.33:50438]
15:12:42.444 [nioEventLoopGroup-3-5] DEBUG io.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder - Encoding WebSocket Frame opCode=1 length=5
15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - write text: TextWebSocketFrame(data: UnpooledUnsafeHeapByteBuf(ridx: 0, widx: 5, cap: 15)) to channel: [id: 0x0d08c657, L:/192.168.1.33:9999 - R:/192.168.1.33:50440]
Если вам понравилась моя статья, обратите внимание на мой официальный аккаунт. Официальный аккаунт посвящен анализу архитектурных технологий. Официальный аккаунт все еще находится в начальной стадии. Спасибо за вашу поддержку.
java架构Получите архитектурные видеоресурсы (другие высококачественные ресурсы будут опубликованы позже).