كيفية القيام بالمعالجة غير المتزامنة "المقسمة إلى صفحات" لنتيجة تحديد EF

Question

أنا أكتب شيئًا يقوم بتحميل السجلات من خادم SQL إلى قائمة انتظار زرقاء.الأمر هو أن عدد العناصر في النتيجة المحددة قد يكون كبيرًا جدًا، لذلك أرغب في بدء عملية وضع قائمة الانتظار بينما لا يزال يتم استرداد البيانات.

أحاول الاستفادة من EF6 (غير المتزامن)، وجميع أساليب المزامنة و TPL للتسلسل الموازي.لذلك أنا أملك:

        // This defines queue that Generator will publsh to and 
        // QueueManager wil read from. More info:
        // http://msdn.microsoft.com/en-us/library/hh228601(v=vs.110).aspx
        var queue = new BufferBlock<ProcessQueueItem>();

        // Configure queue listener first
        var result = this.ReceiveAndEnqueue(queue);

        // Start generation process
        var tasks = generator.Generate(batchId);

يعتبر ReceiveAndEnqueue بسيطًا:

    private async Task ReceiveAndEnqueue(ISourceBlock<ProcessQueueItem> queue)
    {
        while (await queue.OutputAvailableAsync())
        {
            var processQueueItem = await queue.ReceiveAsync();
            await this.queueManager.Enqueue(processQueueItem);
            this.tasksEnqueued++;
        }
    }

توقيع المولد () هو كما يلي:

public void Generate(Guid someId, ITargetBlock<ProcessQueueItem> target)

الذي يستدعي طريقة SendAsync () على الهدف لوضع عناصر جديدة.ما أفعله الآن هو تقسيم العدد الإجمالي للنتائج إلى "دفعات"، وتحميلها، وإرسالها بشكل غير متزامن، حتى يتم الانتهاء من كل شيء:

    public void Generate(Guid batchId, ITargetBlock<ProcessQueueItem> target)
    {
        var accountPromise = this.AccountStatusRepository.GetAccountsByBatchId(batchId.ToString());
        accountPromise.Wait();
        var accounts = accountPromise.Result;

        // Batch configuration
        var itemCount = accounts.Count();
        var numBatches = (int)Math.Ceiling((double)itemCount / this.batchSize);
        Debug.WriteLine("Found {0} items what will be put in {1} batches of {2}", itemCount, numBatches, this.batchSize); 


        for (int i = 0; i < numBatches; i++)
        {
            var itemsToTake = Math.Min(this.batchSize, itemCount - currentIndex);
            Debug.WriteLine("Running batch - skip {0} and take {1}", currentIndex, itemsToTake);

            // Take a subset of the items and place them onto the queue
            var batch = accounts.Skip(currentIndex).Take(itemsToTake);

            // Generate a list of tasks to enqueue the items
            var taskList = new List<Task>(itemsToTake);
            taskList.AddRange(batch.Select(account => target.SendAsync(account.AsProcessQueueItem(batchId))));

            // Return the control when all tasks have been enqueued
            Task.WaitAll(taskList.ToArray());

            currentIndex = currentIndex + this.batchSize;
        } 

ومع ذلك، فقد لاحظ زميلي أن هذا يعمل - "ألا يمكننا أن نجعل الواجهة أكثر بساطة، ونترك Generate() ونجعل الواجهة كما يلي:

public Task<IEnumerable<ProcessQueueItem> Generate(Guid someId)

أكثر نظافة، ولا تعتمد طريقة الإنشاء على مكتبة TPL.أنا أوافقك تماماً، أنا فقط أخشى أنه إذا فعلت ذلك، فسوف أضطر إلى الاتصال

var result = Generate().Wait().Result;

في مرحلة ما، قبل ترتيب كافة العناصر.هذا سيجعلني أنتظر حتى يتم تحميل جميع الأشياء وتكون في الذاكرة.

إذن ما يأتي سؤالي هو:كيف يمكنني البدء في استخدام نتائج استعلام EF بمجرد "تقطرها" من التحديد؟كما لو أن EF ستحقق "عائدًا" على النتائج إذا لاحظت انجرافي.

يحررأعتقد أنني ارتكبت خطأ في التفكير.تقوم EF بتحميل العناصر ببطء بشكل افتراضي.لذلك يمكنني فقط إرجاع جميع النتائج كـ IQueryable<> ولكن هذا لا يعني أنه تم تحميلها بالفعل من قاعدة البيانات.سأكررها بعد ذلك وأضعها في قائمة الانتظار.

تحرير 2لا، هذا لا يعمل، لأنني بحاجة إلى تحويل الكائن من قاعدة البيانات في طريقة Generate()...

Answer 1

حسنًا، هذا ما انتهيت إليه:

    public IEnumerable<ProcessQueueItem> Generate(Guid batchId)
    {
        var accounts = this.AccountStatusRepository.GetAccountsByBatchId(batchId.ToString());

        foreach (var accountStatuse in accounts)
        {
            yield return accountStatuse.AsProcessQueueItem(batchId);
        }
    }

يقوم المستودع بإرجاع IEnumerable لبعض DataContext.Stuff.Where(...).يستخدم المولد طريقة الامتداد لتحويل الكيان إلى نموذج المجال (ProcessQueueItem) والذي يتم إرساله على الفور عن طريق العائد إلى المتصل بالأسلوب، والذي سيبدأ في استدعاء QueueManager لبدء الانتظار.