title: Java BlockingQueue produces/consumes not properly
body: <p>I'm working on a project where I need to retrieve Twitter messages using the TwitterAPI, process them and store them in a database. I am using Producer/Consumer BlockingQueue where the elements act as follows:</p>
  
  <ul>
  <li>Producer : retrieves the Twitter messages using the TwitterAPI and stores it in a BlockingQueue.</li>
  <li>Consumer : takes an element from the queue, processes it and stores it in a database.</li>
  </ul>
  
  <p>Here is the Main class:</p>
  
  <pre><code>    // Creating shared object
      BlockingQueue&lt;TwitterMessage&gt; sharedQueue = new ArrayBlockingQueue&lt;TwitterMessage&gt;(1);
  
      // Creating Producer and Consumer Thread
      Thread prodThread = new Thread(new TwitterStreamProducer(sharedQueue));
      Thread consThread = new Thread(new TwitterStreamConsumer(sharedQueue));
  
      // Starting producer and Consumer thread
      prodThread.start();
      consThread.start();
  </code></pre>
  
  <p>The producer processes the TwitterAPI response and adds the object to the queue.</p>
  
  <pre><code>@Override
  public void run() {
      while (true) {
          try {
              message = extractData(); // extract data from TwitterAPI response and return TwitterMessage object
              sharedQueue.put(message);
  
              System.out.println("Produced: " + message.getTwitterMessage());
          } catch (Exception ex) {
              Logger.getLogger(TwitterStreamProducer.class.getName()).log(Level.SEVERE, null, ex);
          }
      }
  }
  </code></pre>
  
  <p>The consumer does as follows:</p>
  
  <pre><code>private final BlockingQueue&lt;TwitterMessage&gt; sharedQueue;
  private TwitterProcessor twitterProcessor;
  private TwitterMessage twitterMessage;
  
  public TwitterStreamConsumer(BlockingQueue&lt;TwitterMessage&gt; sharedQueue) {
      this.sharedQueue = sharedQueue;
      twitterProcessor = new TwitterProcessor();
  }
  
  @Override
  public void run() {
      while (true) {
          try {
              twitterMessage = this.twitterProcessor.process(sharedQueue.take());
              if (twitterMessage.getTwitterMessage().length() &gt; 1) {
                  System.out.printf("Consumed: %s\n", twitterMessage.getTwitterMessage());
              }
          } catch (InterruptedException ex) {
              Logger.getLogger(TwitterStreamConsumer.class.getName()).log(Level.SEVERE, null, ex);
          }
      }
  }
  </code></pre>
  
  <p>As I would expect to see is the following:</p>
  
  <pre><code>Produced: …twittermessage1… 
  Consumed: …twittermessage1… 
  Produced: …twittermessage2… 
  Consumed: …twittermessage2… 
  Produced: …twittermessage3… 
  Consumed: …twittermessage3…
  ...
  </code></pre>
  
  <p>However, the result I'm getting is the following:</p>
  
  <pre><code>Produced: …twittermessage1…
  Produced: …twittermessage2…  &lt;= problem
  Consumed: …twittermessage1…
  Produced: …twittermessage3…
  Consumed: …twittermessage3…
  Consumed: …twittermessage3…  &lt;= problem
  Produced: …twittermessage4…  &lt;= problem
  Produced: …twittermessage5…
  Consumed: …twittermessage5…
  </code></pre>
  
  <p>As you can see, sometimes there is an overlap between Producer and Consumer where the Producer produces messages too many that are not consumed. Also sometimes a message gets consumed twice (or sometimes even more than twice)</p>
  
  <p><strong>EDIT1</strong>
  Here is what is printed out on the console:</p>
  
  <pre><code>Produced: @1StevenGeorgiou thanks for the follow #ff
  Processed: follow
  Produced: @nmagliozzi6 @_PatrickKealy_ but of course!!!!!
  Produced: @taylorgaglia Thanks Tayl  miss you tooo
  Processed: tayl miss
  Produced: Hate this who to follow tab in #twitter it's shows the most pathetic people you know. Accidently added one I had to act fast to unfollow
  Processed: hate follow tabshow pathet peopl accid ad act fast unfollow
  </code></pre>
  
  <p><strong>EDIT2</strong>
  As John Vint suggested to print out the 'System.identityHashCode(sharedQueue.take())' I get the following:</p>
  
  <pre><code>Produced: …
  Consumed: 1206857787
  Produced: …
  Consumed: 1206857787
  …
  </code></pre>
  
  <p>Can someone help me out here how I can fix this?</p>
  
  <p>Thanks!</p>