BlockingQueue.size（）在发布服务器订阅服务器中返回错误的大小

Question

我有关于一个发布服务器 - 多订阅服务器模式的实现的问题。发布者使用固定大小的缓冲区并对消息进行排队。这些消息被发送给所有用户。订阅者获得的消息的排序必须与发布消息的排序相同。

我使用的BlockingQueue持有发行消息（publisherQueue）并将它们传递到每个订户的BlockingQueue（subscriberQueue）。

问题是缓冲和用户工作正常，但缓冲区大小（publisherQueue.size（））始终返回1.

System.out.println("Actual number of messages in buffer: " + publisherQueue.size()); 

这里是我的全码：

PublisherSubscriberService.java

package program; 

import java.util.HashSet; 
import java.util.Set; 
import java.util.concurrent.BlockingQueue; 
import java.util.concurrent.LinkedBlockingQueue; 


public class PublisherSubscriberService { 
    private int buffer; 
    private int subscribersNumber; 
    static Set<subscriber> subscribers = new HashSet<subscriber>(); 

    public PublisherSubscriberService(int buffer, int subscribersNumber) { 
     this.buffer = buffer; 
     this.subscribersNumber = subscribersNumber; 
    } 

    public void addsubscriber(subscriber subscriber) { 
     subscribers.add(subscriber); 
    } 

    public void start() { 
     publisher publisher = new publisher(buffer); 
     System.out.println("publisher started the job"); 

     for (int i = 0; i < subscribersNumber; i++) { 
      subscriber subscriber = new subscriber(buffer); 
      subscriber.setName(Integer.toString(i + 1)); 
      subscribers.add(subscriber); 
      new Thread(subscriber).start(); 
      System.out.println("Subscriber " + subscriber.getName() + " started the job"); 
     } 
     new Thread(publisher).start(); 
    } 

    public class Publisher implements Runnable { 
     private int buffer; 
     final BlockingQueue<Message> publisherQueue; 

     public Publisher(int buffer) { 
      this.buffer = buffer; 
      publisherQueue = new LinkedBlockingQueue<>(buffer); 
     } 

     @Override 
     public void run() { 
      for (int i = 1; i < 100; i++) { 
       Message messageObject = new Message("" + i); 
       try { 
        Thread.sleep(50); 
        publisherQueue.put(messageObject); 
        System.out.println("Queued message no " +   messageObject.getMessage()); 
        System.out.println("Actual number of messages in buffer:  " + publisherQueue.size()); 
        for (subscriber subscriber : subscribers) { 
         subscriber.subscriberQueue.put(messageObject); 
        } 
        publisherQueue.take(); 
       } catch (InterruptedException e) { 
        System.out.println("Some error"); 
        e.printStackTrace(); 
       } 
      } 
     } 
    } 

    class Subscriber implements Runnable { 
     private String name; 
     private int buffer; 
     final BlockingQueue<Message> subscriberQueue; 

     public Subscriber(int buffer) { 
      this.buffer = buffer; 
      subscriberQueue = new LinkedBlockingQueue<>(buffer); 
     } 

     public void setName(String name) { 
      this.name = name; 
     } 

     public String getName() { 
      return name; 
     } 

     @Override 
     public void run() { 
      try { 
       Message messageObject; 
       while (true) { 
        Thread.sleep(100); 
        messageObject = subscriberQueue.take(); 
        System.out.println(this.getName() + " got message: " + messageObject.getMessage()); 
       } 
      } catch (InterruptedException e) { 
       System.out.println("Some error"); 
       e.printStackTrace(); 
      } 
     } 
    } 
class Message { 
    private String message; 

    public Message(String str) { 
     this.message = str; 
    } 

    public String getMessage() { 
     return message; 
    } 

} 
} 

PublisherSubscriberProgram.java

package program; 

public class ProducerConsumerProgram { 

    public static void main(String[] args) { 
     ProducerConsumerService service = new ProducerConsumerService(10, 3); 
     service.start(); 
    } 
} 

Answer 1

您的出版商从来没有在排队超过1个项目。每次通过你的循环，你把和单个项目：

   **publisherQueue.put(messageObject);** 
       System.out.println("Queued message no " +   messageObject.getMessage()); 
       System.out.println("Actual number of messages in buffer:  " + publisherQueue.size()); 
       for (subscriber subscriber : subscribers) { 
        subscriber.subscriberQueue.put(messageObject); 
       } 
       **publisherQueue.take();** 

随着您提供的代码，甚至有发布者队列的点。