• Java整合t-io实现在线聊天功能


    目录

    📕 介绍

     👍 特点

    👏 t-io性能与对比

    💌 使用介绍

    🎁 业务数据绑定

    🤷‍♂️ 业务数据解绑

    👀 异步发送

    🐱‍👤 阻塞发送

     🙌 获取ChannelContext

     🐱‍👓 断开连接和移除连接

    🚕Tio.remove()和Tio.close()的区别

    🏴 拉黑IP

    🚨 各种流量监控

    🛴 ip的监控数据

     👀 获取TCP会话的流量数据

    🐱‍🚀 监听端口的流量和数据

    ✨ T-io收发消息过程​编辑

    🤣 TCP连接上下文

    📕 TioConfig

     😀 消息来往监听

    🥰 整合JAVA

    🍔 websocket 配置类

    😄 消息处理类

    🚙 启动服务​编辑


    📕 介绍

    注:可以直接看整合java

    t-io是基于java开发的一个开源的网络编程架构,大家都知道现在手机上或者电脑上都装了很多APP,这些APP都不是一个个在手机上或电脑上孤立的使用,而是能访问其他的地方数据或者与其他节点进行实时聊天,故每个APP中都要有一个对外进行通信的模块,这块内容从编程的角度来看就能通过网络编程框架来实现,t-io就是完成这块的内容的最好的编程软件。

    t-io经过创始人谭先生的精心打磨,性能超强,已经达到运营的的级别,用t-io写的程序每秒能处理1000+万条消息,1.9G内存能够支撑30万TCP长连接。

    基于tio开发的即时通讯软件谭聊,目前是业界性能最强的全开源销售即时通讯软件。


     👍 特点

    t-io是基于JVM的网络编程框架,和netty属同类,所以netty能做的t-io都能做,考虑到t-io是从项目抽象出来的框架,所以t-io提供了更多的和业务相关的API,大体上t-io具有如下特点和能力

    • 内置完备的监控和流控能力
    • 内置半包粘包处理
    • 一骑绝尘的资源管理能力
    • 内置心跳检查和心跳发送能力
    • 内置IP拉黑
    • 一流性能和稳定性(第三方权威平台TFB提供性能测试和稳定性服务)
    • 极其稳定的表现(很多用户还是停在t-io 1.x版本,就是因为太过稳定,不想变动)
    • 内置慢攻击防御
    • 唯一一个内置异步发送、阻塞发送、同步发送的网络框架
    • 唯一内置集群分发消息的能力
    • 独创的多端口资源共享能力(譬如一个端口是websocket协议,一个端口是私有的im协议,这两个端口的资源可以共享,这对协议适配极其有用)
    • 独创协议适配转换能力(让基于websocket和基于socket的应用看起来像是同一个协议)
    • 独一档的资源和业务绑定能力:绑定group、绑定userid、绑定token、绑定bsId,这些绑定几乎囊括了所有业务需求

      👏 t-io性能与对比

      t-io测试结果:30W长连接并发压力,每秒1051万条聊天消息

      netty和t-io对比:

    • netty:100秒内生成10000个连接,每个连接每隔2秒发送一条消息,每个连接总共发送200条消息,发消息请求99%的响应时间在1ms以内,其它请求95%的响应时间在10ms以内
    • t-io:100秒内生成10000个连接,每个连接每隔2秒发送一条消息,每个连接总共发送200条消息,发消息请求99%的响应时间在1ms以内
    • CPU使用率:
    • netty:CPU使用率在20%左右
    • t-io:CPU使用率在15%左右

    💌 使用介绍

    🎁 业务数据绑定

    资源绑定是指把业务相关的数据和Tcp连接(即ChannelContext)关联起来,譬如ChannelContext-A代表了用户张三,张三的userid是333

    Tio.bindUser(ChannelContext-A, "333")  

    t-io目前内置了4种资源绑定,譬如给group加前缀"ios- ",从而标记这个用户使用的是ios

    1. Tio.bindGroup(ChannelContext-A, "333");
    2. Tio.bindGroup(ChannelContext-A, "ios-" + "333");

    内置的4种资源绑定方法中,一个ChannelContext是可以绑定到多个groupid的,其它三个绑定都是一对一或多对一的关系,也就是说一个ChannelContext可以同时属于group-a,group-b… …group-n

    1. /**
    2. * 绑定业务id
    3. * @param channelContext
    4. * @param bsId
    5. */
    6. public static void bindBsId(ChannelContext channelContext, String bsId) {
    7. channelContext.tioConfig.bsIds.bind(channelContext, bsId);
    8. }
    9. /**
    10. * 绑定群组
    11. * @param channelContext
    12. * @param group
    13. */
    14. public static void bindGroup(ChannelContext channelContext, String group) {
    15. channelContext.tioConfig.groups.bind(group, channelContext);
    16. }
    17. /**
    18. * 绑定token
    19. * @param channelContext
    20. * @param token
    21. */
    22. public static void bindToken(ChannelContext channelContext, String token) {
    23. channelContext.tioConfig.tokens.bind(token, channelContext);
    24. }
    25. /**
    26. * 绑定用户
    27. * @param channelContext
    28. * @param userid
    29. */
    30. public static void bindUser(ChannelContext channelContext, String userid) {
    31. channelContext.tioConfig.users.bind(userid, channelContext);
    32. }

    🤷‍♂️ 业务数据解绑

    既然有绑定,就肯定会有解绑,这是个和绑定相反的操作

    1. /**
    2. * 解绑业务id
    3. * @param channelContext
    4. */
    5. public static void unbindBsId(ChannelContext channelContext) {
    6. channelContext.tioConfig.bsIds.unbind(channelContext);
    7. }
    8. /**
    9. * 与所有组解除解绑关系
    10. * @param channelContext
    11. */
    12. public static void unbindGroup(ChannelContext channelContext) {
    13. channelContext.tioConfig.groups.unbind(channelContext);
    14. }
    15. /**
    16. * 与指定组解除绑定关系
    17. * @param group
    18. * @param channelContext
    19. */
    20. public static void unbindGroup(String group, ChannelContext channelContext) {
    21. channelContext.tioConfig.groups.unbind(group, channelContext);
    22. }
    23. /**
    24. * 解除channelContext绑定的token
    25. * @param channelContext
    26. */
    27. public static void unbindToken(ChannelContext channelContext) {
    28. channelContext.tioConfig.tokens.unbind(channelContext);
    29. }
    30. // org.tio.core.TioConfig.ipBlacklist
    31. /**
    32. * 解除channelContext绑定的userid
    33. * @param channelContext
    34. */
    35. public static void unbindUser(ChannelContext channelContext) {
    36. channelContext.tioConfig.users.unbind(channelContext);
    37. }
    38. /**
    39. * 解除userid的绑定。一般用于多地登录,踢掉前面登录的场景
    40. * @param tioConfig
    41. * @param userid
    42. */
    43. public static void unbindUser(TioConfig tioConfig, String userid) {
    44. tioConfig.users.unbind(tioConfig, userid);
    45. }

    👀 异步发送

    • 异步发送,指的是业务层把Packet丢给t-io后立即返回,返回时Packet并没有被发送,而只是提交到了待发送队列
    • 异步发送都是以send开头的


    🐱‍👤 阻塞发送

    • 阻塞发送:t-io把Packet送给对方后才返回
    • 阻塞发送都是以bSend开头的

     


     🙌 获取ChannelContext

    • 前面的业务数据绑定,一个重要的目的就是要根据那些业务标识来获取ChannelContext,譬如你绑定了一个userid,那么后面就可以通过这个userid来获取ChannelContext
    • 获取ChannelContext的API都是以get开头的

     


     🐱‍👓 断开连接和移除连接

    断开连接都是以close开头的方法,指的是把当前已经连上的TCP连接断开掉

    移除连接都是以remove开头的方法,指的是彻底抛弃这个连接

    🚕Tio.remove()和Tio.close()的区别

    Tio.remove:不管是用t-io做TCP服务器还是TCP客户端,调用Tio.remove()后,t-io都会彻底删除TCP连接并释放包括ChannelContext在内的所有和该条TCP连接对应的资源,当然那些和群组、Token的绑定关系也全部释放掉

    Tio.close:

    • 如果是用t-io做TCP服务器,此方法等价于Tio.remove();
    • 如果是用t-io做TCP客户端
      • 该方法会断开当前TCP连接
      • 如果业务程序配置了重连策略(就是:ReconnConf):
        • t-io后面会进行重连操作,也就是说并不会抛弃该条TCP连接对应的ChannelContext对象
        • 如果该条TCP连接对应的ChannelContext对象绑定了groupid、userid、token、bsId,那么这些绑定关系会全部释放掉,在重连成功后,业务侧需要再次进行绑定
      • 如果业务程序没有配置重连策略(就是:ReconnConf),此方法等价于Tio.remove()

    出现网络异常或其它异常时,业务需要主动调用这俩方法吗?

    答:不需要的,出现任何网络异常,t-io都会释放掉该条TCP连接对应的全部资源,这也是t-io如此稳定的一大原因。网络编程的很多坑,都是源于资源没释放


    🏴 拉黑IP

    简单到极致,只需要一行代码

    Tio.IpBlacklist.add(tioConfig, channelContext.getClientNode().getIp());

    🚨 各种流量监控

    🛴 ip的监控数据

    ip的监控数据定义在IpStat中

    1. private Date start = new Date();
    2. /**
    3. * 当前统计了多久,单位:毫秒
    4. */
    5. private long duration;
    6. /**
    7. * 时长类型,单位:秒,譬如60,3600等
    8. */
    9. private Long durationType;
    10. /**
    11. * 客户端ip
    12. */
    13. private String ip;
    14. /**
    15. * 解码异常的次数
    16. */
    17. private AtomicInteger decodeErrorCount = new AtomicInteger();
    18. /**
    19. * 收到该IP连接请求的次数
    20. */
    21. private AtomicInteger requestCount = new AtomicInteger();
    22. /**
    23. * 本IP已发送的字节数
    24. */
    25. private AtomicLong sentBytes = new AtomicLong();
    26. /**
    27. * 本IP已发送的packet数
    28. */
    29. private AtomicLong sentPackets = new AtomicLong();
    30. /**
    31. * 本IP已处理的字节数
    32. */
    33. private AtomicLong handledBytes = new AtomicLong();
    34. /**
    35. * 本IP已处理的packet数
    36. */
    37. private AtomicLong handledPackets = new AtomicLong();
    38. /**
    39. * 处理消息包耗时,单位:毫秒
    40. */
    41. private AtomicLong handledPacketCosts = new AtomicLong();
    42. /**
    43. * 本IP已接收的字节数
    44. */
    45. private AtomicLong receivedBytes = new AtomicLong();
    46. /**
    47. * 本IP已接收了多少次TCP数据包
    48. */
    49. private AtomicLong receivedTcps = new AtomicLong();
    50. /**
    51. * 本IP已接收的packet数
    52. */
    53. private AtomicLong receivedPackets = new AtomicLong();

    使用步骤

    • 实现IpStatListener
    1. package org.tio.showcase.websocket.server;
    2. public class ShowcaseIpStatListener implements IpStatListener {
    3. @SuppressWarnings("unused")
    4. private static Logger log = LoggerFactory.getLogger(ShowcaseIpStatListener.class);
    5. public static final ShowcaseIpStatListener me = new ShowcaseIpStatListener();
    6. private ShowcaseIpStatListener() {
    7. }
    8. @Override
    9. public void onExpired(TioConfig tioConfig, IpStat ipStat) {
    10. //在这里把统计数据入库中或日志
    11. // if (log.isInfoEnabled()) {
    12. // log.info("可以把统计数据入库\r\n{}", Json.toFormatedJson(ipStat));
    13. // }
    14. }
    15. @Override
    16. public void onAfterConnected(ChannelContext channelContext, boolean isConnected, boolean isReconnect, IpStat ipStat) throws Exception {
    17. // if (log.isInfoEnabled()) {
    18. // log.info("onAfterConnected\r\n{}", Json.toFormatedJson(ipStat));
    19. // }
    20. }
    21. @Override
    22. public void onDecodeError(ChannelContext channelContext, IpStat ipStat) {
    23. // if (log.isInfoEnabled()) {
    24. // log.info("onDecodeError\r\n{}", Json.toFormatedJson(ipStat));
    25. // }
    26. }
    27. @Override
    28. public void onAfterSent(ChannelContext channelContext, Packet packet, boolean isSentSuccess, IpStat ipStat) throws Exception {
    29. // if (log.isInfoEnabled()) {
    30. // log.info("onAfterSent\r\n{}\r\n{}", packet.logstr(), Json.toFormatedJson(ipStat));
    31. // }
    32. }
    33. @Override
    34. public void onAfterDecoded(ChannelContext channelContext, Packet packet, int packetSize, IpStat ipStat) throws Exception {
    35. // if (log.isInfoEnabled()) {
    36. // log.info("onAfterDecoded\r\n{}\r\n{}", packet.logstr(), Json.toFormatedJson(ipStat));
    37. // }
    38. }
    39. @Override
    40. public void onAfterReceivedBytes(ChannelContext channelContext, int receivedBytes, IpStat ipStat) throws Exception {
    41. // if (log.isInfoEnabled()) {
    42. // log.info("onAfterReceivedBytes\r\n{}", Json.toFormatedJson(ipStat));
    43. // }
    44. }
    45. @Override
    46. public void onAfterHandled(ChannelContext channelContext, Packet packet, IpStat ipStat, long cost) throws Exception {
    47. // if (log.isInfoEnabled()) {
    48. // log.info("onAfterHandled\r\n{}\r\n{}", packet.logstr(), Json.toFormatedJson(ipStat));
    49. // }
    50. }
    51. }
    • 初始化时添加监听器和监控时段
    1. //注意的是:要保证下面两行代码的顺序,不能先addDuration()后setIpStatListener
    2. serverTioConfig.setIpStatListener(ShowcaseIpStatListener.me);
    3. serverTioConfig.ipStats.addDuration(Time.MINUTE_1 * 5);
    • OK了,什么时候拉黑IP以及把监控数据入库都在ShowcaseIpStatListener中实现哦

     👀 获取TCP会话的流量数据

    一个TCP会话对应一个ChannelContext对象,每个ChannelContext对象都有一个ChannelStat对象,定义如下

    public final ChannelStat stat = new ChannelStat();

    ChannelStat包含如下字段和方法(已经略过普通的getter和setter)

    1. /**
    2. * 本次解码失败的次数
    3. */
    4. public int decodeFailCount = 0;
    5. /**
    6. * 最近一次收到业务消息包的时间(一个完整的业务消息包,一部分消息不算)
    7. */
    8. public long latestTimeOfReceivedPacket = SystemTimer.currTime;
    9. /**
    10. * 最近一次发送业务消息包的时间(一个完整的业务消息包,一部分消息不算)
    11. */
    12. public long latestTimeOfSentPacket = SystemTimer.currTime;
    13. /**
    14. * 最近一次收到业务消息包的时间:收到字节就算
    15. */
    16. public long latestTimeOfReceivedByte = SystemTimer.currTime;
    17. /**
    18. * 最近一次发送业务消息包的时间:发送字节就算
    19. */
    20. public long latestTimeOfSentByte = SystemTimer.currTime;
    21. /**
    22. * ChannelContext对象创建的时间
    23. */
    24. public long timeCreated = SystemTimer.currTime;
    25. /**
    26. * 第一次连接成功的时间
    27. */
    28. public Long timeFirstConnected = null;
    29. /**
    30. * 连接关闭的时间
    31. */
    32. public long timeClosed = SystemTimer.currTime;
    33. /**
    34. * 进入重连队列时间
    35. */
    36. public long timeInReconnQueue = SystemTimer.currTime;
    37. /**
    38. * 本连接已发送的字节数
    39. */
    40. public final AtomicLong sentBytes = new AtomicLong();
    41. /**
    42. * 本连接已发送的packet数
    43. */
    44. public final AtomicLong sentPackets = new AtomicLong();
    45. /**
    46. * 本连接已处理的字节数
    47. */
    48. public final AtomicLong handledBytes = new AtomicLong();
    49. /**
    50. * 本连接已处理的packet数
    51. */
    52. public final AtomicLong handledPackets = new AtomicLong();
    53. /**
    54. * 处理消息包耗时,单位:毫秒
    55. * 拿这个值除以handledPackets,就是处理每个消息包的平均耗时
    56. */
    57. public final AtomicLong handledPacketCosts = new AtomicLong();
    58. /**
    59. * 本连接已接收的字节数
    60. */
    61. public final AtomicLong receivedBytes = new AtomicLong();
    62. /**
    63. * 本连接已接收了多少次TCP数据包
    64. */
    65. public final AtomicLong receivedTcps = new AtomicLong();
    66. /**
    67. * 本连接已接收的packet数
    68. */
    69. public final AtomicLong receivedPackets = new AtomicLong();
    70. /**
    71. * 平均每次TCP接收到的字节数,这个可以用来监控慢攻击,配置PacketsPerTcpReceive定位慢攻击
    72. */
    73. public double getBytesPerTcpReceive() {
    74. if (receivedTcps.get() == 0) {
    75. return 0;
    76. }
    77. double ret = (double) receivedBytes.get() / (double) receivedTcps.get();
    78. return ret;
    79. }
    80. /**
    81. * 平均每次TCP接收到的业务包数,这个可以用来监控慢攻击,此值越小越有攻击嫌疑
    82. */
    83. public double getPacketsPerTcpReceive() {
    84. if (receivedTcps.get() == 0) {
    85. return 0;
    86. }
    87. double ret = (double) receivedPackets.get() / (double) receivedTcps.get();
    88. return ret;
    89. }
    90. /**
    91. * 处理packet平均耗时,单位:毫秒
    92. * @return
    93. */
    94. public double getHandledCostsPerPacket() {
    95. if (handledPackets.get() > 0) {
    96. return handledPacketCosts.get() / handledPackets.get();
    97. }
    98. return 0;
    99. }

    🐱‍🚀 监听端口的流量和数据

    TioConfig对象有个GroupStat成员,定义如下

    public GroupStat groupStat = null;

    GroupStat有如下一些字段和方法(去掉了简单的getter和setter)

    1. /**
    2. * 关闭了多少连接
    3. */
    4. public final AtomicLong closed = new AtomicLong();
    5. /**
    6. * 接收到的消息包
    7. */
    8. public final AtomicLong receivedPackets = new AtomicLong();
    9. /**
    10. * 接收到的消息字节数
    11. */
    12. public final AtomicLong receivedBytes = new AtomicLong();
    13. /**
    14. * 处理了的消息包数
    15. */
    16. public final AtomicLong handledPackets = new AtomicLong();
    17. /**
    18. * 处理消息包耗时,单位:毫秒
    19. */
    20. public final AtomicLong handledPacketCosts = new AtomicLong();
    21. /**
    22. * 处理了多少字节
    23. */
    24. public final AtomicLong handledBytes = new AtomicLong();
    25. /**
    26. * 发送了的消息包数
    27. */
    28. public final AtomicLong sentPackets = new AtomicLong();
    29. /**
    30. * 发送了的字节数
    31. */
    32. public final AtomicLong sentBytes = new AtomicLong();
    33. /**
    34. * 本IP已接收了多少次TCP数据包
    35. */
    36. public final AtomicLong receivedTcps = new AtomicLong();
    37. /**
    38. * 平均每次TCP接收到的字节数,这个可以用来监控慢攻击,配置PacketsPerTcpReceive定位慢攻击
    39. */
    40. public double getBytesPerTcpReceive() {
    41. if (receivedTcps.get() == 0) {
    42. return 0;
    43. }
    44. double ret = (double) receivedBytes.get() / (double) receivedTcps.get();
    45. return ret;
    46. }
    47. /**
    48. * 平均每次TCP接收到的业务包数,这个可以用来监控慢攻击,此值越小越有攻击嫌疑
    49. */
    50. public double getPacketsPerTcpReceive() {
    51. if (receivedTcps.get() == 0) {
    52. return 0;
    53. }
    54. double ret = (double) receivedPackets.get() / (double) receivedTcps.get();
    55. return ret;
    56. }
    57. /**
    58. * 处理packet平均耗时,单位:毫秒
    59. * @return
    60. */
    61. public double getHandledCostsPerPacket() {
    62. if (handledPackets.get() > 0) {
    63. return handledPacketCosts.get() / handledPackets.get();
    64. }
    65. return 0;
    66. }

    对于服务器端的groupStat,它是在ServerTioConfig类中的初始化代码在构造函数中,如下

    this.groupStat = new ServerGroupStat();

    对于客户端的groupStat,它是在ClientTioConfig类中的初始化代码在构造函数中,如下

    this.groupStat = new ClientGroupStat();

    获取GroupStat

    1. GroupStat groupStat = tioConfig.groupStat;
    2. //如果确认是服务器端,则可以用强转方式获得ServerGroupStat对象
    3. ServerGroupStat serverGroupStat = (ServerGroupStat)tioConfig.groupStat;
    4. //如果确认是客户端,则可以用强转方式获得ClientGroupStat对象
    5. ClientGroupStat clientGroupStat = (ClientGroupStat)tioConfig.groupStat;

    ✨ T-io收发消息过程

    Packet是用于表述业务数据结构的,我们通过继承Packet来实现自己的业务数据结构,对于各位而言,把Packet看作是一个普通的VO对象即可。

    注意:不建议直接使用Packet对象,而是要继承Packet


    🤣 TCP连接上下文

    每一个tcp连接的建立都会产生一个ChannelContext对象,这是个抽象类,如果你是用t-io作tcp客户端,那么就是ClientChannelContext,如果你是用tio作tcp服务器,那么就是ServerChannelContext

    用户可以把业务数据通过ChannelContext对象和TCP连接关联起来,像下面这样设置属性

    ChannelContext.set(String key, Object value)

    然后用下面的方式获取属性

    ChannelContext.get(String key)

    当然最最常用的还是用t-io提供的强到没对手的bind功能,譬如用下面的代码绑定userid

    Tio.bindUser(ChannelContext channelContext, String userid)

    然后可以通过userid进行操作,示范代码如下

    1. //获取某用户的ChannelContext集合
    2. SetWithLock set = Tio.getChannelContextsByUserid(tioConfig, userid);
    3. //给某用户发消息
    4. Tio.sendToUser(TioConfig, userid, Packet)

    除了可以绑定userid,t-io还内置了如下绑定API

    • 绑定业务id
    Tio.bindBsId(ChannelContext channelContext, String bsId)
    • 绑定token
    Tio.bindToken(ChannelContext channelContext, String token)
    • 绑定群组
    Tio.bindGroup(ChannelContext channelContext, String group)

    ChannelContext对象包含的信息非常多,主要对象见下图

     ChannelContext是t-io中非常重要的类,他是业务和连接的沟通桥梁!


    📕 TioConfig

    • 场景:我们在写TCP Server时,都会先选好一个端口以监听客户端连接,再创建N组线程池来执行相关的任务,譬如发送消息、解码数据包、处理数据包等任务,还要维护客户端连接的各种数据,为了和业务互动,还要把这些客户端连接和各种业务数据绑定起来,譬如把某个客户端绑定到一个群组,绑定到一个userid,绑定到一个token等。
    • TioConfig

    就是解决以上场景的:配置线程池、监听端口,维护客户端各种数据等的。

    • TioConfig是个抽象类
      • 如果你是用tio作tcp客户端,那么你需要创建ClientTioConfig对象
        • 服务器端对应一个ClientTioConfig对象
      • 如果你是用tio作tcp服务器,那么你需要创建ServerTioConfig
        • 一个监听端口对应一个ServerTioConfig ,一个jvm可以监听多个端口,所以一个jvm可以有多个ServerTioConfig对象

     😀 消息来往监听

    TioListener是处理消息的核心接口,它有两个子接口:TioClientListener和TioServerListener

    • 当用tio作tcp客户端时需要实现TioClientListener
    • 当用tio作tcp服务器时需要实现TioServerListener

    它主要定义了如下方法

    1. public interface TioListener {
    2. /**
    3. * 建链后触发本方法,注:建链不一定成功,需要关注参数isConnected
    4. *
    5. * @param channelContext
    6. * @param isConnected 是否连接成功,true:表示连接成功,false:表示连接失败
    7. * @param isReconnect 是否是重连, true: 表示这是重新连接,false: 表示这是第一次连接
    8. * @throws Exception
    9. */
    10. public void onAfterConnected(ChannelContext channelContext, boolean isConnected, boolean isReconnect) throws Exception;
    11. /**
    12. * 原方法名:onAfterDecoded 解码成功后触发本方法
    13. *
    14. * @param channelContext
    15. * @param packet
    16. * @param packetSize
    17. * @throws Exception
    18. */
    19. public void onAfterDecoded(ChannelContext channelContext, Packet packet, int packetSize) throws Exception;
    20. /**
    21. * 处理一个消息包后
    22. *
    23. * @param channelContext
    24. * @param packet
    25. * @param cost 本次处理消息耗时,单位:毫秒
    26. * @throws Exception
    27. */
    28. public void onAfterHandled(ChannelContext channelContext, Packet packet, long cost) throws Exception;
    29. /**
    30. * 接收到TCP层传过来的数据后
    31. *
    32. * @param channelContext
    33. * @param receivedBytes 本次接收了多少字节
    34. * @throws Exception
    35. */
    36. public void onAfterReceivedBytes(ChannelContext channelContext, int receivedBytes) throws Exception;
    37. /**
    38. * 消息包发送之后触发本方法
    39. *
    40. * @param channelContext
    41. * @param packet
    42. * @param isSentSuccess true:发送成功,false:发送失败
    43. * @throws Exception */
    44. public void onAfterSent(ChannelContext channelContext, Packet packet, boolean isSentSuccess) throws Exception;
    45. /**
    46. * 连接关闭前触发本方法
    47. *
    48. * @param channelContext the channelcontext
    49. * @param throwable the throwable 有可能为空
    50. * @param remark the remark 有可能为空
    51. * @param isRemove
    52. * @throws Exception
    53. */
    54. public void onBeforeClose(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) throws Exception;
    55. /**
    56. * 连接关闭前后触发本方法 警告:走到这个里面时,很多绑定的业务都已经解绑了,所以这个方法一般是空着不实现的
    57. *
    58. * @param channelContext the channelcontext
    59. * @param throwable the throwable 有可能为空
    60. * @param remark the remark 有可能为空
    61. * @param isRemove 是否是删除
    62. * @throws Exception
    63. */
    64. // public void onAfterClose(ChannelContext channelContext, Throwable throwable,
    65. // String remark, boolean isRemove) throws Exception;
    66. }

    🥰 整合JAVA

    1. <dependency>
    2. <groupId>org.t-iogroupId>
    3. <artifactId>tio-websocket-serverartifactId>
    4. <version>3.5.9.v20200214-RELEASEversion>
    5. dependency>

    注意:每个版本之前存在差异请查看官方文档

    🍔 websocket 配置类

    1. import com.asurplus.tio.websocket.handle.MyWsMsgHandler;
    2. import org.springframework.beans.factory.annotation.Autowired;
    3. import org.springframework.context.annotation.Bean;
    4. import org.springframework.context.annotation.Configuration;
    5. import org.tio.server.ServerTioConfig;
    6. import org.tio.websocket.server.WsServerStarter;
    7. import java.io.IOException;
    8. /**
    9. * websocket 配置类
    10. */
    11. @Configuration
    12. public class WebSocketConfig {
    13. /**
    14. * 注入消息处理器
    15. */
    16. @Autowired
    17. private MyWsMsgHandler myWsMsgHandler;
    18. /**
    19. * 启动类配置
    20. *
    21. * @return
    22. * @throws IOException
    23. */
    24. @Bean
    25. public WsServerStarter wsServerStarter() throws IOException {
    26. // 设置处理器
    27. WsServerStarter wsServerStarter = new WsServerStarter(6789, myWsMsgHandler);
    28. // 获取到ServerTioConfig
    29. ServerTioConfig serverTioConfig = wsServerStarter.getServerTioConfig();
    30. // 设置心跳超时时间,默认:1000 * 120
    31. serverTioConfig.setHeartbeatTimeout(1000 * 120);
    32. // 启动
    33. wsServerStarter.start();
    34. return wsServerStarter;
    35. }
    36. }

    这里我们注入了 WsServerStarter 的 bean,这样在 SpringBoot 启动的时候,就能启动咱们的 websocket 服务

    • 注明了 websocket 的服务端口为:6789
    • 消息处理类为:myWsMsgHandler,在下一步我们将会去实现这个类
    • 设置了心跳的超时时间为:120秒,默认值,可以不设置

    😄 消息处理类

    1. package com.ying.tiiochat.config;
    2. import com.alibaba.fastjson.JSONObject;
    3. import org.springframework.stereotype.Component;
    4. import org.tio.core.ChannelContext;
    5. import org.tio.core.Tio;
    6. import org.tio.http.common.HttpRequest;
    7. import org.tio.http.common.HttpResponse;
    8. import org.tio.websocket.common.WsRequest;
    9. import org.tio.websocket.common.WsResponse;
    10. import org.tio.websocket.server.handler.IWsMsgHandler;
    11. /**
    12. * 消息处理类
    13. */
    14. @Component
    15. public class MyWsMsgHandler implements IWsMsgHandler {
    16. /**
    17. *
    18. 对httpResponse参数进行补充并返回,如果返回null表示不想和对方建立连接,框架会断开连接,如果返回非null,框架会把这个对象发送给对方
  • *
  • 注:请不要在这个方法中向对方发送任何消息,因为这个时候握手还没完成,发消息会导致协议交互失败。
  • *
  • 对于大部分业务,该方法只需要一行代码:return httpResponse;
  • *
  • * @param httpRequest
  • * @param httpResponse
  • * @param channelContext
  • * @return
  • * @throws Exception
  • */
  • @Override
  • public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
  • // 可以在此做一些业务逻辑,返回null表示不想连接
  • return httpResponse;
  • }
  • /**
  • * 握手成功后触发该方法
  • *
  • * @param httpRequest
  • * @param httpResponse
  • * @param channelContext
  • * @throws Exception
  • */
  • @Override
  • public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
  • // 拿到用户id
  • String id = httpRequest.getParam("userId");
  • // 绑定用户
  • Tio.bindUser(channelContext, id);
  • // 给用户发送消息
  • JSONObject message = new JSONObject();
  • message.put("msg", "连接成功...");
  • message.put("sendName", "系统提醒");
  • WsResponse wsResponse = WsResponse.fromText(message.toString(), "UTF-8");
  • Tio.sendToUser(channelContext.tioConfig, id, wsResponse);
  • }
  • /**
  • *
  • 当收到Opcode.BINARY消息时,执行该方法。也就是说如何你的ws是基于BINARY传输的,就会走到这个方法
  • *
  • * @param wsRequest
  • * @param bytes
  • * @param channelContext
  • * @return 可以是WsResponse、byte[]、ByteBuffer、String或null,如果是null,框架不会回消息
  • * @throws Exception
  • */
  • @Override
  • public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
  • System.out.println("我走了onBytes");
  • return null;
  • }
  • /**
  • * 当收到Opcode.CLOSE时,执行该方法,业务层在该方法中一般不需要写什么逻辑,空着就好
  • *
  • * @param wsRequest
  • * @param bytes
  • * @param channelContext
  • * @return 可以是WsResponse、byte[]、ByteBuffer、String或null,如果是null,框架不会回消息
  • * @throws Exception
  • */
  • @Override
  • public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
  • // 关闭连接
  • Tio.remove(channelContext, "WebSocket Close");
  • return null;
  • }
  • /**
  • *
  • 当收到Opcode.TEXT消息时,执行该方法。也就是说如何你的ws是基于TEXT传输的,就会走到这个方法
  • *
  • * @param wsRequest
  • * @param text
  • * @param channelContext
  • * @return 可以是WsResponse、byte[]、ByteBuffer、String或null,如果是null,框架不会回消息
  • * @throws Exception
  • */
  • @Override
  • public Object onText(WsRequest wsRequest, String text, ChannelContext channelContext) throws Exception {
  • JSONObject message = JSONObject.parseObject(text);
  • // 接收消息的用户ID
  • String receiver = message.getString("receiver");
  • // 发送消息者
  • String sendName = message.getString("sendName");
  • // 消息
  • String msg = message.getString("msg");
  • // 保存聊天记录到DB等业务逻辑...
  • WsResponse wsResponse = WsResponse.fromText(message.toString(), "UTF-8");
  • Tio.sendToUser(channelContext.tioConfig, receiver, wsResponse);
  • JSONObject resp = new JSONObject();
  • resp.put("sendName", "系统提醒");
  • resp.put("msg", "发送成功");
  • return resp.toString();
  • }
  • }
  • 我们实现了 IWsMsgHandler 接口,并重写了该接口的 5 个方法,这 5 个方法从 发送握手包,到消息收发,到断开连接等一系列过程

    🚙 启动服务

     启动成功后,可以看出 tio 的打印结果,我们可以看出服务端口为我们设置的 6789,我们便可以连接测试了

    前端代码(用脚写有点丑):

    1. html>
    2. <html>
    3. <head>
    4. <meta charset="utf-8">
    5. <title>websocket通讯title>
    6. head>
    7. <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js">script>
    8. <script type="text/javascript">
    9. var socket;
    10. var userName;
    11. // 连接
    12. function connect() {
    13. var socketUrl = "ws://localhost:6789/?userId=" + $('#sendName').val();
    14. if (socket != null) {
    15. socket.close();
    16. socket = null;
    17. }
    18. socket = new WebSocket(socketUrl);
    19. //打开事件
    20. socket.onopen = function () {
    21. console.log("开始建立链接....")
    22. };
    23. //关闭事件
    24. socket.onclose = function () {
    25. console.log("websocket已关闭");
    26. };
    27. //发生了错误事件
    28. socket.onerror = function () {
    29. console.log("websocket发生了错误");
    30. };
    31. /**
    32. * 接收消息
    33. * @param msg
    34. */
    35. socket.onmessage = function (msg) {
    36. console.log(msg)
    37. var json = JSON.parse(msg.data);
    38. if (msg.msg != '连接成功') {
    39. $("#msgDiv").append('

      ' + json.sendName + ':'+json.msg+'

      '
      );
    40. }
    41. };
    42. }
    43. /**
    44. * 发送消息
    45. */
    46. function sendMessage() {
    47. var msg = $("#msg").val();
    48. if (msg == '' || msg == null) {
    49. alert("消息内容不能为空");
    50. return;
    51. }
    52. var receiver = $("#receiver").val();
    53. if (receiver == '' || receiver == null) {
    54. alert("接收人不能为空");
    55. return;
    56. }
    57. var sendName = $("#sendName").val();
    58. if (sendName == '' || sendName == null) {
    59. alert("发送人不能为空");
    60. return;
    61. }
    62. var msgObj = {
    63. "receiver": receiver,
    64. "sendName": sendName,
    65. "msg": msg
    66. };
    67. $("#msgDiv").append('

      ' + sendName + ':'+msg+'

      '
      );
    68. try{
    69. socket.send(JSON.stringify(msgObj));
    70. $("#msg").val('');
    71. }catch (e) {
    72. alert("服务器内部错误");
    73. }
    74. }
    75. script>
    76. <body>
    77. 用户名:<input type="text" id="sendName" value="李四">
    78. <input type="button" value="连接" onclick="connect()" ><br>
    79. 发送者:<input type="text" id="sender" value="李四" ><br>
    80. 接受者:<input type="text" id="receiver" value="张三"><br><br>
    81. 消 息:<textarea id="msg">textarea><br><input type="button" value="发送" onclick="sendMessage()"><br><br>
    82. 消息记录:<div id="msgDiv" style="border: 1px red solid;width: 400px;height: 200px">div>
    83. <br>
    84. body>
    85. html>

      这是小编在开发学习使用和总结,  这中间或许也存在着不足,希望可以得到大家的理解和建议。如有侵权联系小编!

  • 相关阅读:
    深度学习【使用seq2seq实现聊天机器人】
    四数之和 - 力扣中等
    Windows使用小技巧
    【软考】14.1 面向对象基本概念/分析设计测试
    [Codeforces] number theory (R1200) Part.9
    中电金信技术实践|Redis哨兵原理及安装部署分享
    Python内置函数--exec()
    java毕业生设计移动在线点菜系统服务端服务端计算机源码+系统+mysql+调试部署+lw
    C#的播放资源文件里的音频例子 - 开源研究系列文章
    spring事物失效场景及其解决方案
  • 原文地址:https://blog.csdn.net/weixin_46522803/article/details/126548065