Feb 24, 2014

Producer-Consumer example with Twitter Stream reading

While looking at examples on Producer Consumer using Java I was not able to find much of practical examples so let’s take some real time example. As explained in last blog we can read stream provided by Twitter using Twitter4J api which sends us real time Twitter data. So now lets use this feed as producer and we design a consumer that will check count of twits that we are getting for particular key word.

So first we will change our StreamReaderService to accept a queue and list of keywords as follows.

package com.techcielo.twitreader.service;

import java.util.concurrent.BlockingQueue;

import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;

import com.techcielo.twitreader.bean.TwitterStreamBean;
import com.techcielo.twitreader.util.TwitterStreamBuilderUtil;

public class StreamReaderService implements Runnable
       private BlockingQueue<TwitterStreamBean> beanQ;
       private String[] keywords;
       public StreamReaderService(BlockingQueue<TwitterStreamBean> beanQ,String[] keywords) {
       public void readTwitterFeed() {

              TwitterStream stream = TwitterStreamBuilderUtil.getStream();

              StatusListener listener = new StatusListener() {

                     public void onException(Exception e) {
                           System.out.println("Exception occured:" + e.getMessage());

                     public void onTrackLimitationNotice(int n) {
                           System.out.println("Track limitation notice for " + n);

                     public void onStatus(Status status) {
                           System.out.println("Got twit:" + status.getText());
                           TwitterStreamBean bean = new TwitterStreamBean();
                           String username = status.getUser().getScreenName();
                           long tweetId = status.getId();
                           if (status != null && status.getRetweetedStatus() != null
                                         && status.getRetweetedStatus().getUser() != null) {
                           String content = status.getText();
                           try {
                           } catch (InterruptedException e) {
                                  System.out.println("Error occured while pusing feed data");

                     public void onStallWarning(StallWarning arg0) {
                           System.out.println("Stall warning");

                     public void onScrubGeo(long arg0, long arg1) {
                           System.out.println("Scrub geo with:" + arg0 + ":" + arg1);

                     public void onDeletionNotice(StatusDeletionNotice arg0) {
                           System.out.println("Status deletion notice");
              FilterQuery qry = new FilterQuery();


       public void run() {

Now we will create a consumer that will accept keywords as constructor argument and when twit comes to this queue we will check for index of keyword in it and display count for each keyword. So following will be our consumer class.

package com.techcielo.twitreader.service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import com.techcielo.twitreader.bean.TwitterStreamBean;

public class StreamConsumerService implements Runnable{
       private BlockingQueue<TwitterStreamBean> beanQ;
       private String[] keywords;
       private Map<String, Integer> countmap=new HashMap<>();
       public StreamConsumerService(BlockingQueue<TwitterStreamBean> beanQ, String[] keywords) {
       public void run() {
              System.out.println("Started twit parser");
                     try {
                           TwitterStreamBean bean = beanQ.poll(1, TimeUnit.MILLISECONDS);
                                  for(int i=0;i<keywords.length;i++){
                                                       countmap.put(keywords[i].toUpperCase(), countmap.get(keywords[i].toUpperCase())+1);
                                                       countmap.put(keywords[i].toUpperCase(), 1);
                     } catch (InterruptedException e) {

Now we will create a class to start producer and consumer service as follows.

package com.techcielo.twitreader.main;

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

import com.techcielo.twitreader.bean.TwitterStreamBean;
import com.techcielo.twitreader.service.StreamConsumerService;
import com.techcielo.twitreader.service.StreamReaderService;

public class Main {
       public static void main(String[] args) {
              String keywords[] = {"Sochi","Whatsapp"};
              BlockingQueue<TwitterStreamBean> beanQ = new LinkedBlockingQueue<>();
              StreamReaderService producer = new StreamReaderService(beanQ,keywords);
              StreamConsumerService consumer = new StreamConsumerService(beanQ,keywords);
              Thread t1 = new Thread(producer);
              Thread t2 = new Thread(consumer);

So when we run this code following will be output.

Started twit parser
[Mon Feb 24 14:17:30 IST 2014]Establishing connection.
[Mon Feb 24 14:17:34 IST 2014]Connection established.
[Mon Feb 24 14:17:34 IST 2014]Receiving status stream.
Got twit:Yaampun -_- cepet download whatsapp ya :p RT @kunuuuuy: nizaarr cpt beli pulsa ya;)
Got twit:I shouldn't whatsapp him anymore. Or maybe once in 2 weeks time only.
Got twit:RT @pauli_roldan: #Whatsapp #Male #Me #Cuenta #De #La #Vida #De #Justin #Y #Avalanna http://t.co/W1Y6RnwX4Y
Got twit:RT @molnia_sport: ?????????? ???: ?????? ???????? ?? ?????? ???? ???????? ? ????????, ?? ? ??????? ?????

http://t.co/DOfXgS5p6l http://t.c…
Got twit:Beego + Linda | Expecting
image by Wafi Abu
Ryaniez Moments Photography

Call/SMS/Whatsapp 0192044408 (Hash) or... http://t.co/WevB66eHGo
Got twit:Millions Flock to Telegram Messaging App After WhatsApp Outage http://t.co/TD46LQEGTE I'm going to check it out
Got twit:RT @MyFashionBDG: Kode: Stitch Air  75.000 (blm ongkir) | For order sms/whatsapp ke 081392239444 http://t.co/NjFOTiCT8n Fb: http://t.co/jAx…
Got twit:(#Houston_0998) ¿Qué provocó la caída de varias horas de WhatsApp?: El fundador de WhatsApp, J... http://t.co/qhERwrXZfy (#Houston_0998)
Got twit:RT @undyt_: Whatsapp jangan diheboh-hebohin doong, tar jadi ribet kayak aplikasi lain deh. :p
Got twit:RT @orangeflower08: ??????????????????????????????????18????????????????? #????

RT @NBCOlympics We get it, Sochi bear. We're crying too. h…
Got twit:RT @A3Noticias: Cuatro millones de usuarios se pasan a Telegram tras las caídas de Whatsapp... para sufrir las mismas averías http://t.co/F…
Got twit:RT @morandiniblog: Le fondateur de WhatsApp s'excuse pour la panne " la plus longue et la plus importante depuis des années" #whatsapp http…
Count:{WHATSAPP=10, SOCHI=1}
Got twit:RT @puur: Ik vind een @YouTube-video van @timtimfed leuk: http://t.co/cKu9YDbXHu SOCHI 2014 - Speed Skating Double Dash Final
Count:{WHATSAPP=10, SOCHI=2}
Got twit:aku gk pake BB @vivinopiw_ aku pake Whatsapp, kmu pake gk ?
Count:{WHATSAPP=11, SOCHI=2}
Got twit:@iliacolmenares te escribo a whatsapp y nada, no se que le pasa a mi teléfono... Poblado esta totalmente trancado desde las 3 am.
Count:{WHATSAPP=12, SOCHI=2}
Got twit:RT @ExpansionaT: ¿Qué provocó la caída de varias horas de WhatsApp? http://t.co/prjP7Nn3Ae
Count:{WHATSAPP=13, SOCHI=2}
Got twit:(#Houston_0998) ¿Qué provocó la caída de varias horas de WhatsApp?: El fundador de WhatsApp, J... http://t.co/SZJ2gz5tpG (#Houston_0998)
Got twit:RT @jonipikkuinen: Varoittakaa lapsianne klikkamasta "whatsapp-vanhenee tänään"-linkkejä ja vastaavia puhelimessa. Joutuvat FunLimen 79€/kk.
Count:{WHATSAPP=14, SOCHI=2}
Count:{WHATSAPP=15, SOCHI=2}
Got twit:Facebook rachète WhatsApp pour 63 fois le prix d'un Airbus A380, le retour de la bulle internet ? http://t.co/DZS8A2AJO8
Count:{WHATSAPP=16, SOCHI=2}
Got twit:Facebook rachète WhatsApp pour 63 fois le prix d'un Airbus A380, le retour de la bulle internet ? http://t.co/MiysQgJC7R
Count:{WHATSAPP=17, SOCHI=2}
Got twit:Ya estamos con la tontería de las cadenas disque para que el whatsapp no t cueste dinero
Count:{WHATSAPP=18, SOCHI=2}