1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.concurrent.ConcurrentHashMap;
  4. import java.util.concurrent.Concurrent久期队列;
  5. import java.util.concurrent.Executors;
  6. import java.util.concurrent.ScheduledExecutorService;
  7. import java.util.concurrent.TimeUnit;
  9. public class DataStreamer {
  11. private final ConcurrentHashMap<String, List<String>> dataCache = new ConcurrentHashMap<>(); // Stores data for each key
  12. private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); // For periodic processing
  13. private final long timeoutMillis; // Timeout duration in milliseconds
  14. private final Runnable cleanupTask; // Task to clean up old data
  16. public DataStreamer(long timeoutMillis, Runnable cleanupTask) {
  17. this.timeoutMillis = timeoutMillis;
  18. this.cleanupTask = cleanupTask;
  19. startCleanupTask();
  20. }
  22. private void startCleanupTask() {
  23. scheduler.scheduleAtFixedRate(cleanupTask, timeoutMillis, timeoutMillis, TimeUnit.MILLISECONDS);
  24. }
  27. public void addData(String key, String data) {
  28. dataCache.computeIfAbsent(key, k -> new ArrayList<>()).add(data); // Add data to the list for the key
  29. }
  31. public List<String> getData(String key) {
  32. return dataCache.getOrDefault(key, new ArrayList<>()); // Return data for the key, or an empty list if not found
  33. }
  35. public void stop() {
  36. scheduler.shutdown(); // Stop the scheduler
  37. }
  38. }

Add your comment