学创消息推送服务接入说明文档

1、接入说明

强烈建议使用Netty接入学创消息服务，因为它具有如下优点：

1、Netty大大简化了网络程序的开发过程；

2、技术稳定可靠，采用异步非阻塞编程模型；

3、功能强大，预置了多种编解码功能，支持多种主流协议；解决了粘包/拆包的问题；

4、经历了大规模的商业应用考验，质量得到验证；

5、多语言、跨平台；

正因为这些优点，Netty逐渐成为 Java NIO 编程的首选框架。

2、Netty客户端接入流程

2.1流程时序图


2.2流程分析

  1. 用户线程创建 Bootstrap 实例 ，通过 API 设置创建客户端相关的参数，异步发起客户端连接。
  2. 创建处理客户端连接 、I/0 读写的 Reactor 线程组 NioEventLoopGroup。
  3. 通过Bootstrap的ChannelFactory 和用户指定的Channel 类型创建用于客户端连接的 NioSocketChannel;
  4. 创建默认的Channel Handler pipeline,用于调度和执行网络事件；
  5. 异步发起 TCP 连接， 判断连接是否成功。如果成功， 则直接将 NioSocketCbannel 注册到多路复用器上，监听读操作位，用于数据报读取和消息发送；如果没有立即连接成功，则注册连接监听位到多路复用器，等待连接结果；
  6. 注册对应的网络监听状态位到多路复用器；
  7. 由多路复用器在I/0现场中轮询各Channel, 处理连接结果；
  8. 如果连接成功，设置Future结果，发送连接成功事件，触发ChannelPipeline执行;
  9. 由Channel Pipeline调度执行系统和用户的Channel Handler, 执行业务逻辑。

 

3、客户端Demo

源码包：


ClientDemo.zip

找到com.enableets.edu.message.client.NettyClientRunner.java文件，运行main方法，控制台日志如下：

若出现服务端拒绝登入的情况，可能是由于重复登入造成，请修改deviceId后重试：

在控制台输入要发送给服务端的消息并回车，客户端将收到服务端返回的消息。Demo中消息的接收方是自身，开发者可自行修改消息的接收方；

 

4、客户端Demo源码解析

源码请参照附件中的“客户端源码”

4.1创建客户端连接

 

服务端的地址通过接口 http:// r30.enable-ets.com/microservice/configservice/v1/configuration 获取，请求将返回可用的服务端配置的集合。

由于服务端可能会存在服务挂掉等不可用的情况，开发者在选择服务进行连接时请使用随机抽选的方式；同样，当所选服务不可用或连接断开需要重连时，从服务列表中重新随机选择一项进行重连；

 

  1. private CloudServiceConfigurationDTO getMessageServer(){  
  2.         if (messageServerList.size() <= 1){  
  3.             return messageServerList.get(0);  
  4.         }else{  
  5.             Random random = new Random();  
  6.             return messageServerList.get(random.nextInt(messageServerList.size()));  
  7.         }  
  8. }  

 

  1. try {  
  2.             Bootstrap b = new Bootstrap();  
  3.             ((Bootstrap)b.group(group)).channel(NioSocketChannel.class)  
  4.                     .option(ChannelOption.TCP_NODELAY, true)  
  5.                     .handler(new ChannelInitializer<SocketChannel>() {  
  6.                         @Override  
  7.                         public void initChannel(SocketChannel ch)  
  8.                                 throws Exception { 
  9.                               // 加入消息处理handler 
  10.                             ch.pipeline().addLast("Handler",new Handler()); 
  11.                     });  
  12.             // 发起异步连接操作  
  13.             ChannelFuture future = b.connect(host, port).sync();  
  14.             future.channel().closeFuture().sync();  
  15. } finally {  
  16.           // 所有资源释放完成之后，清空资源，再次发起重连操作  
  17.             
  18.  }  

注意连接发生错误时，可在finally代码块中待资源释放完成后发起重连操作；

4.2编解码实现

数据对象编解码的处理采用MessagePack进行编解码，这是由其高效的性能，以及良好的多语言支持决定的，目前支持c/c++、java、ruby、python、php等开发语言。

自定义解码处理器可继承MessageToByteEncoder类，覆写其encode()方法：

  1. @Override  
  2. protected void encode(ChannelHandlerContext ctx, NettyMessage msg, ByteBuf out) throws Exception {   
  3.         try{  
  4.             byte[] row;  
  5.             if (msg.getBody() == null){  
  6.                 row = new byte[0];  
  7.             }else{  
  8.                 row = msgpack.write(msg.getBody());  
  9.             }  
  10.             out.writeInt(msg.getHeader().getCrcCode());  
  11.             out.writeInt(row.length);  
  12.             out.writeLong((msg.getHeader().getSessionID()));  
  13.             out.writeByte((msg.getHeader().getType()));  
  14.             out.writeByte((msg.getHeader().getPriority()));  
  15.             out.writeInt((msg.getHeader().getAttachment().size()));  
  16.   
  17.             String key = null;  
  18.             byte[] keyArray = null;  
  19.             String value = null;  
  20.             byte[] valueArray = null;  
  21.             for (Map.Entry<String, String> param : msg.getHeader().getAttachment().entrySet()) {  
  22.                 key = param.getKey();  
  23.                 keyArray = msgpack.write(key);  
  24.                 out.writeInt(keyArray.length);  
  25.                 out.writeBytes(keyArray);  
  26.                 value = param.getValue();  
  27.                 valueArray = msgpack.write(value);  
  28.                 out.writeInt(valueArray.length);  
  29.                 out.writeBytes(valueArray);  
  30.             }  
  31.   
  32.             if (msg.getBody()!=null) {  
  33.                 out.writeBytes(row);  
  34.             }  
  35.   
  36.         } catch (Exception e){    
  37.             throw e;  
  38.         }  
  39. }  

自定义解码处理器可继承ByteToMessageDecoder类，覆写decode()方法：

  1. @Override  
  2. protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {  
  3.         try {  
  4.             if (in == null || in instanceof EmptyByteBuf || in.readableBytes() < 0) {  
  5.                 return ;  
  6.             }  
  7.   
  8.             MessagePack msgpack = new MessagePack();  
  9.             NettyMessage nettyMessage = new NettyMessage();  
  10.             Header header = new Header();  
  11.   
  12.             int crcCode  = in.readInt();  
  13.             int length = in.readInt();  
  14.             long sessionId = in.readLong();  
  15.             byte type = in.readByte();  
  16.             byte priority = in.readByte();  
  17.   
  18.             header.setCrcCode(crcCode);  
  19.             header.setLength(length);  
  20.             header.setSessionID(sessionId);  
  21.             header.setType(type);  
  22.             header.setPriority(priority);  
  23.             nettyMessage.setHeader(header);  
  24.   
  25.             int size = in.readInt();  
  26.             if (size > 0) {  
  27.                 Map<String, String> attch = new HashMap<String, String>(size);  
  28.                 int keySize = 0;  
  29.                 byte[] keyArray = null;  
  30.                 String key = null;  
  31.                 int valueSize = 0;  
  32.                 byte[] valueArray = null;  
  33.                 String value = null;  
  34.                 for (int i = 0; i < size; i++) {  
  35.                     keySize = in.readInt();  
  36.                     keyArray = new byte[keySize];  
  37.                     in.readBytes(keyArray);  
  38.                     key = msgpack.read(keyArray,String.class);  
  39.                     valueSize = in.readInt();  
  40.                     valueArray = new byte[valueSize];  
  41.                     in.readBytes(valueArray);  
  42.                     value =msgpack.read(valueArray,String.class);  
  43.                     attch.put(key, value);  
  44.                 }  
  45.                 keyArray = null;  
  46.                 key = null;  
  47.                 valueArray = null;  
  48.                 value = null;  
  49.                 header.setAttachment(attch);  
  50.             }  
  51.   
  52.   
  53.             if (in.readableBytes() < length) {  
  54.                 in.resetReaderIndex();  
  55.                 return;  
  56.             }  
  57.   
  58.             in.markReaderIndex();  
  59.   
  60.             final ByteBuf frame;  
  61.             final int lg = in.readableBytes();  
  62.             if (lg != 0) {  
  63.                 frame = in.readBytes(lg);  
  64.                  if(type == 0){  
  65.                     nettyMessage.setBody(msgpack.read(frame.array(), ClientToServerMsgBody.class));  
  66.                 }else if (type == 1){  
  67.                     nettyMessage.setBody(msgpack.read(frame.array(), ServerToClientMsgBody.class));  
  68.                 }else{  
  69.                     nettyMessage.setBody(frame.array()[0]);  
  70.                 }  
  71.             }  
  72.   
  73.             out.add(nettyMessage);  
  74.         }catch (Exception e){  
  75.             logger.error(e.getMessage(),e);  
  76.             throw e;  
  77.         }  
  78.   
  79.   
  80. }  

4.3登陆握手

当连接通道成功激活时进行登陆操作，LoginBody包含以下数据项：

  1. @Message  
  2. public class LoginBody {  
  3.     @Index(0)  
  4.     private String clientId;  
  5.     @Index(1)  
  6.     private String secret;  
  7.     @Index(2)  
  8.     private String deviceId;  
  9.     @Index(3)  
  10.     private String tags;  
  11. }  

clientId 及 secret联系管理员进行申请，deviceId为登陆设备标识，必须确保唯一，同一clientId下的同一个deviceId只能同时一个在线，当deviceId已经在线，其它的握手将被拒绝；

tags表示当前连接的身份标识，有多个时请用英文逗号隔开；

向服务端发送握手消息时，将header中的type参数设为3；

服务端接到客户端的握手登入请求后会将处理结果返回给客户端，客户端接到服务端type为4的消息类型是进行握手是否成功做出判断，握手登入成功时服务端返回1（byte类型），失败时返回0（byte类型）：

  1. public void doService(NettyMessage message) {  
  2.         byte loginResult = (byte) message.getBody();  
  3.         if (loginResult != (byte) 0) {  
  4.             // 握手失败，关闭连接  
  5.             Logger.error("登陆失败");  
  6.             getContext().close();  
  7.             System.exit(-1);  
  8.         } else {  
  9.             Logger.error("Login is ok : " + message);  
  10.             getContext().fireChannelRead(message);  
  11.         }   
  12.     }  

4.4心跳发送与接收处理

握手成功后客户端开始以5s/次的固定频率向服务端发送心跳，服务端收到心跳请求后返回心跳响应消息

  1. public void doService(NettyMessage message) {  
  2.         if (message.getHeader().getType() == 4) {  
  3.             heartBeat = getContext().executor().scheduleAtFixedRate(  
  4.                     new HeartBeatReqHandler.HeartBeatTask(getContext()), 0, 5000,  
  5.                     TimeUnit.MILLISECONDS);  
  6.         } else if (message.getHeader().getType() == 6) {  
  7.             LOG.debug("Client receive server heart beat message : ---> " + message);  
  8.         }    
  9. }  

4.5 接收消息

服务端向客户端推送的消息数据结构如下：

  1. @Message  
  2. public class ServerToClientMsgBody {  
  3.     /** 消息来源标识 */  
  4.     @Index(0)  
  5.     private String sourceId;  
  6.   
  7.     /** 消息来源名称 */  
  8.     @Index(1)  
  9.     private String sourceName;  
  10.   
  11.     /** 消息内容 */  
  12.     @Index(2)  
  13.     private String content;  
  14.   
  15.     /** 消息推送时间 */  
  16.     @Index(3)  
  17.     private Date pushTime;  
  18. }  

客户端接收到服务端的推送消息后，可自行按照业务逻辑进行处理；建议收到消息后进行异步处理，否则如果消息通道阻塞时间过长后容易被服务端当作无效连接清除掉。

4.6发送消息

首先当通道连接成功时通过channelActive(ChannelHandlerContext ctx)函数将ChannelHandlerContext对象保存到线程副本对象中：

  1. /** 当前线程context */  
  2. ThreadLocal<ChannelHandlerContext> threadContext;  
  3.   
  4. @Override  
  5. public void channelActive(ChannelHandlerContext ctx) throws Exception {  
  6.         threadContext = ThreadLocal.withInitial(() -> ctx);  
  7. }  

 

当客户端与服务端握手认证成就功后就可以通过该channel对象向服务端发送消息了：

  1. public void sendMessage(NettyMessage message) {  
  2.     if (message == null) {  
  3.         return;  
  4.     }  
  5.     threadContext.get().writeAndFlush(message);  
  6. }  

发送消息参数结构：

  1. @Message  
  2. public class ClientToServerMsgBody {  
  3.     /** 发送对象 */  
  4.     @Index(0)  
  5.     private Audience audience;  
  6.     /** 消息信息 */  
  7.     @Index(1)  
  8.     private String message;   
  9.     /** 额外信息 */  
  10.     @Index(2)  
  11.     private Options options;  
  12.   
  13.     @Message  
  14.     public static class Audience implements Serializable {  
  15.         @Index(0)  
  16.         private Set<String> tags;  
  17.         @Index(1)  
  18.         private Set<String> tagAnd;  
  19.         @Index(2)  
  20.         private Set<String> tagNot;  
  21.     }  
  22.   
  23.     @Message  
  24.     public static class Options implements Serializable {  
  25.         @Index(0)  
  26.         private Integer timeToLive;         
  27.     }  
  28. }  

4.7连接超时处理

当因为网络原因或其它程序错误连接通道长时间（50s即10个心跳周期）未进行通信时，服务端将会断开连接进行资源回收；同理当客户端长时间无法与服务端进行有效通信时，应该主动断开连接并尝试重连。

客户端超时机制的实现可以通过netty内置的ReadTimeoutHandler()实现，创建客户端时将readTimeoutHandler添加进pipeline的处理链即可。

 

5、消息参数详解

5.1 服务端消息参数

参数名称

类型

说明

sourceId

String

消息发送端标识

sourceName

String

消息发送端名称

content

String

消息内容字符串，具体格式规范由发送方定义

pushTime

Date

消息推送的时间

 

客户端接收到的消息与自身握手认证时传入的tags有关，当客户端tags与服务端消息目标tags匹配时将会收到该消息的推送，具体匹配规则参照4.2。通常客户端接收到消息后只需要关注content参数，解析该参数进行具体业务操作。

5.2 客户端消息参数

参数名称

类型

说明

audience

object

推送目标,包括三组字符串数组

tags: 多个标签之间是 OR 的关系，即取并集。

tagAnd: 多个标签之间是 AND 关系，即取交集。

tagNot: 多个标签之间，先取多标签的并集，再对该结果取补集。

message

String

消息内容，将被推送到目标客户端

options

Object

可选参数：

timeToLive: 单位s,当目标离线时消息保存时间，默认为7天

注意，以上三种目标类型至少需要有其一。如果值数组长度为 0，表示该类型不存在。

这几种类型可以并存，多项的隐含关系是 AND，即取几种类型结果的交集。

例如：

"audience" : { "tag" : [ "tag1", "tag2" ], "tag_and" : [ "tag3", "tag4"], "tag_not" : [ "tag5", "tag6"] }

先计算 "tag" 字段的结果 tag1或tag2=A;

再计算 "tag_and" 字段的结果 tag3且tag4=B;

再计算 "tag_not" 字段的结果 非(tag5或tag6)=C

"audience" 的最终结果为 A且B且C 。

6、附录

6.1 消息协议

客户端与服务端通讯消息协议栈包含两部分：

①       消息头

②       消息体

其具体定义见下面参见两张表

消息定义表：

名称

类型

长度

说明

header

Header

变长

消息头定义

body

Object

变长

消息体

 

header定义表

名称

类型

长度

说明

crcCode

整型int

32

固定值: 0xabef0101

length

整型int

32

消息长度整个消息,包括消息头和消息体

sessionID

长整型long

64

会话id

type

byte

8

0: 业务请求消息

1: 业务响应消息

2: 业务ONEWAY消息（既是诮求又是响应消息）

3: 握手请求消息

4: 握手应答消息

5: 心跳请求消息

6: 心跳应答消息

priority

byte

8

优先级

attachment

Map<String, String>

变长

可选字段，用于拓展消息头