Как реализовать насос сообщений в потоке не-пользовательского интерфейса в .NET?

Question

Я читаю этот блог: Используйте темы правильно , и мне интересно:

Как реализовать сообщение ( Примечание. Я не имею в виду сообщение Windows здесь ), работающее в потоке без пользовательского интерфейса?

Я хочу, чтобы сообщение могло быть объектом или командой, скажем, Action<T> / Func<T> и т. Д. Нужно ли использовать отдельную очередь для разных типов сообщений? Скажем, одна очередь для объекта, одна очередь для Action<T> / Func<T>?

Учитывая, что тип сообщений различен, как его реализовать?

EDIT:

То, что я хочу завершить, - это модель «производитель / потребитель», очередь общего ресурса «производитель / потребитель» для связи, очередь может содержать не только объект для потребителя, но и некоторую команду « ». передано потребителю для исполнения.

Answer 1

Термин "насос сообщений" относится , в частности к потокам графического интерфейса в Windows (даже если они на самом деле не создают графический интерфейс пользователя).

Похоже, вы говорите о сообщении очереди , но даже очередь сообщений не будет принимать действий или функции , он будет принимать определенные виды сообщений .

Я, конечно, никогда не слышал о передаче Func<T> в очередь сообщений - что будет с результатом? Собираетесь ли вы хранить его где-нибудь? Отправить сообщение обратно звонящему? Это не кажется мне очень полезным.

Независимо от того, что вы пытаетесь сделать, звучит так, как будто бы лучше справиться со встроенным пулом потоков .NET. Это фактически предназначено для асинхронного выполнения кода , в отличие от очередей и обработки сообщений .

Answer 2

Пример элементарной очереди сообщений

Я хотел бы поделиться рудиментарной очередью сообщений, которую я однажды написал для своих собственных исследований. Это может помочь составить представление о возможной реализации очереди сообщений. Нет претензий на полноту, целостность или отсутствие ошибок, и, вероятно, в большинстве случаев есть лучшие решения, чем использование пользовательской очереди сообщений.

public void run()
{
    while (running)
    {
        mainLoopWaitHandle.WaitOne();

        EventHandlerFunction f = null;
        while (running)
        {
            f = popEvent();
            if (f == null) break;

            f();
        }
    }
}

private void pushEvent(EventHandlerFunction handlerFunction)
{
    lock (eventQueueLock)
    {
        int b = (queueInIndex + 1) & 255;
        if (b == queueOutIndex)
        {
            throw new Exception("Buffer overflow in event queue.");
        }

        eventQueue[queueInIndex] = handlerFunction;
        queueInIndex = b;

        mainLoopWaitHandle.Set();
    }
}


private EventHandlerFunction popEvent()
{
    EventHandlerFunction ret = null;

    lock(eventQueueLock)
    {
        int b = (queueOutIndex + 1) & 255;

        if (queueOutIndex == queueInIndex)
        {
            mainLoopWaitHandle.Reset();
            return null;
        }

        ret = eventQueue[queueOutIndex];
        eventQueue[queueOutIndex] = null;
        queueOutIndex = b;
    }

    return ret;
}

Основной поток начинает использовать очередь сообщений с запуска run(). run() - это метод блокировки, который не возвращается до тех пор, пока атрибут класса running не будет установлен в false. Это можно сделать с помощью метода invoker, описанного ниже.

Для вызова метода в главном потоке нужны два метода. Одна функция EventHandlerFunction (скажем, метод A()), которая фактически вызывается в основном потоке, и метод B(), который выполняется в потоке вызывающей стороны. Я вижу это аналогично функциям пользовательского интерфейса, где B() - это метод формы, а A() get's Invoke d из B().

B() вызывает A(), вызывая

pushEvent(A);

, в то время как pushEvent() и popEvent() - сохранение потока.

Цель метода B() - сохранить любые объекты или параметры в некоторой структуре данных для передачи данных, которая представляет параметры (или задачи) для метода A(). Эта структура может быть List<> с рабочими элементами, например. Метод A () и B () оба должны позаботиться о правильной блокировке этой структуры для безопасности потока.

Метод A() также должен учитывать, что буфер может заполниться полностью либо в очереди сообщений, либо в собственной структуре передачи данных) и должен заботиться о последствиях (например, сбросить вызов или заблокировать вызов, пока не освободится место). в стеке).

Надеюсь, это поможет. Взносы приветствуются.

Answer 3

Сделано это отдельным ответом для форматирования кода

Хорошо, так что после прочтения вашего обновления я думаю, что вы хотите то, что я описываю во «втором случае», вы просто хотите

Broadcast<T>("Foo") 

где T - делегат.

Тогда ваш потребитель будет делать

Subscribe<T>("Foo",HandlerMethod)

Таким образом, сценарий для производителя будет выглядеть следующим образом

internal static class MessagePump
    {

        public static void Subscribe<T>(String foo, Action<String> handlerMethod)
        {
            throw new NotImplementedException();
        }

        public static void BroadcastMessage<T>(String foo, Action<String> someAction)
        {
            throw new NotImplementedException();
        }
    }

    public class Producer
    {
        void SendMessage()
        {
            MessagePump.BroadcastMessage<Action<String>>("Foo", SomeAction);
        }

        void SomeAction(String param)
        {
            //Do Something
        }
    }


    public class Consumer
    {

        public Consumer()
        {
            MessagePump.Subscribe<Action<String>>("Foo", HandlerMethod);
        }

        void HandlerMethod(String param)
        {
            // Do Something
        }

    }

Это просто что-то не в моей голове, и это надуманный пример, так что возьмите это с крошкой соли. Это почти то, что я делаю в рамках курьера, который я разместил ранее. Вы можете погрузиться в этот код, чтобы получить более конкретный пример реализации.

Вам необходимо подумать о том, как вы будете управлять потребителями, как вы проверяете широковещательную рассылку и подписки, а также для своего конкретного случая, как вы собираетесь гарантировать, что делегат, которого вы передаете, вызывается правильно? Или тебя это волнует?

Помогает ли это?

Answer 4

Я немного запутался в том, какова ваша конечная цель, но кажется, что вы ищете две вещи.

Во-первых, в случае, когда вы хотите передать объекты (предположительно, вы думаете об очереди сообщений в целях межпотоковой связи?), Это звучит так, как будто вы хотите реализовать паб / подтип? Это довольно хорошо задокументировано, и есть много примеров в C #

Во втором случае, когда вы хотите передать делегата в качестве полезной нагрузки сообщения. Я предполагаю, что вы пытаетесь реализовать своего рода двустороннюю связь между издателем и подписчиком? Что-то вроде обратного вызова?

Это то, где я запутался. Что именно является вопросом / проблемой здесь? Вы можете реализовать одну очередь сообщений, которая понимает, как обрабатывать различные типы полезных сообщений. Что-то вроде BroadcastMessage, где T - это объект для вашего первого случая и делегат (Func / action) для второго случая.

У меня есть проект codeplex, представляющий собой простую реализацию очереди сообщений, которую я использую для определенных целей в приложениях MVC / MVVM. Не уверен, что это то, что вы ищете, но это может помочь вам прояснить свой вопрос дальше?

http://courier.codeplex.com/

Answer 5

Что ж, вам понадобится Thread.SetApartmentState (), чтобы переключить Thread в STA перед его запуском, а затем вызвать Application.Run () в функции потока, чтобы запустить цикл обработки сообщений.

Это не имеет ничего общего с делегатами Action и Func. Очередь сообщений - это внутренняя структура Windows, в которой хранятся сообщения мыши и клавиатуры. Это не подходит для хранения ваших собственных объектов. Не уверен, куда ты хочешь пойти с этим.

---

Обычный объект System.Collections.Generic.Queue <> может хранить любой тип объекта, включая объект класса, который представляет «команду». Сделайте это поточно-ориентированным с ReaderWriterLockSlim.