Единственный производитель Единственный потребитель теперь мне нужно несколько потребителей

Question

У меня есть ситуация, когда я написал простую модель Producer Consumer для чтения порциями данных из Bluetooth, а затем каждые 10 Кбайт я записываю это в файл.Я использовал стандартную модель ПК с использованием вектора в качестве моего сообщения.Так как мне изменить это так, чтобы несколько потребителей Thread могли читать одни и те же сообщения, я думаю, что термин будет Multicaster?Я на самом деле использую это на телефоне Android, поэтому JMS, вероятно, не вариант.

static final int MAXQUEUE = 50000; 
private Vector<byte[]> messages = new Vector<byte[]>(); 

/**
 * Put the message in the queue for the Consumer Thread
 */
private synchronized void putMessage(byte[] send) throws InterruptedException { 

    while ( messages.size() == MAXQUEUE ) 
        wait(); 
    messages.addElement( send ); 
    notify(); 
} 


/**
 * This method is called by the consumer to see if any messages in the queue
 */
public synchronized byte[] getMessage()throws InterruptedException { 
    notify(); 
    while ( messages.size() == 0 && !Thread.interrupted()) {
        wait(1); 
    }
    byte[] message = messages.firstElement(); 
    messages.removeElement( message ); 
    return message; 
} 

Я ссылаюсь на код из книги Oreilly Анализатор сообщений раздел

Answer 1

Механизм Pub-sub - определенно способ достичь того, чего вы хотите.Я не уверен, почему разработка для Android ограничит вас от использования JMS, которая так же проста, как и спецификация.Проверьте эту тему на SO.

Answer 2

Это то, что я придумал в качестве примера, когда копался в коде и модифицировал некоторые существующие примеры.

package test.messaging;

import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;

public class TestProducerConsumers {

    static Broker broker;

    public TestProducerConsumers(int maxSize) {
        broker = new Broker(maxSize);
        Producer p = new Producer();
        Consumer c1 = new Consumer("One");
        broker.consumers.add(c1);
        c1.start();

        Consumer c2 = new Consumer("Two");
        broker.consumers.add(c2);
        c2.start();

        p.start();
    }

    // Test Producer, use your own message producer on a thread to call up
    // broker.insert() possibly passing it the message instead.
    class Producer extends Thread {

        @Override
        public void run() {
            while (true) {
                try {
                    broker.insert();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class Consumer extends Thread {
        String myName;
        LinkedBlockingQueue<String> queue;

        Consumer(String m) {
            this.myName = m;
            queue = new LinkedBlockingQueue<String>();
        }

        @Override
        public void run() {
            while(!Thread.interrupted()) {
                try {
                    while (queue.size() == 0 && !Thread.interrupted()) {
                        ;
                    }
                    while (queue.peek() == null && !Thread.interrupted()) {
                        ;
                    }
                    System.out.println("" + myName + " Consumer: " + queue.poll());
                } catch (Exception e) { }
            }
        }
    }

    class Broker {
        public ArrayList<Consumer> consumers = new ArrayList<Consumer>();

        int n;
        int maxSize;

        public Broker(int maxSize) {
            n = 0;
            this.maxSize = maxSize;
        }

        synchronized void insert() throws InterruptedException {
                    // only here for testing don't want it to runaway and 
                    //memory leak, only testing first 100 samples.
            if (n == maxSize)
                wait();
            System.out.println("Producer: " + n++);
            for (Consumer c : consumers) {
                c.queue.add("Message " + n);
            }
        }

    }

    public static void main(String[] args) {
        TestProducerConsumers pc = new TestProducerConsumers(100);

    }
}

Answer 3

Вы должны определенно использовать очередь вместо Vector!
Присвойте каждому потоку свою очередь и, когда получено новое сообщение, add() новое сообщение в очередь каждого потока. Для гибкости может быть полезен и шаблон слушателя.

Edit:

Хорошо, я чувствую, что тоже должен добавить пример:
(Классическая картина наблюдателя)

Это интерфейс, который должны реализовывать все потребители:

public interface MessageListener {
  public void newMessage( byte[] message );
}

Производитель может выглядеть так:

public class Producer {
  Collection<MessageListener> listeners = new ArrayList<MessageListener>();


  // Allow interested parties to register for new messages
  public void addListener( MessageListener listener ) {
    this.listeners.add( listener );
  }

  public void removeListener( Object listener ) {
    this.listeners.remove( listener );
  }

  protected void produceMessages() {
    byte[] msg = new byte[10];

    // Create message and put into msg

    // Tell all registered listeners about the new message:
    for ( MessageListener l : this.listeners ) {
      l.newMessage( msg );
    }

  }
}

И потребительский класс может быть (используя блокирующую очередь, которая делает все это wait() ing и notify() ing для нас):

public class Consumer implements MessageListener {

  BlockingQueue< byte[] > queue = new LinkedBlockingQueue< byte[] >();

  // This implements the MessageListener interface:
  @Override
  public void newMessage( byte[] message ) {
    try {
      queue.put( message );
    } catch (InterruptedException e) {
        // won't happen.
    }
  }

    // Execute in another thread:       
    protected void handleMessages() throws InterruptedException {
        while ( true ) {
            byte[] newMessage = queue.take();

            // handle the new message.
        }
    }


}