Chapter 1 起步

你是否纠结过：如何避免为各种常见的工业硬件编写“重复”的代码？

Itminus.Tags 是我编写的一套面向工业硬件通信场景的极简通信类库。无论底层通信是OpcUa、S7、还是ModbusTcp，或者是串口、甚至可能是MQTT，Itminus.Tags 都能让你以统一的方式，编写横跨各种工业硬件的通信代码。它有如下几个设计理念：

  • 模块化+可扩展： 各模块之间互相独立，没有耦合。以.AddS7Support()的方式添加S7通信支持，以.AddComScannerSupport()方式添加串口扫码枪通信支持……没发现你想要的硬件支持？你可以自己轻易扩展。
  • 允许组合读写：可以用整体读写的方式来减少IO次数。
  • 刚好够用：只提供刚好够用的功能，简单易学，符合直觉，没有魔法。

这个类库是前些年我过年放假在家时编写的，主要目的是为了降低门槛（足够简单）、提高灵活性，后来经过逐步完善，成为一个免费且开源项目，在这个过程中，也慢慢用到了公司的各个商业项目中。

在着手编写这个项目之前，我倾向于使用强类型的方式，在类型实例和一段连续的字节数组之间做递归式地自动映射，这种做法运行效率非常高，而且还有超级友好的类型提示，我曾偏爱这种代码风格接近十年。但不可否认，这也带来了两个坏处：

一是要求编程人员熟知内存布局，比如要求大家知道int在C#中会占据4个字节，而在西门子PLC中占据2个字节；再比如，编程者需要知道西门子PLC的字符串会在头部附加两个字节，用来存储总字节数和有效字节数。这一点无疑拔高了编程门槛。

二是点表和程序代码一一对应的，这意味着动态修改比较麻烦。熟悉我的朋友都知道，我一贯强烈反对把修改代码的工作延迟到现场，但是我直到最近才认可这和提供灵活性不完全冲突。

快速上手

依赖包

首先，我们需要添加基础包依赖：

dotnet add package Itminus.Tags

然后，根据项目所需要通信的底层硬件，选择一个或者多个具体的实现包。目前内置的硬件支持包括：

驱动包名备注
西门子S7Itminus.Tags.S7支持
OpcUAItminus.Tags.OpcUa支持
华杰智控IO盒子Itminus.Tags.Hjzk支持
串口Itminus.Tags.ComScanner实验
ModbusTcpItminus.Tags.ModbusTcp实验
卓兰IO盒子Itminus.Tags.ZLan实验

Itminus.Tags支持的协议和硬件会持续增加。如果暂时没有你想要的，可以自由扩展实现你自己的驱动。

然后向注册相关服务和硬件支持：

services.AddTagsProjectServices(builder =>
{
    builder.AddS7Support();
    // 增加其它更多的硬件支持，比如：
    // builder.AddComScanner();
    // ...
});

如果你需要使用Rx.NET，可以添加扩展包： Itminus.Tags.RxExtensions

项目构建和运行

让我们构建一个项目，并将它跑起来：

using var scope = this._ssf.CreateScope();
var sp = scope.ServiceProvider;

// 提供项目运行目录，其中有通信点表和可能用到的插件
var dir ="D:/manufacture/pl01/";    
// 构建项目
var factory = sp.GetRequiredService<ITagsProjectFactory>();
using var proj = factory.Create(dir!, root: null);

// ...可选：hooks

// 运行项目
await proj.RunAsync(ct);

这里目录下通常会有一份测点索引(index.xml)，内容类似于：

<?xml version="1.0" encoding="utf-8"?>
<root>

	<!-- 通道，可以配置多个-->
	<Channel name="S7-1" driver="S7" >
		<IpAddr>localhost</IpAddr>
		<Rack>0</Rack>
		<Slot>1</Slot>
	</Channel>

	<!-- 测点配置，可以配置多个 -->
	<TagGrp name="CallMat" isEntry="true" isEnabled="true" channel="S7-1" scanInterval="10">

		<TagGrp name="1#">
			<TagGrp name="songliao">
				<TagCbnt name="PLC" address="DB1990.0">
					<Tag name="叫料请求" address="$$10.0" type="BIT" access="RO"></Tag>
					<Tag name="进站响应" address="$$10.1" type="BIT" access="RO"></Tag>
					<Tag name="超时响应" address="$$10.2" type="BIT" access="RO"></Tag>
					<Tag name="叫料序号" address="$$12" type="INT32" access="RO" endian="BigEndian"></Tag>
				</TagCbnt>
				<TagCbnt name="MST" address="DB1991.0" access="RW">
					<Tag name="叫料应答" address="$$10.0" type="BIT"></Tag>
					<Tag name="进站请求" address="$$10.1" type="BIT"></Tag>
					<Tag name="超时请求" address="$$10.2" type="BIT"></Tag>
					<Tag name="状态" address="$$12" type="INT16" endian="BigEndian"></Tag>
					<Tag name="NG原因" address="$$14" type="INT16" endian="BigEndian"></Tag>
					<Tag name="叫料编号" address="$$16" type="STR" maxlen="24"></Tag>
				</TagCbnt>
			</TagGrp>
			
			<TagGrp name="paomian">
				<TagCbnt name="PLC" address="DB1990.0">
					<Tag name="叫料请求" address="$$30.0" type="BIT" access="RO"></Tag>
					<Tag name="进站响应" address="$$30.1" type="BIT" access="RO"></Tag>
					<Tag name="超时响应" address="$$30.2" type="BIT" access="RO"></Tag>
					<Tag name="叫料序号" address="$$32" type="INT32" access="RO" endian="BigEndian"></Tag>
				</TagCbnt>
				<TagCbnt name="MST" address="DB1991.0" access="RW">
					<Tag name="叫料应答" address="$$60.0" type="BIT"></Tag>
					<Tag name="进站请求" address="$$60.1" type="BIT"></Tag>
					<Tag name="超时请求" address="$$60.2" type="BIT"></Tag>
					<Tag name="状态" address="$$62" type="INT16" endian="BigEndian"></Tag>
					<Tag name="NG原因" address="$$64" type="INT16" endian="BigEndian"></Tag>
					<Tag name="叫料编号" address="$$66" type="STR" maxlen="24"></Tag>
				</TagCbnt>
			</TagGrp>
		</TagGrp>
	</TagGrp>

	<!-- 逻辑配置，可以配置0~N个 -->
	<!--<Logicet>Samples.Plugin1.dll</Logicet>-->
</root>

除了使用约定的index.xml，我们也可以用程序的方式自己构建出一个XML节点，然后传递给factory.Create(dir!, root: null)这里的root参数，这样项目工厂就不会去搜寻index.xml文件了。

备注

如果我的代码能帮到你，我们希望你也能在力所能及的情况下，回馈开源社区。程序员的生涯也许是非常短暂的，但是一旦你的作品开始发出微弱的光芒，就能轻松照亮无数的后来者快步前行。

如果需要和我探讨编程技术，可以通过(aXRtaW51c0Bob3RtYWlsLmNvbQ==)联系我。或许在我有空的时候，大家可以畅聊人生。

由于个人精力有限，我只能优先实现我自己需要的特性。对于一些尚未支持的通信协议，我们鼓励你自己动手写实现。如果你自己不想花太多精力，也可以付费购买其他开发者的时间。

Chapter 2 基本概念

Itminus.Tags用测点项目(ITagsProject)来组织各个模块：

\( \boxed{\text{测点项目}} = \boxed{\text{通道}} + \boxed{\text{测点}} + \boxed{\text{逻辑}} \)

通道(ITagChannel)

顾名思义，它规定了和底层硬件通信的信息“通道”:

public interface ITagChannel : IDisposable
{
    /// <summary>
    /// 通道名称
    /// </summary>
    string ChannelName { get; }

    /// <summary>
    /// 驱动
    /// </summary>
    string Driver { get; }

    /// <summary>
    /// 连接
    /// </summary>
    /// <returns></returns>
    Task EnsureConnectedAsync(bool force, CancellationToken ct);

    /// <summary>
    /// 关闭
    /// </summary>
    /// <returns></returns>
    Task DisconnectAsync(CancellationToken ct);
}

每种通道的读写方式并不一致。比如，几乎每种PLC都会使用一整片连续字节数组来存储数据；而对应扫码枪而言，它很可能基于串口或者USB通信的方式，收取最近一次扫描到的值。

一个测点项目中可以定义1~n个通道(<Channel/>)。

测点(ITag)

在工业通信中，往往有一层非常简单的抽象——用一个寄存器表示一个值。比如用浮点型寄存器表示当前机器人的某个关节轴的角度，或者用一个比特表示外部开关的通断信号。

测点——可观测的点位——是对上面这些表达方式的绝佳抽象。它们往往和硬件寄存器的地址相关，又可以在高层读取或者写入。Itminus.Tags提供了三种层次的测点抽象：

  • Tag: 代表单个叶子节点，比如一个数字输入信号、一个数字输出信号、一个16位整型变量、一个32位浮点型变量等等。所有的Tag都是叶子节点，它们都不能再嵌套子节点。
  • TagCbnt：代表一个测点组合(这里Cbnt是Combination的缩写)。有经验的程序员都知道，工业相关程序中IO往往是瓶颈，减少IO至关重要。我们可以把一些测点进行组合，从底层硬件中整体读取数据、并向底层整体写入，这种整体读写的组合，我们叫TagCbnt(Cbnt是Combination的缩写),TagCnbt的子元素只能是Tag。
  • TagGrp: 很多测点往往共享一些特征，比如通道、功能等，这些测点在逻辑上可以划分到一个群组，我们用TagGrp来表达这个概念(显然，Grp是Group的缩写)。测点群组可以继续嵌套TagGrp、TagCbnt、或者Tag。通常我们会定义 1~n个测点组(TagGrp)作为入口。

测点组可以通过[]索引的方式访问直接子元素。

var child = grp[childName]

这种索引式操作非常类似于字典元素的访问方式——如果对应的元素不存在，它也会报异常。

要访问内部某个子元素下的某个子元素，可以使用多级索引的方式：

var child2 = grp[_path][_to][_child]

此外，我们还可以通过Descendant("path/to/child")方法返回子孙元素：

TagUnion descendant = grp.Descendant("path/to/child")

不过这种.Descendant()方式有两个局限：

  1. 这里的路径用“/”进行分隔，所以要求路径中的每个片段内都不得含有/。如果必须要在名称内使用/，只能使用[]逐层索引的方式获取。
  2. 这里返回的类型是TagUnion，它代表 TagUnion.TagUnit | TagUnion.TagUnit | TagUnion.TagGrp的一种。可以通过.Value属性得到内部的ITag、ITagCbnt、ITagGrp。

为了避免.Descendant()仍需要使用手动类型转化的麻烦，我们提供了三个扩展方法：

ITag tag0 = grp.SelectTag("path/to/child");       // 直接拿到 ITag
ITagCbnt cbnt0 = grp.SelectCbnt("path/to/child"); // 直接拿到 ITagCbnt
ITagGrp grp0 = grp.SelectGrp("path/to/child");    // 直接拿到 ITag

示例：

_grp = tags.SelectGrp("IoBox");

_ledRed = _grp.SelectTag("输出/红灯");
_ledReset = _grp.SelectTag("输出/复位_提示灯");
_btnReset = _grp.SelectTag("输入/复位_执行键");

要从特定的ITag取值，可以直接读取.Value属性——不过为了通用，.Value返回的是object类型。我们可以使用.GetTagValue<TValue>()方法避免手动类型转化：

var resetBnt = _btnReset.GetTagValue<bool>();
var resetLed = this._ledReset.GetTagValue<bool>();

想要写入值，可以直接对属性赋值：

_ledRed.Value = true;

注意: 本类库不会自动尝试任何隐式自动转化，哪怕是你看起来觉得没什么问题的隐式类型转换！这意味着类库使用者不得混写类型，比如不可以把short写给int，也不要把int写给float！

逻辑(ILogicet)

一个项目如果只能对点位进行读写，那大概是只一个无聊的采集应用。

Itminus.Tags 允许你加入自己的业务。在这里，业务逻辑被抽象成简单的ILogicet接口：

public interface ILogicet
{
    /// <summary>
    /// 运行顺序
    /// </summary>
    int Order { get; }

    /// <summary>
    /// 通道
    /// </summary>
    IReadOnlyList<ITagChannel> Channels { get; }

    /// <summary>
    /// 测点
    /// </summary>
    ITagGrp Tags { get; }

    /// <summary>
    /// 启用？
    /// </summary>
    bool Enabled { get; }

    /// <summary>
    /// 是否能匹配入口？返回true表示应该处理当前entry，否则应该跳过处理
    /// </summary>
    /// <param name="entry"></param>
    /// <returns></returns>
    bool MatchEntry(ITagGrp entry);

    /// <summary>
    /// 处理
    /// </summary>
    /// <returns></returns>
    Task ProcessAsync(ITagGrp entry, ITagChannel? thisChannel);
}

你可以在你的程序中以代码的方式注册你的业务逻辑实现，也可以在xml中以dll的方式手动注册。

针对每个入口(TagGrp)，都有一个ITagGrpRunner执行 [扫描->处理 ->输出] loop:

  • 扫描: 从底层读取
  • 处理: 依次执行各Logicet的ProcessAsync(entry, thisChannel)
  • 输出: 向底层刷写

这种机制十分类似于PLC的工作原理。你可能觉得这种机制过于死板、浪费性能、不够酷炫，但是工程项目不是个人花活，想想波音飞机的飞控程序吧，它也以在类似的机制工作着。

我们也提供.Watch()的事件机制，不过这种事件式的方式并不会改变主循环。作为简介，我们不打算在这里过多阐述这种模式。

项目的构造和运行

通常，我们使用ITagsProjectFactory构建项目，然后运行:

// var dir =  // 指定项目运行的目录

// 构造测点项目
var factory = sp.GetRequiredService<ITagsProjectFactory>();
ITagsProject proj = factory.Create(dir, rootXml);   

// ... 可选：注册自己的逻辑组件，定制事件钩子等

// 执行 [扫描 -> 处理 -> 输出] 循环
await proj.RunAsync(ct);

如之前我们在起步中所说，这里目录"dir"下通常会有一份测点索引文件(index.xml)。这个索引同样可以拆成三个部分：通道+测点+逻辑，我们将在下一章详细展开如何配置和构建一个项目。

Chapter 3 配置和构建

配置

用代码的方式构建项目的好处是强类型支持和优秀的智能提示，但是纯代码的方式既不够简洁，也不够灵活。我们可以把测点项目中的纯静态结构用一个配置文件加以描述：

<?xml version="1.0" encoding="utf-8"?>
<root>
    <!-- 可以有1~N个通道 -->
	<Channel name="S7-1" driver="S7">
		<IpAddr>192.168.10.2</IpAddr>
		<Rack>0</Rack>
		<Slot>1</Slot>
	</Channel>
	<Channel name="S7-2" driver="S7">
		<IpAddr>192.168.10.2</IpAddr>
		<Rack>0</Rack>
		<Slot>1</Slot>
	</Channel>

    <!-- 可以有1~N个测点组作为入口 -->
	<TagGrp name="信号区-1#" isEntry="true" isEnabled="true" channel="S7-1" scanInterval="10">
		<TagCbnt name="PLC" address="DB10.16" access="RO">
			<Tag name="心跳" address="DB10.16.0" type="BIT"></Tag>
			<Tag name="复位" address="DB10.16.1" type="BIT"></Tag>
			<Tag name="故障" address="DB10.18" type="INT16" endian="BigEndian"></Tag>
            <!--  ...more... -->
		</TagCbnt>
		<TagCbnt name="IPC" address="DB10.0" access="RW">
			<Tag name="心跳" address="DB10.0.0" type="BIT"></Tag>
			<Tag name="复位" address="DB10.0.1" type="BIT"></Tag>
			<Tag name="自动" address="DB10.0.2" type="BIT"></Tag>
			<Tag name="结束确认" address="DB10.0.3" type="BIT"></Tag>
            <!--  ...more... -->
		</TagCbnt>
	</TagGrp>

	<TagGrp name="信号区-2#" isEntry="true" isEnabled="true" channel="S7-1" scanInterval="10">
        <!--  ...more... -->
	</TagGrp>


    <!-- 可以有0~N个逻辑组件 -->
	<Logicet>Samples.Plugin1/Samples.Plugin1.dll</Logicet>
	<Logicet>Samples.Plugin2/Samples.Plugin2.dll</Logicet>
	<!--  ...more... -->
</root>

这里的有些基本规则：

  • 整个<root/>可以分成三大部分：
    • <Channel/>(1~N个)：描述了项目中用到的通道。
    • <TagGrp/>(1~N个): 描述了项目中用到的测点树。
    • <Logicet/>(0~N个): 描述了项目中用到的业务逻辑(插件)。
  • <Channel/>：
    • name属性：表示通道的名称，会被测点(以及TagCbnt，TagGrp)所引用。
    • driver属性：表示该通道使用的底层驱动，e.g. S7、COM等。每种驱动的通信方式并不相同。
  • <TagGrp>：
    • 通常顶层的<TagGrp/>会标注isEntry来表明这是一个入口。一个项目可以有多个入口，这些入口会被并行执行。
    • <TagGrp/> 表示一个逻辑分组，其下可以套多层，最终会嵌<TagCbnt/>或者<Tag/>。但是有一些底层驱动不支持批量读写，这时候<TagCbnt/>就是非法的；而另一些底层驱动可能并没有针对单个测点的读写提供支持，这时候直接嵌套<Tag/>也是非法的。总之，使用 <TagCbnt/>还是<Tag/>，要查看通道测点的驱动。
    • 测点的地址是和通道驱动密切相关的，每种驱动的地址都不相同。比如在S7下DB100.10.0代表的是DB100的第10个字节的第0个比特，特别的，可以用$$10.0代表使用和父级一致的Block；而对于OpcUA，地址则可能是ns=4;s=|var|CODESYS Control Win V3 x64.Application.PLC_PRG.var3。
    • 测点的访问方式有RO| WO | RW | R1W，对应的是 只读 、只写、读写、只在开始时读1次，后续只写，默认是RW。
    • 测点的通道检索是冒泡的。即如果本层节点未指定channel特性，则会向上冒泡检索，直至在祖先节点上检索到非空的通道(或者最终都没有检索到，会给出null)
  • <Logicet/>: 如果你偏好使用代码的方式注册逻辑，可以不用这里的动态插件机制。

构建

然后，我们可以使用这个配置文件来构建ITagsProject：

var dir = Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData); 

// 或者，如果不指定root，将会自动加载`dir/`下的`index.xml`文件
var root = XDocument.Load(xmlFileName).Root;
var proj = factory.Create(dir!, root);  

// ... 可选：可以手动注册 Logicets 、可以手动监听事件

// 运行测点项目
await proj.RunAsync(ct);

Chapter 4 逻辑组件

逻辑组件是我们编写业务代码的地方。

internal class HeartBeatLogicet : LogicetBase
{
    public HeartBeatLogicet(IReadOnlyList<ITagChannel> channels, ITagGrp tags, int robot) 
        : base(channels, tags)
    {
        this.Robot = robot;
        this._heartbeatReq = this.Tags.SelectTag("PLC/信号区-1#/心跳请求");
        this._heartbeatAck = this.Tags.SelectTag("MST/信号区-1#/心跳响应");
    }

    public override int Order => 100;

    public int Robot { get; }

    private ITag _heartbeatReq;
    private ITag _heartbeatAck;

    public override bool MatchEntry(ITagGrp entry) => entry.Name == "信号交互";

    public override Task ProcessAsync(ITagGrp entry, ITagChannel? thisChannel)
    {
        this._heartbeatAck.Value = this._heartbeatReq.GetTagValue<bool>();
        return Task.CompletedTask;
    }
}

组件编写

逻辑组件并不一定要继承自LogicetBase，只需要实现ILogiect即可。LogicetBase本身只是一层非常薄的抽象类，避免让你写规定了的无聊属性。

  • MatchEntry(ITagGrp entry): 用来表示本逻辑组件能否在指定入口 Loop 中执行。
  • Order: (如果能被执行的话)本逻辑组件在的执行顺序。序号越小，越会被优先执行。
  • ProcessAsync(ITagGrp entry, ITagChannel? thisChannel): 具体的业务逻辑。

逻辑组件的实例一旦被创建，在整个ITagsProject的生命期内都会被重用。

注册

逻辑组件需要注册到测点项目中，才能被执行。

手动注册

proj.Logicets.Add(new HeartBeatLogicet(proj.Channels, proj.Tags, 1));

自动注册

<root>
    <!-- ... -->

    <!-- 动态指定程序集 -->
    <Logicet>Samples.Plugin1/Samples.Plugin1.dll</Logicet>
    <Logicet>Samples.Plugin2/Samples.Plugin2.dll</Logicet>
</root>

注意上面的逻辑组件，我们可以定义0~n个逻辑组件(<Logicet/>)，只需指定*.dll文件路径，即可自动完成动态加载程序集、自动注册其中的ILogicet的公开(public)且具体(Concrete)的实现。

逻辑组件的动态管理

Itminus.Tags会在 ITagsProject 初始化时，动态加载相关程序集、并构造ILogicet实例；最后在Dispose()时，自动卸载不再需要的相关程序集。

开发者可以注册自己的 ILogicetCreator，来控制如何构造相关Logicet实例，如果未提供这样的自定义实现，Itminus.Tags会回退到默认实现。

组件的运行

还记得我们之前说过的运行机制吗？

针对每个入口(TagGrp)，都有一个ITagGrpRunner执行 [扫描->处理 ->输出] loop:

  • 扫描: 从底层读取
  • 处理: 依次执行各Logicet的ProcessAsync(entry, thisChannel)
  • 输出: 向底层刷写

这里我有意模仿了PLC的工作原理，请注意这是一个重大抉择: 我们有意放弃了(也许存在的)性能提升和花活，来换取所有的逻辑组件都是串行轮询的。

注意：由于所有逻辑组件都是规定的顺序串行轮询的，所以，在任何时候，都应该让ProcessAsync(entry, thisChannel)这个方法尽早结束，否则，串行约束会让后续的逻辑组件长时间得不到执行。

Chapter 5 练习1

本文示例源码地址是: Itminus.Tags.WPFDemo ，位于分支 eg-1 中

我们打算新建一个简单的WPF项目，该项目涉及PLC两个信号：

  • 心跳请求，位于 DB200.100.0， 固定以0.5Hz的频率闪烁。
  • 心跳响应，位于 DB201.100.0， 跟随心跳请求这个信号跳动

创建完一个模板项目后，请添加如下依赖：

  • Itminus.Tags.Core
  • Itminus.Tags
  • Itminus.Tags.S7
  • Itminus.Tags.RxExtensions

以及日志相关依赖：

  • Serilog
  • Serilog.AspNetCore
  • Serilog.Sinks.Console
  • Serilog.Sinks.File

备注：暂时还有一个间接依赖 FSharp.Core(>= 8.0.100)，需要我们手动安装。我们计划在将来调整这个nuget依赖关系，让用户不再需要手动安装。

项目配置和构建

让我们建立一个xml文件来表示通道和这两个测点:

<?xml version="1.0" encoding="utf-8"?>
<root>
	<Channel name="S7-1" driver="S7" >
		<IpAddr>172.16.10.20</IpAddr>
		<Rack>0</Rack>
		<Slot>1</Slot>
	</Channel>

	<TagGrp name="IoBox" isEntry="true" isEnabled="true" channel="S7-1" scanInterval="0">
		<TagGrp name="通用状态">
			<TagCbnt name="PLC" address="DB200.100">
				<Tag name="心跳请求" address="$$100.0" type="BIT" access="RO" />
			</TagCbnt>
			<TagCbnt name="MST" address="DB201.100" access="R1W">
				<Tag name="心跳响应" address="$$100.0" type="BIT" />
			</TagCbnt>
		</TagGrp>
	</TagGrp>
</root>

然后再添加一个心跳随动逻辑组件：

public class HeartBeatLogicet : LogicetBase
{
    private readonly ITag heartReq;
    private readonly ITag heartAck;
    private readonly ILogger<HeartBeatLogicet> _logger;

    public HeartBeatLogicet(IReadOnlyList<ITagChannel> channels, ITagGrp tags, ILogger<HeartBeatLogicet> logger) : base(channels, tags)
    {
        var grp = tags.SelectGrp("IoBox/通用状态");

        heartReq = grp.SelectTag("PLC/心跳请求");
        heartAck = grp.SelectTag("MST/心跳响应");
        this._logger = logger;
    }

    public override int Order => 2;

    public override bool MatchEntry(ITagGrp entry) => entry.Name == "IoBox";

    public override Task ProcessAsync(ITagGrp entry, ITagChannel? thisChannel)
    {
        var now = Stopwatch.GetTimestamp();
        heartAck.Value = heartReq.Value;
        return Task.CompletedTask;
    }
}

项目启停控制

测点项目的启停并不是一个简单的过程，在启动时可能需要构造很多东西，而在停止时往往需要释放一些资源。让我们创建一个TagsProjectCtrl来控制ITagsProject的启停。这里TagsProjectCtrl提供了两个核心方法：

  • 在运行StartPollAsync(dir, hook)的时候，它会创建ITagsProject实例，并调用hook函数对所创建的项目实例进行操作，最后运行该项目实例。
  • 在运行StopAsync()的时候，它会尝试断开底层设备通道，并释放之前创建的项目实例。
public class TagsProjectCtrl
{
    private readonly IServiceScopeFactory _ssf;
    private readonly ILogger<TagsProjectCtrl> _logger;

    public TagsProjectCtrl(IServiceScopeFactory ssf,ILogger<TagsProjectCtrl> logger) 
    {
        this._ssf = ssf;
        this._logger = logger;
    }

    public ITagsProject? Project { get; private set; }

    CancellationTokenSource? _cts;


    public async Task StartPollAsync(string? dir, Func<ITagsProject, CancellationToken, Task> hook)
    {
        if(this.Project != null)
        {
            throw new Exception("当前测点项目已经启动！");
        }

        using var scope = this._ssf.CreateScope();
        var sp = scope.ServiceProvider;


        try
        {
            this._cts = new CancellationTokenSource();
            this.Project = sp.MakeProject(dir);
            var ct = _cts.Token;
            await hook(this.Project, ct);
            this.StartedOrStopped?.Invoke(this, new TagsProjectEventArgs(true, this.Project));
            await this.Project.RunAsync(ct);
        }
        catch(Exception ex)
        {
            this._logger.LogError(ex, "TagsProjectCtrl.StartPollAsync error");
        }
        finally
        {
            if (this.Project is not null)
            {
                try
                {
                    this.Project.Dispose();
                }
                catch { }
                finally
                {
                    this.Project = null;
                }
            }

            this._cts = null;
        }
    }

    public async Task StopAsync()
    {
        // reset proj ctrl
        var oldchannels = this.Project?.Channels;
        try
        {
            if (this._cts != null)
            {
                this._cts.Cancel();
            }
            if (this.Project is not null)
            {
                try
                {
                    this.Project.Dispose();
                }
                catch { }
                finally
                {
                    this.Project = null;
                }
            }
        }
        catch
        {
            // ignore error when cancelling
        }

        try
        {
            // disconnect from each channel
            if (oldchannels is not null)
            {
                foreach (var ch in oldchannels)
                {
                    try
                    {
                        if (ch is not null)
                        {
                            await ch.DisconnectAsync(CancellationToken.None);
                        }
                    }
                    catch
                    {

                    }
                }
            }

            this.StartedOrStopped?.Invoke(this, new TagsProjectEventArgs(false, this.Project!));
        }
        catch
        {
            // ignore errors thrown by StartedOrStopped event handlers
        }
    }

    public event TagsProjectStartedOrStopped? StartedOrStopped;
}

这里的两个方法都会触发StartedOrStopped事件，该事件的定义是：

/// 项目启动或者停止事件
public delegate void TagsProjectStartedOrStopped(TagsProjectCtrl sender, TagsProjectEventArgs args);

public class TagsProjectEventArgs: EventArgs
{
    public TagsProjectEventArgs(bool isstarted, ITagsProject project)
    {
        this.IsStarted = isstarted;
        Project = project;
    }

    public ITagsProject Project { get; }

    public bool IsStarted { get; }
}

组合到一起

现在，我们可以把上面这些代码组合到一起了。

首先为App添加一个静态方法，以构造一个ServiceCollection，并注册相关服务:

public partial class App : Application
{
    private static ServiceCollection ConfigureServiceCollections()
    {
        var services = new ServiceCollection();
        services.AddWpfDemoTags();
        services.AddSerilog((sp, lc) => lc
            .ReadFrom.Services(sp)
            .WriteTo.Console()
            .WriteTo.File("logs/log.txt", outputTemplate: "{Timestamp:yyyy-MM-dd HH:mm:ss} [{Level:u3}] [{SourceContext}] {Message}{NewLine}{Exception}", rollingInterval: RollingInterval.Day)
            .Enrich.FromLogContext()
        );
        return services;
    }

其中AddWpfDemoTags是我们为这个示例项目编写的扩展方法:

public static IServiceCollection AddWpfDemoTags(this IServiceCollection services)
{
    services.AddLogging();
    services.AddTagsProjectServices(builder =>
    {
        builder.AddS7Support();
    });
    services.AddSingleton<TagsProjectCtrl>();
    return services;
}

然后为App指定Startup=Application_Startup，以在程序启动的时候就自动启动测点项目：

public partial class App : Application
{
    public IServiceProvider? Root { get; private set; }
    internal TagsProjectCtrl? Ctrl { get; private set; }

    private void Application_Startup(object sender, StartupEventArgs e)
    {
        ServiceCollection services = ConfigureServiceCollections();
        this.Root = services.BuildServiceProvider();
        this.Ctrl = this.Root.GetRequiredService<TagsProjectCtrl>();

        var th = new Thread(async () =>
        {
            var loggerFactory = this.Root.GetRequiredService<ILoggerFactory>();
            var logger = loggerFactory.CreateLogger<App>();

            var dir = Directory.GetParent(Assembly.GetExecutingAssembly().Location);
            await this.Ctrl.StartPollAsync(Path.Combine(dir!.FullName, "Tags"), (proj, ct) =>
            {
                // 注册心跳业务逻辑
                proj.Logicets.Add(new HeartBeatLogicet(
                    proj.Channels,
                    proj.Tags,
                    loggerFactory.CreateLogger<HeartBeatLogicet>()
                ));
                // ... 可以引入更多业务逻辑

                proj.TurnStarted += (grp, ch) => {
                    logger.LogInformation("Tags处理开始,grp={grpName}", grp.Name);
                    return Task.CompletedTask;
                };
                proj.TurnCrashed += (grp, ch, ex) => {
                    logger.LogError(ex, "Tags处理异常,grp={grp.Name}", grp.Name);
                    return Task.CompletedTask;
                };

                return Task.CompletedTask;
            });
        });
        th.IsBackground = true;
        th.Start();
    }

最后，我们还需要在程序退出的时候自动停止我们的测点项目：

    protected override async void OnExit(ExitEventArgs e)
    {
        using (var mre = new ManualResetEventSlim(false))
        {
            if(this.Ctrl is not null)
            {
                await this.Ctrl.StopAsync();
            }
            mre.Wait(TimeSpan.FromSeconds(10));
        }
        Environment.Exit(0);
    }

至此，运行这个项目，PLC上已经可以观测到你在跟随它的心跳随动了。

UI 监控

但光有后台逻辑还不够直观，让我们再花两分钟编写一个简单界面来显示这个两个信号的实时值：

MainWindow 
    <Grid>
        <Grid.ColumnDefinitions>
            <ColumnDefinition Width="Auto" />
            <ColumnDefinition />
        </Grid.ColumnDefinitions>

        <Grid.RowDefinitions>
            <RowDefinition/>
            <RowDefinition />
        </Grid.RowDefinitions>

        <Grid.Resources>
            <Style TargetType="Label" >
                <Setter Property="VerticalAlignment" Value="Center" />
            </Style>
            <Style TargetType="TextBlock" >
                <Setter Property="VerticalAlignment" Value="Center" />
            </Style>
        </Grid.Resources>

        <Label Grid.Row="0" Grid.Column="0">心跳请求</Label>
        <TextBlock  Name="txtReq" Grid.Row="0" Grid.Column="1" />

        <Label Grid.Row="1" Grid.Column="0">心跳响应</Label>
        <TextBlock Name="txtAck" Grid.Row="1" Grid.Column="1" />
    </Grid>

让我们用code-behind的方式把测点的值实时绑定到界面上：

public partial class MainWindow : Window, IDisposable
{
    private CompositeDisposable _disposables;

    public MainWindow()
    {
        InitializeComponent();
        this._disposables = new CompositeDisposable();

        var app = App.Current as App ?? throw new InvalidCastException("App.Current is not of type App");
        var tags = app.Ctrl!.Project!.Tags;
        SubscribeTags(tags);
    }

    private void SubscribeTags(ITagGrp tags)
    {
        var req = tags.SelectTag("IoBox/通用状态/PLC/心跳请求");
        var ack = tags.SelectTag("IoBox/通用状态/MST/心跳响应");


        req.Watch()
            .ObserveOn(DispatcherScheduler.Current)
            .Subscribe(evt =>
            {
                this.Dispatcher.Invoke(() =>
                {
                    this.txtReq.Text = evt.EventArgs.NewValue?.ToString();
                });
            })
            .DisposeWith(_disposables);

        ack.Watch()
            .ObserveOn(DispatcherScheduler.Current)
            .Subscribe(evt =>
            {
                this.Dispatcher.Invoke(() =>
                {
                    this.txtAck.Text = evt.EventArgs.NewValue?.ToString();
                });
            })
            .DisposeWith(_disposables);
    }

    public void Dispose()
    {
        _disposables.Dispose();
    }
}

小结

至此，你已经完成了一个简单的测点项目。不管看起来有多简单，我都强烈推荐你亲自动手一试，哪怕是克隆上面的源码仓库到本地，做些实验也很好。

Chapter 6 外部意图

最理想的情况当然是把所有的处理都放到串行轮询这个LOOP的中第二部分：

\[ \boxed{ \boxed{ \text{1. 读入数据} } \ \boxed{ \text{2. 处理逻辑} } \ \boxed{ \text{3. 刷写底层} } } \]

这样一来，各个ILogicet实例会以固定的顺序串行执行。

但事实上很多时候，我们需要从这个LOOP 的外部改变状态。在这种场景下，如果直接尝试写入，会带来非常大的不确定性——你甚至都不能确定对应的脏数据有没有被刷写到底层：也许在LOOP的某一轮结束后，你恰好手动更新了测点；但在同一时刻，下一轮正在从底层硬件读取缓存……这种并行不确定性会给程序员带来极大的不安。于是我引入了如下机制，以让来自外部的更改和自身的LOOP 串行执行：

\[ \boxed{ \boxed{ \text{0.执行意图} } \boxed{ \text{1. 读入数据} } \ \boxed{ \text{2. 处理逻辑} } \ \boxed{ \text{3. 刷写底层} } } \]

  1. \(\boxed{\text{执行意图}}\)：在LOOP每一轮开始前，先处理队列头部的前几个意图，直到队列已空或者达到单轮处理的最大意图数量。完成之后，如果缓存区已脏，则立刻刷写底层。
  2. \(\boxed{\text{读入数据}}\): 从底层硬件读取数据到缓存区。
  3. \(\boxed{\text{处理逻辑}}\): 串行执行逻辑组件。
  4. \(\boxed{\text{刷写底层}}\): 刷写变更的点位到底层硬件。

意图入队

例如：

// 意图入队: 返回true 表示入队成功；如果入队失败，await t 将直接抛出异常
var res = proj.WriteIntent(
    "IoBox", 
    (entry, ct) => {
        ack.Value = true;
        return ValueTask.CompletedTask;
    }, 
    out var t
);

// 等待意图被执行完成
await t;

注意，写意图本身只是将"意图"入队，out出的参数Task task并不会立刻被执行。意图一旦入队成功，则会在将来某个合适时间点被执行。你可以使用await task的手法，等待相关意图被执行完成，如果相关意图未能被正确执行，相关await会抛出异常。

意图执行和死锁陷阱

让我们重新审视这里的串行轮询机理:

\[ \boxed{ \boxed{ \text{0.执行意图} } \boxed{ \text{1. 读入数据} } \ \boxed{ \text{2. 处理逻辑} } \ \boxed{ \text{3. 刷写底层} } } \]

引入意图，是为了在原有的扫描循环之外，增加一种可以临时改变Tag的手段。但是如果意图执行等待得不恰当，有可能会引发死锁。

考虑这样一个情况：

  1. 某个Logicet中，捕获到相关业务A的请求信号(req)。该Logicet遂对外发布通知，要求处理。
  2. 外部处理器捕获到相同通知，执行完相关业务逻辑，最后写入意图，用以通知PLC开始另一个业务B。
  3. 等待上面都做完之后，该Logicet再向PLC反馈与业务A请求(req)相匹配的ack信号。

你能发现这里潜在的死锁吗？

重点看第3步：

  • 该Logicet在等待外部处理器执行完；
  • 而外部处理器向运行时写入了意图，然后干了什么？它在等待该意图被执行。

这里出现了环路等待的情况：Logicet在等外部处理器处理完；而外部处理器在等意图被执行完——但是，意图执行需要等到LOOP下一轮开始的时候才有可能被执行！

正确的做法是，让Logicet尽早结束：

if (req.GetTagValue<bool>() && !ack.GetTagValue<bool>() && Interlocked.CompareExchange(ref isBusy, 1, 0) == 0)
{

    // 如果外部处理器也需要写入意图并等待意图被执行完成，那么这里就不要等待，尽早结束
    _ = Task.Run(async () =>
    {
        // 这里的代码脱离了`Logicet`，所有的变更应当使用`WriteIntent`机制
        try
        {
            // 等待外部处理完成
            var evt = new ScrewResultEvt(block.CompleteBlock.DataBlock.ToResult());
            await _mediator.Publish(evt);

            // 写意图：确认
            var res = ctrl.Project!.WriteIntent(
                "IoBox", (entry, ct) =>
                {
                    ack.Value = true;
                    return ValueTask.CompletedTask;
                }, 
                out var t
            );
            await t;
        }
        catch (Exception ex)
        {
            // ... 错误处理
        }
        finally
        {
            var busy = Interlocked.CompareExchange(ref isBusy, 0, 1);
        }
    });
}
else if (!req.GetTagValue<bool>())
{
    ack.Value = false;
}

还记得我们在上一章逻辑组件中提到的注意点吗？

由于所有逻辑组件都是规定的顺序串行轮询的，所以，在任何时候，都应该让ProcessAsync(entry, thisChannel)这个方法尽早结束，否则，串行约束会让后续的逻辑组件长时间得不到执行。

Chapter 7 练习2

Chapter 8 Rx 扩展

Itminus.Tags.RxExtensions为ITag增加Rx支持。在开始之前，需要添加包引用：

dotnet add package Itminus.Tags.RxExtensions

目前我们主要提供了三个扩展方法：

  • tag.ObserveOnRead(): 观测从底层读取
  • tag.ObserveOnWritten(): 观测向底层写入
  • tag.Watch(): 观测测点发生了变化

观测从底层读取事件

本质上，ITag定义了一个事件OnTagRead：

public interface ITag 
{
    //...

    /// <summary>
    /// 在每次从底层读取后，触发事件
    /// </summary>
    event TagSyncEventHandler OnTagRead
}

ObserveOnRead()无非是把这个事件转成了Rx上可观测的事件流：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> ObserveOnRead(this ITag tag)
{
    var read = Observable.FromEventPattern<TagSyncEventHandler, ITag, TagSyncEventArgs>(
        h => tag.OnTagRead += h,
        h => tag.OnTagRead -= h
        );
    return read;
}

观测向底层写入事件

类似地，ITag也定义了一个事件OnTagWritten：

public interface ITag 
{
    //...

    /// <summary>
    /// 在每次向底层写入后，触发事件
    /// </summary>
    public event TagSyncEventHandler OnTagWritten;
}

ObserveOnWritten()也无非是把这个事件转成了Rx上可观测的事件流：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> ObserveOnWritten(this ITag tag)
{
    var written = Observable.FromEventPattern<TagSyncEventHandler, ITag, TagSyncEventArgs>(
        h => tag.OnTagWritten += h,
        h => tag.OnTagWritten -= h
        );
    return written;
}

观测测点改变事件

不管是从底层读取，还是向底层写入，都可以视作“内存”和“底层”之间发生了“同步”。tag.Watch()就是组合了这两种事件的一个便利方法：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> Watch(this ITag tag, bool startWithCurrent=true)
{
    var read = tag.ObserveOnRead();
    var written = tag.ObserveOnWritten();
    var obs = read.Merge(written);
    if(startWithCurrent)
    {
        var ev = new EventPattern<ITag, TagSyncEventArgs>(
            tag,
            new TagSyncEventArgs(tag.Value, tag.Timestamp, TagSyncEventArgs.Kinds.None)
            );
        return obs.StartWith(ev);
    }
    else
    {
        return obs;
    }
}

这里的默认值true表示需要把当时的值作为初始事件发给下游监听者。有些测点长时间不读写，而监听又通常又发生比较晚，指定true作为参数，可以有效避免这种长时间观测不到值的情况。

Chapter 9 练习3

Chapter 10 访问者模式

我们之前在《基本概念》中讲到：

TagUnion = TagGrp | TagCbnt | TagUnit

由于这是一个非常稳定的数据结构，所以我们很自然地想到应该为它提供一些基础设施以支持访问者模式。

访问者

访问者模式的核心理念是，将行为与相对稳定的数据结构相剥离，并将新的行为抽象成对数据结构进行访问的新类。 在这里，TagUnion就是相对稳定的数据结构——它要么是TagUnit，要么是TagCbnt，要么是TagGroup。

为了支持访问者模式，TagUnion应该可以接受一个访问者对自身进行访问。但是具体怎么访问，我们留待子类实现——因为只有子类才能知道自己是什么类型的节点：

public abstract record TagUnion
{
    /// 接受一个访问者对自己进行"访问"
    public abstract void Accept(ITagUnionVisitor visitor);
}

这里ITagUnionVisitor是一个接口抽象，规定了每种访问者应该支持访问 TagGrp、TagCbnt、和TagUnit：

public interface ITagUnionVisitor
{
    /// 访问 TagGrp 节点
    void Visit(TagUnion.TagGrp grp);

    /// 访问 TagCbnt 节点
    void Visit(TagUnion.TagCbnt cbnt);

    /// 访问 TagUnit 节点
    void Visit(TagUnion.TagUnit tag);
}

然后我们为每种数据节点提供实现——这里就是委托给访问者的访问方法，由于有了自身的类型信息，编译器可以清晰地知道它调用的是哪个重载方法：

public record TagUnit(ITag Value) : TagUnion 
{
    public override void Accept(ITagUnionVisitor visitor) 
    {
        visitor.Visit(this);  // 调用的是 Visit(TagUnion.TagUnit)方法
    }
};


public record TagCbnt(ITagCbnt Value) : TagUnion
{
    public override void Accept(ITagUnionVisitor visitor)
    {
        visitor.Visit(this); // 调用的是 Visit(TagUnion.Cbnt)方法
    }
}

public record TagGrp(ITagGrp Value): TagUnion
{
    public override void Accept(ITagUnionVisitor visitor)
    {
        visitor.Visit(this); // 调用的是 Visit(TagUnion.TagGrp)方法
    }
}

遍历器

让我们用访问者模式实现一个最简单的节点遍历器：

public class TagUnionTraverser : ITagUnionVisitor
{
    private Action<ITagGrp>? _procGrp;
    private Action<ITagCbnt>? _procCbnt;
    private Action<ITag>? _procUnit;

    /// <summary>
    /// c'tor
    /// </summary>
    /// <param name="procGrp">处理 ITagGrp节点本身，不处理子节点</param>
    /// <param name="procCbnt">处理 ITagCbnt 节点本身，不处理子节点</param>
    /// <param name="procTag">处理 ITag 节点本身</param>
    public TagUnionTraverser(
        Action<ITagGrp>? procGrp, 
        Action<ITagCbnt>? procCbnt, 
        Action<ITag>? procTag
    ){
        this._procGrp = procGrp;
        this._procCbnt = procCbnt;
        this._procUnit = procTag;
    }

    /// <inheritdoc/>
    public void Visit(TagUnion.TagGrp grp)
    {
        this._procGrp?.Invoke(grp.Value);
        foreach (var kvp in grp.Value.Children)
        {
            var child = kvp.Value;
            child.Accept(this);
        }
    }

    /// <inheritdoc/>
    public void Visit(TagUnion.TagCbnt cbnt)
    {
        this._procCbnt?.Invoke(cbnt.Value);

        foreach (var kvp in cbnt.Value.Children)
        {
            var key = kvp.Key;
            var child = cbnt[key];
            child.Accept(this);
        }
    }

    /// <inheritdoc/>
    public void Visit(TagUnion.TagUnit tag)
    {
        this._procUnit?.Invoke(tag.Value);
    }

}

于是我们可以这样使用这个遍历器：

var data = new TagUnion.TagGrp(grp);

// 构造遍历器对数据进行访问来统计 Tag 节点
var tagCount = 0;
var traverser = new TagUnionTraverser(
    null,                  // 什么都不做
    null,                  // 什么都不做
    tag => tagCount + 1    // 递增计数器
);

// 接受访问者
data.Accept(traverser);
Console.WriteLine($"tagCount = {tagCount}");