如何在没有代理的情况下将消息传递给消费者

Question

制作人 不同的线程有自己的 SynchronizedBlockingQueue。 每个生产者都将消息放入自己的队列中。

消费者

不同的线程将从任何一个队列中获取消息并启动进程。

现在为了沟通生产者和消费者，我们需要经纪人。这可能是瓶颈。消费者是否有任何其他方式从任何生产者那里获取一条消息并启动流程。

Answer 1

由于您没有指定语言，我想我会提供一个使用 Ada 编程语言的通用示例。此示例让消费者简单地打印来自生产者的消息，但它提供了您描述的生产者-消费者架构。

with Ada.Task_Identification; use Ada.Task_Identification;

package Multiple_Producer is
   type Producer_Message is private;

   protected type Buffer is
      entry Set_Message (Item : in Producer_Message);
      entry Get_Message (Item : out Producer_Message);
   private
      Msg    : Producer_Message;
      Is_New : Boolean := False;
   end Buffer;
   type Buf_Alias is access all Buffer;
   type Buf_Array is array (Positive range <>) of aliased Buffer;
   type Buf_Access is access all Buf_Array;

   task type Producer is
      entry Set_Buffer (Item : Buf_Alias);
      entry Stop;
   end Producer;


   task Consumer is
      entry Set_Buffers (Item : Buf_Access);
      entry Stop;
   end Consumer;

private
   type Producer_Message is record
      the_Task : Task_Id;
      Value    : Integer;
   end record;
end Multiple_Producer;

with Ada.Text_IO; use Ada.Text_IO;

package body Multiple_Producer is

   --------------
   -- Producer --
   --------------

   task body Producer is
      Message : Producer_Message := (Current_Task, 0);
      The_Buf : Buf_Alias;
   begin
      accept Set_Buffer(Item : in Buf_Alias) do
         The_Buf := Item;
      end Set_Buffer;

      loop
         select
            accept Stop;
            exit;
         else
            delay 0.01;
            The_Buf.Set_Message(Message);
            Message.Value := Message.Value + 1;
         end select;
      end loop;
   end Producer;

   --------------
   -- Consumer --
   --------------

   task body Consumer is
      Message : Producer_Message;
      Buffers : Buf_Access;
   begin
      accept Set_Buffers(Item : Buf_Access) do
         Buffers := Item;
      end Set_Buffers;

      loop
         select
            accept Stop;
            exit;
         else
            -- Poll the buffers
            for I in Buffers'Range loop
               select
                  Buffers(I).Get_Message(Message);
                  Put_Line(Image(Message.The_Task) & ": " &
                             Integer'Image(Message.Value));
               or
                  delay 0.001;
               end select;
            end loop;
         end select;
      end loop;
   end Consumer;

   ------------
   -- Buffer --
   ------------

   protected body Buffer is

      -----------------
      -- Set_Message --
      -----------------

      entry Set_Message (Item : in Producer_Message) when not Is_New is
      begin
         Msg := Item;
         Is_New := True;
      end Set_Message;

      -----------------
      -- Get_Message --
      -----------------

      entry Get_Message (Item : out Producer_Message) when Is_New is
      begin
         Item := Msg;
         Is_New := False;
      end Get_Message;

   end Buffer;

end Multiple_Producer;

with Multiple_Producer; use Multiple_Producer;

procedure Main is
   subtype Producer_Range is Positive range 1..5;
   The_Producers : array(Producer_Range) of Producer;
   The_Buffers   : Buf_Access := new Buf_Array(Producer_Range); 
begin
   for I in Producer_Range loop
      The_Producers(I).Set_Buffer(The_Buffers(I)'Access);
   end loop;
   Consumer.Set_Buffers(The_Buffers);
   delay 4.0;
   for P of The_Producers loop
      P.Stop;
   end loop;
   Consumer.Stop;
end Main;