如何通过ID RX分组和油门对象

Question

我有同一类型的传入对象，但是如果对象属性Isthrottlable设置为false，则无论我不想限制ID，但如果Isthrottlable，则设置确实，我想每3秒通过ID防止对象。因此，如果具有相同ID的对象在50次以3秒为单位出现，我想发送最后一个对象的httpsend。

namespace BoatThrottle
{
    class MData
    {
        public int ID { get; set; }
        public bool IsThrottlable { get; set; }
        public string Description { get; set; }
    }
    class Program
    {
        static void Main(string[] args)
        {
            Random rand = new Random();

            while (true)
            {
                var data = GenerateRandomObj(rand);
                SendData(data);
                Task.Delay(rand.Next(100, 2000));
            }
        }

        static MData GenerateRandomObj(Random rand)
        {
            return new MData() { ID = rand.Next(1, 20), Description = "Notification....", IsThrottlable = (rand.Next(2) == 1) };
        }

        static void SendData(MData mData)
        {
            if (mData.IsThrottlable)
            {
                _doValues.OnNext(mData);
                var dd = ThrottledById(DoValues);

                var observable =
                   dd
                    .Throttle(TimeSpan.FromMilliseconds(3000.0))
                    .ObserveOn(Scheduler.ThreadPool.DisableOptimizations());

                _subscription =
                    observable
                    .ObserveOn(Scheduler.ThreadPool.DisableOptimizations())
                        .Subscribe(y =>
                        {
                            HTTPSend(y);
                        });

            }
            else
            {
                // MData object coming in IsThrottlable set to false always send this data NO throttling
                HTTPSend(mData);
            }

        }
        private static IDisposable? _subscription = null;

        public static IObservable<MData> ThrottledById(IObservable<MData> observable)
        {
            return observable.Buffer(TimeSpan.FromSeconds(3))
                .SelectMany(x =>
                    x.GroupBy(y => y.ID)
                    .Select(y => y.Last()));
        }

        private static readonly Subject<MData> _doValues = new Subject<MData>();

        public static IObservable<MData> DoValues { get { return _doValues; } }

        static void HTTPSend(MData mData)
        {
            Console.WriteLine("===============HTTP===>>  " + mData.ID + "  " + mData.Description + " " + mData.IsThrottlable);
        }
    }
}

编辑：

例如全部在3秒内收到

• mdata ID = 1，isthrottlable = false，description =“ notify”

• mdata id = 2，isthrottlable = true，description =“ notify1”

• mdata id = 2，isthrottlable = true，description =“ notify2”

• mdata id = 9，isthrottlable = false，description =“ notify2”

• mdata id = 2，isthrottlable = true，description =“ notify3”

• mdata id = 2，isthrottlable = true，description =“ notify4”

• mdata id = 3，isthrottlable = true，description =“ notify”

• mdata id = 4，isthrottlable = true，description =“ notify”

• mdata id = 5，isthrottlable = true，description =“ notify1”

• mdata id = 5，isthrottlable = true，description =“ notify2”

• mdata id = 8，isthrottlable = true，description =“ notify1”

• mdata id = 8，isthrottlable = true，description =“ notify2”

• mdata id = 8，isthrottlable = true，description =“ notify3”

• mdata id = 8，isthrottlable = true，description =“ notify4”

• mdata id = 8，isthrottlable = true，description =“ notify5”

• mdata id = 8，isthrottlable = true，description =“ notify6”

前3秒：

• mdata id = 1，isthrottlable = false = false，description =“ notify”
• mdata id = 9，isthrottlable = false ，descrivation =“ notify2”
• mdata id = 2，isthrottlable = true，description
• =“ notify4” mdata id = 3 ，isthrottlable = true
• = true，description =“ notify”
• mdata id = 4，isthrottlable = true = true，descria = 5，isthrottlable = true，description =“ notify2”
• mdata id = 8，isthrottlable = true，description =“ notify6”

Answer 1

我决定按照您的问题中发布的最终实现，但这应该是一个答案，并以最愚蠢的RX方式清理查询。

这是我的代码版本：

public MainWindow()
{
    InitializeComponent();

    Debug.Print("========================");

    _subscription =
        Observable
            .Generate(0, x => true, x => x + 1,
                x => new MData() { ID = Random.Shared.Next(1, 3), Description = "Notification....", IsThrottlable = Random.Shared.Next(2) == 1 },
                x => TimeSpan.FromMilliseconds(Random.Shared.Next(100, 2000)))
            .GroupBy(m => m.IsThrottlable)
            .SelectMany(g =>
                g.Key
                ? g.GroupBy(x => x.ID).SelectMany(g2 => g2.Throttle(TimeSpan.FromSeconds(3.0)))
                : g)
            .SelectMany(m => Observable.Start(() => HTTPSend(m)))
            .Subscribe();
}

最终.selectmany（m =＆gt; observable.start（（）=＆gt; httpsend（m）））可能需要将其写入.select（ m =＆gt; observable.start（（）=＆gt; httpsend（m））。合并（1）。

Answer 2

一种方法是通过isthrottlable属性对序列进行分组。这样，您将获得一个嵌套序列，其中包含两个子序列，一个序列包含可节流元素，另一个包含不可碰撞的元素。然后，您可以相应地转换两个子序中的每个子序列，最后使用selectmany操作员将嵌套序列缩回到一个扁平序列中，该序列包含两个转换的子序列发出的元素。

包含不可碰撞元素的子序列不需要转换，因此您可以按原样返回。

包含可节流元素的子序列需要通过id属性将其进一步分组，从而产生仅包含具有相同ID的可节流元素的更薄的子序列。这些是需要进行限制的序列：

IObservable<MData> throttled = source
    .GroupBy(x => x.IsThrottlable)
    .SelectMany(g1 =>
    {
        if (!g1.Key) return g1; // Not throttleable, return it as is.
        return g1
            .GroupBy(x => x.ID)
            .SelectMany(g2 => g2.Throttle(TimeSpan.FromSeconds(3)));
    });

最后，您将获得一个平坦的序列，其中包含可节流和不可侵入的项目，并且已经由ID插入了可节流项目。

selectmany运算符本质上是选择+MERGE运算符的组合。