BlockingQueue.size() returns wrong size in Publisher-Subscribers

I have the problem regarding the implementation of One Publisher - Multiple Subscribers pattern. The Publisher uses the fixed-size buffer and queue the messages. The messages are send to all subscribers. The ordering of messages get by subscribers must be the same as the ordering of publishing messages.

I use BlockingQueue to hold publisher messages (publisherQueue) and pass them to each subscriber BlockingQueue (subscriberQueue).

The issue is that the buffer and subscribers are working correctly, but the buffer size (publisherQueue.size()) always returns 1.

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

Here is my full code:

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();
    }
}

Copyright License：
Author:「Admx」,Reproduced under the CC 4.0 BY-SA copyright license with link to original source & disclaimer.
Link to：https://stackoverflow.com/questions/28255559/blockingqueue-size-returns-wrong-size-in-publisher-subscribers