投稿
  1. 首页
  2. java

t-io实战训练

code • 2022-4-26 • java • 阅读 109

t-io实战训练,第1张

  • 上一讲我们了解了请求分发的一点小思路,这一讲进行实战训练,项目搭建过程忽略,如果需要源码,去公众号回复"tio实战源码",后续会分享一些个人在项目方面的思考,在不同的业务下,项目的构建思路,闲言少叙

  • 这个项目是因为有个朋友需要搞个socket项目,然后让我帮忙搞一下,本着偷懒的原则然后试着引入了t-io,相关资料自行百度(https://www.tiocloud.com/2/product/tio.html, https://www.oschina.net/p/t-io),t-io作者说如果网络编程很痛苦，那一定是没用 t-io,经过使用,确实比netty要感觉简单很多

  • 首先看下项目结构


tio-demo-common 模块放一些公共的工具类以及bean
tio-demo-socket socket放socket相关的核心类
tio-demo-socket-biz biz放业务类

  • 结构比较简单

-

  • 消息包:
/**
* socket消息包
*
* @author
*/
public class TioDemoPacket extends Packet {
   /**
    * 消息头的长度
    */
   public static final int HEADER_LENGTH = 4;
   private byte[] body;
   public byte[] getBody() {
       return body;
   }
   public void setBody(byte[] body) {
       this.body = body;
   }
}
  • 接收参数解码:
    /**
    * 解码：把接收到的ByteBuffer，解码成应用可以识别的业务消息包
    * 总的消息结构：消息头 + 消息体
    * 消息头结构：    4个字节，存储消息体的长度
    * 消息体结构：   对象的json串的byte[]
    * 根据ByteBuffer解码成业务需要的Packet对象.
    * 如果收到的数据不全，导致解码失败，请返回null，在下次消息来时框架层会自动续上前面的收到的数据
    *
    * @param buffer         参与本次希望解码的ByteBuffer
    * @param limit          ByteBuffer的limit
    * @param position       ByteBuffer的position，不一定是0哦
    * @param readableLength ByteBuffer参与本次解码的有效数据（= limit - position）
    * @param channelContext
    * @return
    * @throws AioDecodeException
    */

   @Override
   public TioDemoPacket decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext) throws AioDecodeException {
       if (buffer.remaining() < TioDemoPacket.HEADER_LENGTH) {
           return null;
       } else {
           TioDemoPacket imPacket = new TioDemoPacket();
           if (buffer.remaining() < TioDemoPacket.HEADER_LENGTH) {
               return null;
           } else {
               byte[] bytes = new byte[limit];
               buffer.get(bytes);
               imPacket.setBody(bytes);
               return imPacket;
           }
       }
   }
  • 接收参数之后的解析
  TioDemoPacket packingPacket = (TioDemoPacket) packet;
      byte[] body = packingPacket.getBody();
      if (body != null) {
          String reqJson = new String(body);
          if (logPrint) {
              log.info("接收到服务器的请求信息:{},{},收到消息：{}", cc.getId(), cc.getToken(), reqJson);
          }
          if (!JSONUtil.isTypeJSON(reqJson)) {
              log.error("参数不是json");
              return;
          }
          JSONObject jb = JSONUtil.parseObj(reqJson);
          String service = jb.getStr("service");
          if (StrUtil.isEmpty(service)) {
              log.error("接口参数缺失");
              return;
          }
          if (!InterfaceNameEnum.interfaceNameIsExist(service)) {
              log.error("接口未接入");
              return;
          }
          RequestMessageHandler requestMessageHandler = requestMessageHandlerContainer.getMessageHandler(service);
          // 获得  MessageHandler 处理器 的消息类
          Class<? extends RequestMessage> messageClass = RequestMessageHandlerContainer.getMessageClass(requestMessageHandler);
          // 解析消息
          RequestMessage requestMessage = JacksonUtils.toBean(reqJson, messageClass);
          //线程池执行
          tioDemoTaskExecutor.submit(() -> {
              //先判断是否是认证接口
              if (InterfaceNameEnum.CHECK_KEY.getName().equals(service)) {
                  requestMessageHandler.execute(cc, requestMessage);
                  return;
              }
              String token = cc.getToken();
              if (StrUtil.isEmpty(token)) {
                  log.error("调用的接口未经认证,不能调用:{}", service);
                  Tio.close(cc, token);
                  return;
              }
              requestMessageHandler.execute(cc, requestMessage);

          });
          return;
      }
  • 响应参数编码:

    /**
     * 编码：把业务消息包编码为可以发送的ByteBuffer
     * 总的消息结构：消息头 + 消息体
     * 消息头结构：    4个字节，存储消息体的长度
     * 消息体结构：   对象的json串的byte[]
     */
    @Override
    public ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext channelContext) {
        TioDemoPacket serverPacket = (TioDemoPacket) packet;
        byte[] body = serverPacket.getBody();
        int bodyLen = 0;
        if (body != null) {
            bodyLen = body.length;
        }
        //bytebuffer的总长度是 = 消息头的长度 + 消息体的长度
        int allLen = bodyLen;
        //创建一个新的bytebuffer
        ByteBuffer buffer = ByteBuffer.allocate(allLen);
        //设置字节序
        buffer.order(tioConfig.getByteOrder());
        //写入消息头----消息头的内容就是消息体的长度
        //写入消息体
        if (body != null) {
            buffer.put(body);
        }
        return buffer;
    }
  • listener 在监听到关闭以后清理token
 /**
     * 连接关闭前触发本方法
     *
     * @param cc        the channelcontext
     * @param throwable the throwable 有可能为空
     * @param remark    the remark 有可能为空
     * @param isRemove
     * @throws Exception
     */
    @Override
    public void onBeforeClose(ChannelContext cc, Throwable throwable, String remark, boolean isRemove) throws Exception {
        log.info("server onBeforeClose");
        log.info("关闭后清除认证信息");
        Tio.unbindToken(cc);
    }

    /**
     * @param cc
     * @param interval              已经多久没有收发消息了，单位：毫秒
     * @param heartbeatTimeoutCount 心跳超时次数，第一次超时此值是1，以此类推。此值被保存在：channelContext.stat.heartbeatTimeoutCount
     * @return 返回true，那么服务器则不关闭此连接；返回false，服务器将按心跳超时关闭该连接
     */
    @Override
    public boolean onHeartbeatTimeout(ChannelContext cc,
                                      Long interval,
                                      int heartbeatTimeoutCount) {
        Tio.unbindToken(cc);
        return false;
    }
  • 进行认证:
package cn.tio.demo.biz.auth;

import cn.tio.demo.socket.RequestMessageHandler;
import cn.tio.demo.socket.SocketUtil;
import com.tio.common.enums.InterfaceNameEnum;
import com.tio.common.enums.ResultCodeEnum;
import com.tio.common.socket.request.AuthRequestMessage;
import com.tio.common.socket.response.AuthResponseMessage;
import com.tio.common.util.JacksonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;

import javax.validation.ConstraintViolation;
import java.util.Objects;
import java.util.Set;

/**
 * 客户端认证请求
 *
 * @author
 */
@Slf4j
@Component
public class AuthRequestMessageHandler implements RequestMessageHandler<AuthRequestMessage> {

    @Value("${parking.parkId}")
    private String parkId;

    @Value("${parking.parkKey}")
    private String parkKey;

    @Value(("${parking.logPrint:false}"))
    private Boolean logPrint;

    @Override
    public String getService() {
        return InterfaceNameEnum.CHECK_KEY.getName();
    }

    @Override
    public void execute(ChannelContext cc, AuthRequestMessage message) {
        if (logPrint) {
            log.info("客户端认证信息:{}", message);
        }
        AuthResponseMessage authResponseMessage = new AuthResponseMessage();
        authResponseMessage.setService(InterfaceNameEnum.CHECK_KEY.getName());
        Set<ConstraintViolation<AuthRequestMessage>> constraintViolationSet = validParam(message);
        if (Objects.isNull(constraintViolationSet) || !constraintViolationSet.isEmpty()) {
            authResponseMessage.setResultCode(ResultCodeEnum.FAIL_2.getCode());
            authResponseMessage.setMessage("参数不能为空");
            SocketUtil.sendMsg(cc, JacksonUtils.toJson(authResponseMessage));
            return;
        }

        String reParkId = message.getParkId();
        String reParkKey = message.getParkKey();
        if (!reParkId.equals(parkId) || !reParkKey.equals(parkKey)) {
            authResponseMessage.setResultCode(ResultCodeEnum.FAIL_1.getCode());
            authResponseMessage.setMessage("认证失败，无此车场");
            SocketUtil.sendMsg(cc, JacksonUtils.toJson(authResponseMessage));
            return;
        }
        //绑定用户
        Tio.bindToken(cc, parkId);
        authResponseMessage.setResultCode(ResultCodeEnum.SUCCESS.getCode());
        authResponseMessage.setMessage("认证成功");
        SocketUtil.sendMsg(cc, JacksonUtils.toJson(authResponseMessage));

    }


}
  • 心跳
package cn.tio.demo.biz.heartbeat;


import cn.tio.demo.socket.RequestMessageHandler;
import cn.tio.demo.socket.SocketUtil;
import com.tio.common.enums.InterfaceNameEnum;
import com.tio.common.enums.ResultCodeEnum;
import com.tio.common.socket.request.HeartBeatRequestMessage;
import com.tio.common.socket.response.HeartbeatResponseMessage;
import com.tio.common.util.JacksonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;

import javax.validation.ConstraintViolation;
import java.util.Objects;
import java.util.Set;

/**
 * 服务器心跳响应结果
 *
 * @author
 */
@Slf4j
@Component
public class HeartBeatRequestMessageHandler implements RequestMessageHandler<HeartBeatRequestMessage> {

    @Value(("${parking.logPrint:false}"))
    private Boolean logPrint;

    @Override
    public String getService() {
        return InterfaceNameEnum.HEART_BEAT.getName();
    }


    @Override
    public void execute(ChannelContext cc, HeartBeatRequestMessage message) {
        if (logPrint) {
            log.info("接收到的客户端心跳结果:{}", message);
        }
        HeartbeatResponseMessage responseMessage = new HeartbeatResponseMessage();
        responseMessage.setService(InterfaceNameEnum.HEART_BEAT.getName());
        Set<ConstraintViolation<HeartBeatRequestMessage>> constraintViolationSet = validParam(message);
        if (Objects.isNull(constraintViolationSet) || !constraintViolationSet.isEmpty()) {
            responseMessage.setResultCode(ResultCodeEnum.FAIL_2.getCode());
            responseMessage.setMessage("参数不能为空");
            SocketUtil.sendMsg(cc, JacksonUtils.toJson(responseMessage));
            return;
        }
        SocketUtil.connect = cc;
        responseMessage.setResultCode(ResultCodeEnum.SUCCESS.getCode());
        responseMessage.setMessage("在线");
        SocketUtil.sendMsg(cc, JacksonUtils.toJson(responseMessage));
    }


}
  • 一个简单的骨架就搭建好了,试着跑一下:
  • 关键代码已经贴出来了,还是不懂得可以留言,看到就会回复
  • 这是我的微信公众号二维码,你的关注是我持续更新的动力,谢谢

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/langs/724396.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-04-26
下一篇 2022-04-26

发表评论

登录后才能评论

评论列表（0条）

保存