摘要：
消息处理流程构建一个MDU模块，注册入框架中，初始MDU没有注册PID，未构建消息处理任务。通过消息框架提供的消息发送接口直接发送消息，消息框架根据接收PID信息，将消息填入对应MDU的消息队列中。MDU消息队列会被多个任务并发写入消息，被消息处理任务读取消息处理，需要对消息队列进行互斥和同步。如果MDU中包含太多PID，由于所有PID在一个消息队列中串行运行，会影响PID的响应，影响系统性能。耗时PID和实时要求高的PID不放入一个MDU中。
- 基本框架
- MDU(消息分发单元):包含一个消息处理任务，包含自身的消息队列，是一个消息调度的基本单位。
- PID (功能子模块) :框架中用PID作为模块的划分，每个模块具有自己的PID编号，根据功能和调度需求可以安排多个PID到一个MDU中，PID是消息通信的一个基本单位，每个PID提供一个消息处理入口。
- MQ (消息队列) :使用消息队列作为任务通信的数据结构。
- 消息处理流程
- 构建一个MDU模块，注册入框架中，初始MDU没有注册PID，未构建消息处理任务。
- 构建PID，注册入对应的MDU中，如果是MDU中第一个PID，构建消息处理任务。消息处理任务从该MDU对应的消息队列中取消息处理。
- 消息处理任务获取消息后根据消息中携带的接收PID的信息分发到对应的PID模块处理。
- 完整的消息交互流程
- 任务A申请消息，消息内容必须包括发送模块PID编号、接收模块PID编号、消息内容。
- 通过消息框架提供的消息发送接口直接发送消息，消息框架根据接收PID信息，将消息填入对应MDU的消息队列中。
- MDU的消息处理任务B从消息队列中获取消息处理。
- MDU消息队列会被多个任务并发写入消息，被消息处理任务读取消息处理，需要对消息队列进行互斥和同步。详见http://www.cnblogs.com/chencheng/p/2893421.html
- MUD、PID规划
- MDU作为一个调度基本单元，如果一个MDU中只有一个PID会导致系统中任务多，任务切换的开销大。
- 如果MDU中包含太多PID，由于所有PID在一个消息队列中串行运行，会影响PID的响应，影响系统性能。
- 功能紧耦合的PID放入一个MDU中。
- 耗时PID和实时要求高的PID不放入一个MDU中。
- 实现
MDU:
importmyQueue from myThread importmyThread classmdu: def __init__(self, mduID): self.mduId =mduID self.msgQue = myQueue.myQueue(10) self.map ={} defgetMduID(self): returnself.mduId defregistPid(self, pidID, pid): self.map[pidID] =pid if 1==len(self.map): self.run() defmsgEnQueue(self,msg): self.msgQue.enQueue(msg) defmsgProcess(self): whileTrue: msg =self.msgQue.deQueue() recvPid =msg.getRecvPid() self.map[recvPid].msgProcess(msg); defrun(self): t =myThread(self.msgProcess) t.start()
PID:
importmessage importsupport importmdu importpdb classpid: def __init__(self, pidID): self.pidID =pidID self.registMe() defregistMe(self): support.registPid(self) defgetPidID(self): return self.pidID
SUPPORT:
importmdu importpdb mduMap ={} defregistMdu(mdu): mduMap[mdu.getMduID()] =mdu defgetMdu(revPid): return mduMap[revPid&0xFFFF0000>>16] defregistPid(pid): mdu =getMdu(pid.getPidID()) #pdb.set_trace() mdu.registPid(pid.getPidID(), pid) defsendMsg(msg): mdu =getMdu(msg.getRecvPid()) mdu.msgEnQueue(msg)
MQ:
from threading importLock from threading importCondition importthreading classmyQueue: def __init__(self, size): self.size =size self.list =list() self.lock =Lock() self.notFullCond =Condition(self.lock) self.notEmptyCond =Condition(self.lock) defisFull(self): if self.size ==len(self.list): returnTrue returnFalse defisEmpty(self): if 0 ==len(self.list): returnTrue returnFalse defenQueue(self, elem): self.lock.acquire() whileself.isFull(): print('queue is full, waiting...') self.notFullCond.wait() print(threading.current_thread().getName() + 'product ' +str(elem)) self.list.append(elem) self.notEmptyCond.notify() self.lock.release() defdeQueue(self): self.lock.acquire() whileself.isEmpty(): print('queue is empty, waiting...') self.notEmptyCond.wait() elem =self.list[0] del(self.list[0]) print(threading.current_thread().getName() + 'consume ' +str(elem)) self.notFullCond.notify_all() self.lock.release() return elem