实战案例：使用Redis Streams构建一个实时数据管道

import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.stream.XAddParams;


public class DataProducer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost");
        String streamKey = "realtimeData";


        while (true) {
            String data = "Temperature: " + (int)(Math.random() * 100) + "°C, Humidity: " + (int)(Math.random() * 100) + "%";


            System.out.println("Adding data to stream '" + streamKey + "': " + data);
            jedis.xadd(streamKey, XAddParams.none(), "data", data);


            try {
                Thread.sleep(1000); // 模拟每秒产生一次数据
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.params.stream.StreamReadOptions;
import redis.clients.jedis.params.stream.StreamReadParams;
import redis.clients.jedis.resps.stream.StreamEntry;


public class DataConsumer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost");
        String streamKey = "realtimeData";


        while (true) {
            StreamEntry entry = jedis.xread(StreamReadOptions.Builder.id("$"), StreamReadParams.Builder.key(streamKey)).get(0).getEntries().get(0);
            if (entry != null) {
                System.out.println("Consumed data from stream '" + streamKey + "': " + entry.getFields().get("data"));


                // 处理数据
                // ...


                jedis.xack(streamKey, entry.getId());
            } else {
                // 如果队列为空，则等待一段时间再尝试
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}