- android - RelativeLayout 背景可绘制重叠内容
- android - 如何链接 cpufeatures lib 以获取 native android 库?
- java - OnItemClickListener 不起作用,但 OnLongItemClickListener 在自定义 ListView 中起作用
- java - Android 文件转字符串
重要:有关结果的描述和更多详细信息,请同时查看我的回答
我需要对通常被复制的一系列对象/事件进行分组和过滤,并使用 TimeSpan 间隔对它们进行缓冲。我尝试用一些大理石图更好地解释它:
X-X-X-X-X-Y-Y-Y-Z-Z-Z-Z-X-X-Y-Z-Z
会产生
X---Y---Z---X---Y---Z
其中 X、Y 和 Z 是不同的事件类型,'---' 表示间隔。此外,我还想通过一个关键属性来区分它在所有类型上都可用,因为它们有一个共同的基类:
X, Y, Z : A
并且A包含一个属性Key。使用符号 X.a 表示 X.Key = a,最终示例将是:
X.a-X.b-X.a-Y.b-Y.c-Z.a-Z.a-Z.c-Z.b-Z.c
会产生
X.a-X.b---Y.b-Y.c-Z.a-Z.c-Z.b
任何人都可以帮助我将所需的 Linq 运算符(可能是 DistinctUntilChanged 和 Buffer)放在一起来实现此行为吗?谢谢
更新 18.08.12:
应要求,我尽量给出更好的解释。我们有设备收集事件并将其发送到网络服务。这些设备有一个旧的逻辑(由于向后兼容,我们无法更改它)并且它们会不断发送事件,直到它们收到确认;确认后,他们发送队列中的下一个事件,依此类推。事件包含单元的网络地址和一些其他属性,用于区分每个设备队列中的事件。一个事件看起来像这样:
class Event
{
public string NetworkAddress { get; }
public string EventCode { get; }
public string AdditionalAttribute { get; }
}
目标是每 5 秒处理一次从所有设备接收到的可区分事件,将信息存储在数据库中(这就是我们不想分批处理的原因)并将确认发送到设备。让我们举一个只有两个设备和一些事件的例子:
Device 'a':
Event 1 (a1): NetworkAddress = '1', EventCode = A, AdditionalAttribute = 'x'
Event 2 (a2): NetworkAddress = '1', EventCode = A, AdditionalAttribute = 'y'
Event 3 (a3): NetworkAddress = '1', EventCode = B, AdditionalAttribute = 'x'
Device 'b':
Event 1 (b1): NetworkAddress = '2', EventCode = A, AdditionalAttribute = 'y'
Event 2 (b2): NetworkAddress = '2', EventCode = B, AdditionalAttribute = 'x'
Event 3 (b3): NetworkAddress = '2', EventCode = B, AdditionalAttribute = 'y'
Event 4 (b4): NetworkAddress = '2', EventCode = C, AdditionalAttribute = 'x'
Pn are the operations done by our server, explained later
可能的弹珠图(输入流+输出流):
Device 'a' : -[a1]-[a1]-[a1]----------------[a2]-[a2]-[a2]-[a3]-[a3]-[a3]-...
Device 'b' : ------[b1]-[b1]-[b2]-[b2]-[b2]------[b3]-[b3]-[b4]-[b4]-[b4]-...
Time : ------------[1s]-----------[2s]------------[3s]------------[4s]-
DB/acks (rx output) : ------------[P1]-----------[P2]------------[P3]------------[P4]-
P1: Server stores and acknowledges [a1] and [b1]
P2: " " " " [b2]
P3: " " " " [a2] and [b3]
P4: " " " " [a3] and [b4]
最后我认为它可能是基本运算符的简单组合,但我是 Rx 的新手并且我有点困惑,因为似乎有很多运算符(或运算符的组合)来获得相同的输出流。
更新 19.08.12:
请记住,此代码在服务器上运行,并且应该运行数天而不会出现内存泄漏...我不确定对象的行为。目前,对于每个事件,我都在服务上调用推送操作,该服务调用主题的 OnNext,我应该在其上构建查询(如果我对主题的使用没记错的话)。
更新 20.08.12:
当前实现,包括验证测试;这是我尝试过的,它似乎与@yamen 的建议相同
public interface IEventService
{
// Persists the events
void Add(IEnumerable<Event> events);
}
public class Event
{
public string Description { get; set; }
}
/// <summary>
/// Implements the logic to handle events.
/// </summary>
public class EventManager : IDisposable
{
private static readonly TimeSpan EventHandlingPeriod = TimeSpan.FromSeconds(5);
private readonly Subject<EventMessage> subject = new Subject<EventMessage>();
private readonly IDisposable subscription;
private readonly object locker = new object();
private readonly IEventService eventService;
/// <summary>
/// Initializes a new instance of the <see cref="EventManager"/> class.
/// </summary>
/// <param name="scheduler">The scheduler.</param>
public EventManager(IEventService eventService, IScheduler scheduler)
{
this.eventService = eventService;
this.subscription = this.CreateQuery(scheduler);
}
/// <summary>
/// Pushes the event.
/// </summary>
/// <param name="eventMessage">The event message.</param>
public void PushEvent(EventMessage eventMessage)
{
Contract.Requires(eventMessage != null);
this.subject.OnNext(eventMessage);
}
/// <summary>
/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
/// </summary>
/// <filterpriority>2</filterpriority>
public void Dispose()
{
this.Dispose(true);
}
private void Dispose(bool disposing)
{
if (disposing)
{
// Dispose unmanaged resources
}
this.subject.Dispose();
this.subscription.Dispose();
}
private IDisposable CreateQuery(IScheduler scheduler)
{
var buffered = this.subject
.DistinctUntilChanged(new EventComparer())
.Buffer(EventHandlingPeriod, scheduler);
var query = buffered
.Subscribe(this.HandleEvents);
return query;
}
private void HandleEvents(IList<EventMessage> eventMessages)
{
Contract.Requires(eventMessages != null);
var events = eventMessages.Select(this.SelectEvent);
this.eventService.Add(events);
}
private Event SelectEvent(EventMessage message)
{
return new Event { Description = "evaluated description" };
}
private class EventComparer : IEqualityComparer<EventMessage>
{
public bool Equals(EventMessage x, EventMessage y)
{
return x.NetworkAddress == y.NetworkAddress && x.EventCode == y.EventCode && x.Attribute == y.Attribute;
}
public int GetHashCode(EventMessage obj)
{
var s = string.Concat(obj.NetworkAddress + "_" + obj.EventCode + "_" + obj.Attribute);
return s.GetHashCode();
}
}
}
public class EventMessage
{
public string NetworkAddress { get; set; }
public byte EventCode { get; set; }
public byte Attribute { get; set; }
// Other properties
}
和测试:
public void PushEventTest()
{
const string Address1 = "A:2.1.1";
const string Address2 = "A:2.1.2";
var eventServiceMock = new Mock<IEventService>();
var scheduler = new TestScheduler();
var target = new EventManager(eventServiceMock.Object, scheduler);
var eventMessageA1 = new EventMessage { NetworkAddress = Address1, EventCode = 1, Attribute = 4 };
var eventMessageB1 = new EventMessage { NetworkAddress = Address2, EventCode = 1, Attribute = 5 };
var eventMessageA2 = new EventMessage { NetworkAddress = Address1, EventCode = 1, Attribute = 4 };
scheduler.Schedule(() => target.PushEvent(eventMessageA1));
scheduler.Schedule(TimeSpan.FromSeconds(1), () => target.PushEvent(eventMessageB1));
scheduler.Schedule(TimeSpan.FromSeconds(2), () => target.PushEvent(eventMessageA1));
scheduler.AdvanceTo(TimeSpan.FromSeconds(6).Ticks);
eventServiceMock.Verify(s => s.Add(It.Is<List<Event>>(list => list.Count == 2)), Times.Once());
scheduler.Schedule(TimeSpan.FromSeconds(3), () => target.PushEvent(eventMessageB1));
scheduler.AdvanceTo(TimeSpan.FromSeconds(11).Ticks);
eventServiceMock.Verify(s => s.Add(It.Is<List<Event>>(list => list.Count == 1)), Times.Once());
}
此外,我再次强调,该软件能够毫无问题地运行数天,处理数以千计的消息,这一点非常重要。明确地说:测试未通过当前实现。
最佳答案
我不确定这是否完全符合您的要求,但您可能会使用 group
显式地对元素进行分组。关键字,然后操纵各种IObservable
s 在重新组合之前分开。
例如如果我们有类定义,例如
class A
{
public char Key { get; set; }
}
class X : A { }
...
和一个Subject<A>
Subject<A> subject = new Subject<A>();
然后我们可以写
var buffered =
from a in subject
group a by new { Type = a.GetType(), Key = a.Key } into g
from buffer in g.Buffer(TimeSpan.FromMilliseconds(300))
where buffer.Any()
select new
{
Count = buffer.Count,
Type = buffer.First().GetType().Name,
Key = buffer.First().Key
};
buffered.Do(Console.WriteLine).Subscribe();
我们可以使用您提供的数据对此进行测试:
subject.OnNext(new X { Key = 'a' });
Thread.Sleep(100);
subject.OnNext(new X { Key = 'b' });
Thread.Sleep(100);
subject.OnNext(new X { Key = 'a' });
Thread.Sleep(100);
...
subject.OnCompleted();
获取您提供的输出:
{ Count = 2, Type = X, Key = a }
{ Count = 1, Type = X, Key = b }
{ Count = 1, Type = Y, Key = b }
{ Count = 1, Type = Y, Key = c }
{ Count = 2, Type = Z, Key = a }
{ Count = 2, Type = Z, Key = c }
{ Count = 1, Type = Z, Key = b }
关于c# - Rx 运算符到不同的序列,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/11916038/
Or 运算符 对两个表达式进行逻辑“或”运算。 result = expression1 Or expression2 参数 result 任意数值变量。 expression1 任意
Not 运算符 对表达式执行逻辑非运算。 result = Not expression 参数 result 任意数值变量。 expression 任意表达式。 说明 下表显示如何
Is 运算符 比较两个对象引用变量。 result = object1 Is object2 参数 result 任意数值变量。 object1 任意对象名。 object2 任意
\ 运算符 两个数相除并返回以整数形式表示的结果。 result = number1\number2 参数 result 任意数值变量。 number1 任意数值表达式。 numbe
And 运算符 对两个表达式进行逻辑“与”运算。 result = expression1 And expression2 参数 result 任意数值变量。 expression1
运算符(+) 计算两个数之和。 result = expression1 + expression2 参数 result 任意数值变量。 expression1 任意表达式。 exp
我对此感到困惑snippet : var n1 = 5-"4"; var n2 = 5+"4"; alert(n1); alert(n2); 我知道 n1 是 1。那是因为减号运算符会将字符串“4”转
我想我会得到 12,而不是 7。 w++,那么w就是4,也就是100,而w++, w 将是 8,1000;所以 w++|z++ 将是 100|1000 = 1100 将是 12。 我怎么了? int
Xor 运算符 对两个表达式进行逻辑“异或”运算。 result = expression1 Xor expression2 参数 result 任意数值变量。 expression1
Mod 运算符 两个数值相除并返回其余数。 result = number1 Mod number2 参数 result 任意数值变量。 number1 任意数值表达式。 numbe
Imp 运算符 对两个表达式进行逻辑蕴涵运算。 result = expression1 Imp expression2 参数 result 任意数值变量。 expression1 任
Eqv 运算符 执行两个表达式的逻辑等价运算。 result = expression1 Eqv expression2 参数 result 任意数值变量。 expression1 任
我有一个运算符重载的简单数学 vector 类。我想为我的运算符(operator)获取一些计时结果。我可以通过计时以下代码轻松计时我的 +=、-=、*= 和/=: Vector sum; for(s
我是用户定义比较运算符的新手。我正在读一本书,其中提到了以下示例: struct P { int x, y; bool operator、运算符<等),我们
在 SQL 的维基百科页面上,有一些关于 SQL 中 bool 逻辑的真值表。 [1] 维基百科页面似乎来源于 SQL:2003 标准。 等号运算符 (=) 的真值表与 SQL:2003 草案中的 I
我遇到了一个奇怪的 C++ 运算符。 http://www.terralib.org/html/v410/classoracle_1_1occi_1_1_number.html#a0f2780081f
我正在阅读关于 SO 和 answers 中的一个问题,它被提到为: If no unambiguous matching deallocation function can be found, pr
我偶然发现了这个解决方案,但我无法理解其中到底发生了什么。谁能解释一下! 据我了解,它试图通过计算一半的单元格然后将其加倍来计算 a*b 网格中的单元格数量。但是我无法理解递归调用。 请不要建议其他解
Go的基本类型 布尔类型bool 长度:1字节 取值:布尔类型的取值只能是true或者false,不能用数字来表示 整型 通用整型 int / uint(有符号 / 无符号,下面也类似) 长度:根据运
在本教程中,您将学习JavaScript中可用的不同运算符,以及在示例的帮助下如何使用它们。 什么是运算符? 在JavaScript中,运算符是一种特殊符号,用于对运算数(值和变量)执行操作。例如,
我是一名优秀的程序员,十分优秀!