1. import java.util.concurrent.*;
  2. import java.util.concurrent.atomic.AtomicInteger;
  3. import org.json.JSONObject;
  5. public class JsonResponseQueue {
  7. private final BlockingQueue<JSONObject> taskQueue;
  8. private final int numConsumers;
  9. private final AtomicInteger taskCount = new AtomicInteger(0);
  11. public JsonResponseQueue(int numConsumers) {
  12. this.taskQueue = new LinkedBlockingQueue<>();
  13. this.numConsumers = numConsumers;
  14. }
  16. public void enqueue(JSONObject task) {
  17. taskQueue.offer(task); // Add task to the queue
  18. }
  20. public void startConsumers() {
  21. ExecutorService executor = Executors.newFixedThreadPool(numConsumers);
  23. for (int i = 0; i < numConsumers; i++) {
  24. executor.submit(new Consumer(taskQueue));
  25. }
  26. }
  28. // Consumer class to process JSON responses with retry logic
  29. private class Consumer implements Runnable {
  30. private final BlockingQueue<JSONObject> queue;
  32. public Consumer(BlockingQueue<JSONObject> queue) {
  33. this.queue = queue;
  34. }
  36. @Override
  37. public void run() {
  38. while (true) {
  39. try {
  40. JSONObject task = queue.take(); // Block until a task is available
  41. processTask(task);
  42. } catch (InterruptedException e) {
  43. // Handle interruption (e.g., shutdown signal)
  44. Thread.currentThread().interrupt();
  45. return;
  46. }
  47. }
  48. }
  50. private void processTask(JSONObject task) {
  51. int retryCount = 0;
  52. while (retryCount < 3) { // Retry up to 3 times
  53. try {
  54. // Simulate a network call or processing that might fail
  55. if (simulateFailure()) {
  56. retryCount++;
  57. System.out.println("Task failed, retry attempt: " + retryCount);
  58. Thread.sleep(1000); // Wait before retrying
  59. } else {
  60. System.out.println("Task processed successfully: " + task.toString());
  61. return; // Task completed successfully
  62. }
  63. } catch (Exception e) {
  64. System.err.println("Error processing task: " + e.getMessage());
  65. retryCount++;
  66. Thread.sleep(2000); // Wait before retrying
  67. }
  68. }
  69. System.err.println("Task failed after multiple retries: " + task.toString());
  70. }
  72. private boolean simulateFailure() {
  73. // Simulate a failure condition (e.g., network error)
  74. return Math.random() < 0.3; // 30% chance of failure
  75. }
  76. }
  78. public static void main(String[] args) throws InterruptedException {
  79. JsonResponseQueue queue = new JsonResponseQueue(3); // Create a queue with 3 consumers
  81. // Start the consumers
  82. queue.startConsumers();
  84. // Enqueue some tasks
  85. for (int i = 0; i < 10; i++) {
  86. JSONObject task = new JSONObject().put("id", i).put("data", "some data");
  87. queue.enqueue(task);
  88. Thread.sleep(200); // Simulate task generation
  89. }
  91. // Keep the main thread alive for a while to allow consumers to process tasks
  92. Thread.sleep(5000);
  93. }
  94. }

Add your comment