黑马程序员技术交流社区

标题: 常用通信接口的开发实践 [打印本页]
作者: chenjianxiang    时间: 2014-12-2 20:38
标题: 常用通信接口的开发实践
常用的接口之间的通信协议有Tcp和Http的以及webService的，这里使用介绍一下如何使用Mina框架开发一个Tcp协议的通信接口：sendRenChuangMessage：
@Component("sendRenChuangMessage")
@Scope("prototype")
public class SendRenChuangMessage {
    protected Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 超时时间
     */
    @Value("${renchuang.socket-keepAliveRequestTimeout}")
    private String connectTimeOut;

    /**
     * 请求处理类
     */
    @Autowired
    @Qualifier("renChuangMessageHandler")
    private IoHandler serverProtocolHandler;

    /**
     * 编解码工厂类
     */
    @Autowired
    @Qualifier("renChuangCodec")
    private ProtocolCodecFactory factory;


    /**
     * 服务端主机IP
     */
    @Value("${renChuang.IP}")
    private String host;
    /**
     * 服务端端口号
     */
    @Value("${renChuang.port}")
    private String port;

    private String ResultMsg;

    /**
     * 连接，调用“缴费销账”接口
     *
     * @param reqXml “缴费销账”接口的请求报文
     */
    public void sendMsg(String reqXml, Exchange exchange) {
        Object obj = new Object();
        logger.info("调用缴费销账接口，开始!");
        logger.info("调用缴费销账接口,请求地址:{},请求端口:{}", host, port);
        IoConnector conn = new NioSocketConnector();
        // 设置链接超时时间
        conn.setConnectTimeoutMillis(30000L);
        // 添加过滤器
//        conn.getFilterChain().addLast("codec",new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF-8"))));
        conn.getFilterChain().addLast("codec", new ProtocolCodecFilter(factory));
        // 添加业务处理handler
        conn.setHandler(serverProtocolHandler);
        IoSession session = null;
        try {
            ConnectFuture future = conn.connect(new InetSocketAddress(host, Integer.parseInt(port)));// 创建连接
            future.awaitUninterruptibly();// 等待连接创建完成
            session = future.getSession();// 获得session
            //保存“锁”对象，用于在RenChuangDecoder中唤醒这个线程
            Long sessionId = session.getId();
            LockUtil.hashMap.put(sessionId,obj);
            //发送请求报文
            WriteFuture writeFuture = session.write(reqXml);// 发送消息
            // 等待结果返回，如果超过指定时间就超时了
            synchronized (obj) {
                try {
                    obj.wait(Integer.parseInt(connectTimeOut) * 1000L);
                } catch (InterruptedException e) {
                    // 记录日志
                    logger.error("wait timeout " + e.getMessage());
                }
            }
        } catch (Exception e) {
            logger.info("调用充值接口发送异常：{}", e);
            return;
        }
        // 输出结果信息
        Message mes = exchange.getOut();
        exchange.getOut().copyFrom(exchange.getIn());
        mes.setBody(RechargeResultUtil.rechargeResultMap.get(session.getId()), String.class);
        session.close(true);
    }


RenChuangEncoder：
  1. @Component
  2. public class RenChuangEncoder extends ProtocolEncoderAdapter {
  3.     private Logger logger = LoggerFactory.getLogger(getClass());
  4.     @Override
  5.     public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
  6.         logger.debug("RenChuangEncoder In. Msg:{}", message);

  8.         String value = (String) message;

  10.         IoBuffer buffer = IoBuffer.allocate(100).setAutoExpand(true);
  11.         buffer.put(value.getBytes("GBK"));

  13.         buffer.flip();
  14.         out.write(buffer);

  16.         logger.debug("RenChuangEncoder Out.");
  17.     }
  18. }
复制代码

RenChuangDecoder：
  1. @Component
  2. public class RenChuangDecoder extends CumulativeProtocolDecoder {
  3.     private Logger logger = LoggerFactory.getLogger(getClass());

  5.     @Override
  6.     protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
  7.         int size = 0;
  8.         byte[] sizes = null;
  9.         try {

  11.             int fullSize = in.remaining();
  12.             logger.info("仁创返回报文的所有字节数为:{}",fullSize);
  13.             // <EP>00000244
  14.             if (in.remaining() >= 12) {// 有数据时，超过“报文头长度时”，读取82字节判断消息长度

  16.                 sizes = new byte[4];
  17.                 in.mark();// 标记当前位置，以便reset
  18.                 //读取报文的前4个字节存入这个字节数组中
  19.                 in.get(sizes);

  21.                 //报文长度
  22.                 size = Integer.parseInt(new String(sizes));
  23.                 byte[] rechargeResultCodeByte = new byte[8];
  24.                 in.get(rechargeResultCodeByte);
  25.                 //得到“交易结果码”
  26.                 String rechargeResultCode = new String(rechargeResultCodeByte);
  27. //                //这种情况说明返回“报文头”中的“交易结果说明”不是空的，并且是50位长
  28. //                if (!StringUtil.equals("10000000", rechargeResultCode)) {
  29. //                    byte[] rechargeResultDescByte = new byte[50];
  30. //                    in.get(rechargeResultDescByte);
  31. //                    //交易结果说明
  32. //                    String rechargeResultDesc = new String(rechargeResultCodeByte);
  33. //                    logger.info("交易结果说明:{}，长度:{}", size, rechargeResultCodeByte.length);
  34. //                }
  35.                 logger.info("接受到报文体长度:{},交易结果码:{}", size,rechargeResultCode);

  37.                 if (size > in.remaining()) {// 如果消息内容不够，则重置，相当于不读取size
  38.                     logger.info("消息内容不全. size:{}, receveSize:{}", size, in.remaining());
  39.                     in.reset();
  40.                     return false;// 接收新数据，以拼凑成完整数据
  41.                 } else {
  42. //                    in.flip();
  43.                     in.clear();
  44.                     byte[] bytes = new byte[fullSize];
  45.                     in.get(bytes, 0, fullSize);
  46.                     out.write(new String(bytes));
  47.                     logger.info("RenChuangDecoder 返回的字符：{}",new String(bytes));
  48.                     if (in.remaining() > 0) {
  49.                         return true;
  50.                     }
  51.                 }
  52.             }
  53.         }catch (NumberFormatException e){
  54.             logger.info("报文长度转换异常，报文长度字符串sizes:{}",new String(sizes));
  55.         }catch (Exception e) {
  56.             logger.error("仁创接受消息解码失败, ByteBuffer:{}", in, e);
  57.         }
  58.         return false;// 处理成功，让父类进行接收下个包
  59.     }
  60. }
复制代码








欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/) 黑马程序员IT技术论坛 X3.2