ddpush 学习之路 13 Receiver.java

本篇文章。我们来学习ddpush 的 Receiver.java 这个类的作用是 专门处理或者说是专门用来接收 客户端向UDP服务器发送消息的这个 接收 和存储 的过程。这里的存储 是存放在该类的一个消息队列中。

好了来上代码。我都尽量添加比较多的注释。方便大家看懂。

//UDP 服务器 接受客户端消息的处理类
public class Receiver implements Runnable{

//UDP 服务器Cannel
protected DatagramChannel channel;
//接受客户端消息的最大消息长度
protected int bufferSize = 1024;
//服务器是否被stop
protected boolean stoped = false;
//byte 缓冲区
protected ByteBuffer buffer;
//socket 地址
private SocketAddress address;
//添加到消息队列中的消息计数器
protected AtomicLong queueIn = new AtomicLong(0);
//从消息队列中移除的消息计数器
protected AtomicLong queueOut = new AtomicLong(0);
//消息队列 这里的消息队列主要存放 ClientMessage 客户端消息包对象
protected ConcurrentLinkedQueue<ClientMessage> mq = new ConcurrentLinkedQueue<ClientMessage>();

//构造
public Receiver(DatagramChannel channel){
this.channel = channel;
}

//初始化 给buffer缓冲区分配内存空间
public void init(){
buffer = ByteBuffer.allocate(this.bufferSize);
}

//停止服务器
public void stop(){
this.stoped = true;
}

//Receiver 消息接收线程体
public void run(){
//如果服务器没有被stop 就一直接收消息
while(!this.stoped){
try{
//synchronized(enQueSignal){
//接收消息
processMessage();
//  if(mq.isEmpty() == true){
//      enQueSignal.wait();
//  }
//}
}catch(Exception e){
e.printStackTrace();
}catch(Throwable t){
t.printStackTrace();
}
}
}

//接收消息
protected void processMessage() throws Exception{
//清除上次收到的数据
address = null;
buffer.clear();
try{
//接收客户端发送的UDP消息
address = this.channel.receive(buffer);
}catch(SocketTimeoutException timeout){

}
//如果消息发送者的地址没找到 服务器就丢弃这个接收到的客户端消息
if(address == null){
try{
Thread.sleep(1);
}catch(Exception e){

}
return;
}

//limit 设置位当前position   position 设置为0
buffer.flip();
//将buffer里接收到的客户端发送过来的数据拷贝到一个新的临时数组
byte[] swap = new byte[buffer.limit() - buffer.position()];
System.arraycopy(buffer.array(), buffer.position(), swap, 0, swap.length);
//根据这个数据发送者的地址 和发送过来的数据 创建一个ClientMessage对象就是客户端消息包
ClientMessage m = new ClientMessage(address,swap);
//将这个收到的客户端小细胞 放入 消息队列中
enqueue(m);
//System.out.println(DateTimeUtil.getCurDateTime()+" r:"+StringUtil.convert(m.getData())+" from:"+m.getSocketAddress().toString());

}

//入队 将收到的客户端消息包 放入 消息队列中
protected boolean enqueue(ClientMessage message){
boolean result = mq.add(message);
if(result == true){
queueIn.addAndGet(1);
}
return result;
}

//出队 从消息队列中取出一个 客户端消息对象
protected ClientMessage dequeue(){
ClientMessage m = mq.poll();
if(m != null){
queueOut.addAndGet(1);
}
return m;
}

//取出一个 通过数据类型检查的 客户端消息
//就是从消息队列中取出一个MessageClient 客户端发过来的消息
//并且做一下 消息包的格式验证。
public ClientMessage receive(){

ClientMessage m = null;
while(true){
m = dequeue();
if(m == null){
return null;
}
if(m.checkFormat() == true){//检查包格式是否合法，为了网络快速响应，在这里检查，不在接收线程检查
return m;
}
}
}
}

前面上面我就说了这个类的主要作用。然后发出来了经过我注释的代码。

下面我们来分析一下。当然我们还要再重复一遍 这个类的作用是 接受客户端发送过来的UDP消息。并把UDP消息存储起来 同时提供一个函数用来在外部获取客户端发送过来的消息。

好了。大概说一下流程把。代码也比较简单。相信我说过流程之后。大家再把代码看一遍。就能看的很清晰 很明确了。也会很懂了。

好了。这个类和上一个 UDP 发射器类 一样。是一个线程类。 有run函数。

这个类工作的流程如下

外部创建这个Receiver类线程对象 并 start 然后对象内部的run函数 就在一直循环的processMessage 这个processMessage里 接受客户端发送过来的消息。然后把这个消息封装成 ClientMessage 客户端消息对象 并把这个客户端消息对象存放到 Receiver 类对象内部的 消息队列 中 这个消息队列就是

ConcurrentLinkedQueue<ClientMessage> mq

这个接受线程就一直在做这个工作。接收 添加到消息队列、接收 添加到消息队列。

然后这个被添加到消息队列中的消息 外部怎么去处理呢。 这个类中提供了一个外部获取客户端消息的函数。

public ClientMessage receive()

这个函数。就是专门供外部调用的。用来获取一个经过数据格式检查过的有效的客户端消息包 的函数。

基本上就以上这么多东西。 下面在简短的说一下 ddpush 中 UDP服务器接受消息的流程。

run函数内部一直在while循环不停的去读取客户端发过来的数据。接受到一个客户端发送过来的数据 就把这个接受到的 客户端数据 包装成 客户端消息 对象 然后添加到 消息队列中。

然后在外部。调用Receiver 里面的receive()函数。从 客户端消息队列中 取出 有效的 客户端消息 然后在外部处理这个收到的客户端UDP消息

好了以上 就是 ddpush 学习之路 13 Receiver.java 的内容。

by brok1n 20150328