数据入队模块的-ExeModule
DataIn.ExeModule() 方法解析
这个方法与刚才的 DataOut 是配对的——DataIn 是生产者(往队列写数据),DataOut 是消费者(从队列读数据)。对照着看会更容易理解。
1. 确保队列存在(懒创建)
if (!Solution.Ins.QueueDic.ContainsKey(QueueKey))
Solution.Ins.QueueDic[QueueKey] = new JGTechVision.Services.DataOut(QueueKey);
和 DataOut 不同,DataIn 不要求队列事先存在——如果全局字典里没有这个 key,就直接创建一个。这样无论哪个模块先运行,队列都能正常工作。
2. 定义槽位结构(只做一次)
if (!_slotsDefined)
{
DefineAllSlots(outQueue); // 根据每个槽位的DataType,创建对应的List<double/int/string/bool>
_slotsDefined = true;
}
DefineAllSlots 遍历所有启用的槽位,根据数据类型调用 outQueue.DefineDoubleQueue() / DefineIntQueue() 等,在队列内部初始化对应的 List<T>。这个操作只执行一次(_slotsDefined 标志防止重复定义)。
3. 从上游取数据,写入队列
lock (outQueue)
{
foreach (var slot in QueueSlots.Where(s => s.IsEnable))
{
object val = GetLinkValue(slot.LinkVar.Text); // 从上游模块获取值
if (val == null) continue;
int idx = slot.SlotIndex;
switch (slot.DataType)
{
case "double":
var dList = (List<double>)outQueue.GetDataQueue(idx);
if (dList.Count >= outQueue.LimitLength) dList.RemoveAt(0); // 满了就丢掉最旧的
dList.Add(Convert.ToDouble(val));
break;
// ... int / string / bool 同理
}
}
}
这里的核心逻辑:
-
GetLinkValue(slot.LinkVar.Text):LinkVar.Text存的是类似"&模块名.变量名"的引用字符串,GetLinkValue通过这个引用去上游模块拿到实际值。这是框架的变量链接机制——相当于画一条线把上游模块的输出接到本模块的输入。 -
环形缓冲区(ring buffer):
if (dList.Count >= outQueue.LimitLength) dList.RemoveAt(0)—— 如果队列已满(达到LimitLength上限),就丢掉最旧的那条数据(索引0),然后追加新数据。这保证了队列不会无限增长。
4. 唤醒消费者(关键一步)
if (Solution.Ins.QueueSignDic.ContainsKey(QueueKey))
Solution.Ins.QueueSignDic[QueueKey].Set();
这就是和 DataOut 的 signal.WaitOne(TimeOut) 配对的地方。DataIn 写完数据后调用 Set(),把信号量置为有信号状态,这会唤醒正在 WaitOne 阻塞等待的 DataOut 线程,让它立刻去读取数据。
DataIn vs DataOut 对照
| DataIn(入队/生产者) | DataOut(出队/消费者) | |
|---|---|---|
| 方向 | 从上游模块拿值 → 写入队列 | 从队列取值 → 暴露为输出参数 |
| 数据来源 | GetLinkValue(LinkVar.Text) 变量链接 |
GetSlotValue(outQueue, slot) 从队列读 |
| 队列满时 | RemoveAt(0) 丢弃最旧数据 |
不适用(只读) |
| 队列空时 | 不关心 | 根据 IsWait 决定阻塞等待还是报错 |
| 信号量 | signal.Set() 唤醒对方 |
signal.WaitOne(TimeOut) 等待被唤醒 |
| 队列不存在时 | 自动创建 | 报错返回 |
一句话总结
DataIn 的 ExeModule 是一个带环形缓冲的生产者:它从上游模块通过变量链接拿到数据,按槽位和类型写入全局队列,写完后发信号唤醒等待中的 DataOut 消费者。队列满时自动丢弃最旧数据,防止内存无限增长。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐




所有评论(0)