AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

30分钟快速打造一个完善的直播聊天系统

2016-08-11 7470
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 下面的代码基于高性能的通信王牌工具 Netty。我们将一些实际场景都添加进去，比如用户身份的验证，游客只能浏览不能发言，多房间（频道）的聊天。 这篇文章非常适合和我一样的 Java 新手，适合作为学习 Java 的切入点，不需要考虑tomcat、spring、mybatis等。唯一的知识点就是 m

下面的代码基于高性能的通信王牌工具 Netty。我们将一些实际场景都添加进去，比如用户身份的验证，游客只能浏览不能发言，多房间（频道）的聊天。

这篇文章非常适合和我一样的 Java 新手，适合作为学习 Java 的切入点，不需要考虑tomcat、spring、mybatis等。唯一的知识点就是 maven 的基础使用。

完整的代码地址

https://github.com/zhoumengkang/netty-websocket

项目结果如下

├── WebSocketServer.java                启动服务器端口监听
├── WebSocketServerInitializer.java     初始化服务
├── WebSocketServerHandler.java         接管WebSocket数据连接
├── dto
│   └── Response.java                   返回给客户端数据对象
├── entity
│   └── Client.java                     每个连接到WebSocket服务的客户端对象
└── service
    ├── MessageService.java             完成发送消息
    └── RequestService.java             WebSocket初始化连接握手时的数据处理

功能设计概述

身份认证

客户端将用户 id 、进入的房间的 rid、用户 token json_encode，例如{id:1;rid:21;token:'xxx'}。然后在 base64 处理，通过参数request传到服务器，然后在服务器做 id 和 token 的验证（我的做法是 token 存放在redis string 5秒的过期时间）

房间表

使用一个Map channelGroupMap 来存放各个房间（频道），以客户端传握手时传过来的base64 字符串中获取到定义的房间 ID，然后为该房间 ID 新建一个ChannelGroup（ChannelGroup 方便对该组内的所有客户端广播消息）

在 pom.xml 中引入netty 5

现在大家都有自己的包管理工具，不需要实现下载了然后放到本地lib库中，和 nodejs 的 npm, php 的 compser 一样。

<dependencies>
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>5.0.0.Alpha2</version>
    </dependency>
    <dependency>
        <groupId>com.jcraft</groupId>
        <artifactId>jzlib</artifactId>
        <version>1.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.json</groupId>
        <artifactId>json</artifactId>
        <version>20141113</version>
    </dependency>
    <dependency>
        <groupId>commons-codec</groupId>
        <artifactId>commons-codec</artifactId>
        <version>1.10</version>
    </dependency>
</dependencies>

创建服务器

package net.mengkang;
 
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
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;
 
 
public final class WebSocketServer {
 
    private static final int PORT = 8083;
 
    public static void main(String[] args) throws Exception {
 
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new WebSocketServerInitializer());
 
            Channel ch = b.bind(PORT).sync().channel();
            ch.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

package net.mengkang;
 
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
 
 
public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
 
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
 
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new WebSocketServerCompressionHandler());
        pipeline.addLast(new WebSocketServerHandler());
    }
}

处理长连接

下面程序中最的处理在握手阶段handleHttpRequest，里面处理参数的判断，用户的认证，登录用户表的维护，直播房间表维护。详细的请大家对照代码来浏览。

握手完成之后的消息传递则在handleWebSocketFrame中处理。

整理的执行流程，大家可以对各个方法打断点予以调试，就会很清楚整个执行的脉络啦。

package net.mengkang;
 
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import net.mengkang.dto.Response;
import net.mengkang.entity.Client;
import net.mengkang.service.MessageService;
import net.mengkang.service.RequestService;
import org.json.JSONObject;
 
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
import static io.netty.handler.codec.http.HttpHeaderNames.HOST;
import static io.netty.handler.codec.http.HttpMethod.GET;
import static io.netty.handler.codec.http.HttpResponseStatus.*;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
 
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
 
    // websocket 服务的 uri
    private static final String WEBSOCKET_PATH = "/websocket";
 
    // 一个 ChannelGroup 代表一个直播频道
    private static Map<Integer, ChannelGroup> channelGroupMap = new HashMap<>();
 
    // 本次请求的 code
    private static final String HTTP_REQUEST_STRING = "request";
 
    private Client client = null;
 
    private WebSocketServerHandshaker handshaker;
 
    @Override
    public void messageReceived(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof FullHttpRequest) {
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }
 
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }
 
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
        // Handle a bad request.
        if (!req.decoderResult().isSuccess()) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
            return;
        }
 
        // Allow only GET methods.
        if (req.method() != GET) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
            return;
        }
 
        if ("/favicon.ico".equals(req.uri()) || ("/".equals(req.uri()))) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
            return;
        }
 
        QueryStringDecoder queryStringDecoder = new QueryStringDecoder(req.uri());
        Map<String, List<String>> parameters = queryStringDecoder.parameters();
 
        if (parameters.size() == 0 || !parameters.containsKey(HTTP_REQUEST_STRING)) {
            System.err.printf(HTTP_REQUEST_STRING + "参数不可缺省");
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
            return;
        }
 
        client = RequestService.clientRegister(parameters.get(HTTP_REQUEST_STRING).get(0));
        if (client.getRoomId() == 0) {
            System.err.printf("房间号不可缺省");
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));
            return;
        }
 
        // 房间列表中如果不存在则为该频道,则新增一个频道 ChannelGroup
        if (!channelGroupMap.containsKey(client.getRoomId())) {
            channelGroupMap.put(client.getRoomId(), new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
        }
        // 确定有房间号,才将客户端加入到频道中
        channelGroupMap.get(client.getRoomId()).add(ctx.channel());
 
        // Handshake
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(getWebSocketLocation(req), null, true);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            ChannelFuture channelFuture = handshaker.handshake(ctx.channel(), req);
 
            // 握手成功之后,业务逻辑
            if (channelFuture.isSuccess()) {
                if (client.getId() == 0) {
                    System.out.println(ctx.channel() + " 游客");
                    return;
                }
 
            }
        }
    }
 
    private void broadcast(ChannelHandlerContext ctx, WebSocketFrame frame) {
 
        if (client.getId() == 0) {
            Response response = new Response(1001, "没登录不能聊天哦");
            String msg = new JSONObject(response).toString();
            ctx.channel().write(new TextWebSocketFrame(msg));
            return;
        }
 
        String request = ((TextWebSocketFrame) frame).text();
        System.out.println(" 收到 " + ctx.channel() + request);
 
        Response response = MessageService.sendMessage(client, request);
        String msg = new JSONObject(response).toString();
        if (channelGroupMap.containsKey(client.getRoomId())) {
            channelGroupMap.get(client.getRoomId()).writeAndFlush(new TextWebSocketFrame(msg));
        }
 
    }
 
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
 
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
        }
 
        broadcast(ctx, frame);
    }
 
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        if (res.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            HttpHeaderUtil.setContentLength(res, res.content().readableBytes());
        }
 
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpHeaderUtil.isKeepAlive(req) || res.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
 
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        System.out.println("收到" + incoming.remoteAddress() + " 握手请求");
    }
 
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        if (client != null && channelGroupMap.containsKey(client.getRoomId())) {
            channelGroupMap.get(client.getRoomId()).remove(ctx.channel());
        }
    }
 
    private static String getWebSocketLocation(FullHttpRequest req) {
        String location = req.headers().get(HOST) + WEBSOCKET_PATH;
        return "ws://" + location;
    }
}

客户端程序

<html>
<head><title></title></head>
<body>
<script type="text/javascript">
var socket;
if (!window.WebSocket) {
  window.WebSocket = window.MozWebSocket;
}
if (window.WebSocket) {
  socket = new WebSocket("ws://localhost:8083/websocket/?request=e2lkOjE7cmlkOjI2O3Rva2VuOiI0MzYwNjgxMWM3MzA1Y2NjNmFiYjJiZTExNjU3OWJmZCJ9");
  socket.onmessage = function(event) {
      console.log(event.data);
  };
  socket.onopen = function(event) {
    console.log("websocket 打开了");
  };
  socket.onclose = function(event) {
    console.log("websocket 关闭了");
  };
}
 
function send(message) {
  if (!window.WebSocket) { return; }
  if (socket.readyState == WebSocket.OPEN) {
    socket.send(message);
  } else {
    alert("The socket is not open.");
  }
}
</script>
<form onsubmit="return false;">
  <input type="text" name="message" value="Hello, World!"/>
  <input type="button" value="Send Web Socket Data" onclick="send(this.form.message.value)" />
</form>
</body>
</html>

可以伪造多个房间（修改客户端请求的request参数），在 php cli 模式运行下命令来获取 
php -r "echo base64_encode('{id:1;rid:26;token:\"xxx\"}');"

并发压测

感谢同事 https://github.com/ideal 写的压测脚本

https://github.com/zhoumengkang/netty-websocket/blob/master/benchmark.py

并测试为N个客户端，每个客户端发送10条消息，服务器配置2核4G内存，广播给所有的客户端，我们测试1000个并发的时候，负载在后期陡升。

实际情况下，不可能那么多人同时说话广播，而是说话的人少，接受广播的人多。

实际线上之后，在不限制刷帖频率大家狂轰滥炸的情况下，1500多人在线，半小时，负载一直都处于0.5以下。

文章标签：
Java
PHP
周梦康
目录
相关文章
软件求生
|
1月前
|
存储 编解码 算法
带货直播这么流畅，原来是这套技术系统在支撑！
大家好，我是小米。今天聊聊社区直播带货的流程。主播通过RTMPS协议将加密直播流发送至POP内的代理服务器，再由代理服务器转发至数据中心的网关服务器，经端口转换后，使用一致性哈希算法分配至编码服务器进行转码和输出，最终通过DASH协议实现流畅直播及持久化存储，确保高效稳定的直播体验。这一流程背后有复杂的技术支撑，希望能帮大家更好地理解直播背后的机制。
软件求生
33 2
东莞梦幻网络科技
|
6月前
|
开发者
提升用户黏性：现成体育直播源码开发设计哪些关键功能
面对激烈的市场竞争，如何通过关键功能设计提升用户黏性，使之成为用户长期依赖的首选平台，是每一个开发者必须深思的问题。如下参考“东莞梦幻网络科技”现成体育直播源码，为了吸引更多用户并提高他们的黏性，开发哪些关键功能，帮助实现这一目标：
东莞梦幻网络科技
43 0
山东布谷科技魏
|
编解码 运维 安全
直播系统源码，系统分析篇：不可或缺的云转码系统
云转码系统作为直播系统源码平台不可或缺的系统之一，它的根据不同设备终端与网络状况等因素变换质量，让平台用户能够在不同的时间和地点享受到高清、低延迟的优质体验，并且可以在多种场景中起到关键的作用，云转码系统使直播系统源码平台具有广泛的应用前景和价值。
山东布谷科技魏
138 0
直播系统源码，系统分析篇：不可或缺的云转码系统
游客3eenxn3lyh7ye
|
缓存 NoSQL 前端开发
浅析开发体育赛事直播系统的设计与实现
东莞梦幻网络科技的“体育赛事直播源码”主要是用于搭建类似于雷速体育和斗球体育等平台，该系统的出现能帮助快速搭建平台和降低开发成本。
游客3eenxn3lyh7ye
200 0
云豹科技晓彤
|
存储 编解码 网络协议
分析婚恋系统源码的直播功能架构是怎样的
直播功能实现起来比较复杂，技术人员要完善直播系统架构，让直播更加稳定，那么婚恋系统源码的直播系统架构是怎样的呢？
云豹科技晓彤
375 0
后端老鸟
|
Web App开发 编解码 JavaScript
互动直播之WebRTC服务开源技术选型
介绍了直播的基础知识，对比几种传输标得出WebRTC的优势，常见的WebRTC架构及开源方案。
后端老鸟
4336 0
互动直播之WebRTC服务开源技术选型
云豹短视频嘉兴
|
存储 算法 程序员
一对一直播系统开发的原理只需记好这七点
直播行业如火如荼的加速前进，一对一直播系统开发开发紧跟着脚步加速前进，你知道一对一直播系统开发开发应当具备哪些条件吗？
云豹短视频嘉兴
841 0
一对一直播系统开发的原理只需记好这七点
yunbao111
|
开发工具 CDN
一对一直播系统开发，实现多人聊天有哪些开发难点
一对一直播系统开发，实现多人聊天有哪些开发难点
yunbao111
537 0
yunbao111
|
监控 算法 开发工具
直播短视频系统开发，新系统开发需要哪些功能
直播短视频系统开发，新系统开发需要哪些功能
yunbao111
1030 0
云豹kj的晨曦
|
监控
盘点直播卖货系统开发要点功能
直播卖货系统开发作为行业内最脍炙人口的开发模式，凭借自身的优势在疫情期间稳定迅速的发展壮大起来，想要开发直播卖货系统，就要先了解需要具备哪些功能，本文就为大家简单盘点一下直播卖货系统的重要功能。
云豹kj的晨曦
438 0

热门文章

最新文章

  • 1
    Fiddler抓包：下载、安装及使用
  • 2
    SpringBoot开发案例之整合Dubbo提供者(一)
  • 3
    阿里AI上岗双11服装车间，能识别头发丝直径十分之一的瑕疵
  • 4
    RLE格式分割标注文件表示
  • 5
    微服务注册中心技术选型：5种主流注册中心，哪个最香？
  • 6
    云服务器VPS和独立服务器的区别是什么?
  • 7
    利用Windows Server Backup功能备份活动目录
  • 8
    从美国市场看全球移动技术演进:2015你5G了吗？
  • 9
    访谈李开复：谁主宰微软未来
  • 10
    RabbitMQ(从安装到使用)
  • 1
    Kotlin - 参数与异常
    18
  • 2
    Kotlin - 分支与循环
    13
  • 3
    Kotlin - 运算符与中缀表达式
    17
  • 4
    Kotlin - 类成员
    17
  • 5
    Kotlin - 分支与循环
    19
  • 6
    Kotlin - 常量与变量
    13
  • 7
    Kotlin - 运算符与中缀表达式
    14
  • 8
    Kotlin - 区间与数组
    12
  • 9
    Kotlin - 面向对象之抽象类与接口
    13
  • 10
    使用 Python 实现深度学习模型：智能食品生产线优化
    28

    • 相关课程

    更多
  • 开放语聊技术与应用
  • 大数据分析之企业级网站流量运营分析系统开发实战（第四阶段）
  • 场景实践 - 搭建个性化推荐引擎系统
  • 大数据分析之企业级网站流量运营分析系统开发实战（第二阶段）
  • 大数据分析之企业级网站流量运营分析系统开发实战（第三阶段）
  • 2020版大数据实战项目之DMP广告系统（第四阶段）

    • 相关电子书

    更多
  • 海量消息的直播互动系统演进历程
  • 快手在大数据驱动下的直播体验优化
  • 微信H5视频播放器在海量业务下的实践

    • 相关实验场景

    更多
  • 【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
  • SAE 极速部署专属 AI 证件照神器
  • 奥运时刻！零代码生成奥运风格AI写真。
  • 借助OSS搭建在线教育视频课程分享网站
  • 快速搭建移动端可视化看板
  • 利用大模型大规模分发技术，实现AIGC在线应用秒级弹性
    • 下一篇
    阿里云OSS设置跨域访问