How to use java callable interface to asynchronously process queue items?

Question

I used to implement the Runnable interface to peek() an item from a queue and send it to an API.

But now I need to use Callable interface to peek() the queue and send an item to an API. If return 200, then delete the item from the queue.

Here is the code I used to implement this functionality. How can I modify the code? Any examples or reference about this? Thanks.

public class QueueProcessor implements Runnable{

private static ObjectQueue<JSONObject> objectQueue;

static {
 objectQueue = new ObjectQueue<JSONObject>();
}

public void run() {

//add items to the queue
   objectQueue.add(jsonObeject)
    Random r = new Random();
    try {
        while (true) {
            try {
                if (!objectQueue.isEmpty()) {
                    JSONObject o = objectQueue.remove();
                    sendRequest(o.toString());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

            Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
        Thread.currentThread().interrupt();
    } 
}

 public void sendRequest(JSONObject json) {

 Client client = ClientBuilder.newClient();
 WebTarget baseTarget = client.target("someUrl");
 Invocation.Builder builder = baseTarget.request();
 Response response = builder.post(Entity.entity(json.toString(), MediaType.APPLICATION_JSON));

 int code = response.getStatus();   
 if (200 == code) {
    objectQueue.remove();
 }  

}

Runnable and Callable are not that much different, besides of course the fact that callable returns a Future Object. What have you tried so far? — Tomaz Fernandes, Oct 22 '18 at 05:01

score 3 · Accepted Answer · answered Oct 22 '18 at 08:25

Just to get you started , refer to this other SO question

and note the point # 1 in the question itself.

To achieve asynchronous calls, you first need to decouple - task submission / execution ( pick item from queue and make API call ) and response processing after API call ( remove item from queue if response status is 200 ) . This decoupling can be achieved by - ExecutorService

So first introduce ExecutorService into your Runnable code i.e. start executing your Runnable from some controller class ( class with main method ) which uses an Executor to submit/execute requests. You have not shown how you triggering your thread so you might already be doing that.

Now change your Runnable into Callable<Response> i.e. create a Callable similar to your Runnable and implement Callable<Response> and in the call() method , make your API call. You do need to share your ObjectQueue<JSONObject> with your main controller class and this Callable so that queue implementation needs to be thread safe or you need to make call() method to be thread - safe.

I mean, you either loop around your queue in controller and keep submitting requests for each item or pass on whole queue to Callble and lopping is done there - its your call.

Point to note till this point is that call() method returns a value - Callable while run() method of Runnable doesn't return any value and that is a major difference between the two.

Now going back to controller class - submit or execute method will wrap your Response into a Future submit

Now use a combination of isDone() & get() methods on your Future to remove the item from queue.

Remember that you should be able to identify your processed object in queue from API response - if not then you need to combine API response with submitted JSONObject and wrap it in Future to figure out which object to remove to. Only status is not going to be enough and you might need another data structure to hold objects if queue is restricted for removal of only top element. This complication doesn't arise if you are simply replacing runnable with callable but doesn't wish to make your program truly asynchronous.

These are simply broad guidelines , providing ready made code is something that I wouldn't do. You will find lots of examples on Internet provided your basics are correct. Also, please do make a practice to include import statements while pasting code.

Few Links

How to send parallel GET requests and wait for result responses?

How to send multiple asynchronous requests to different web services?

How to use java callable interface to asynchronously process queue items?

1 Answers1