java redis 消息队列

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

public class RedisMessageQueueExample {

    // Redis服务器地址和端口
    private static final String REDIS_HOST = "localhost";
    private static final int REDIS_PORT = 6379;

    // 消息队列的Stream名称
    private static final String STREAM_NAME = "message_queue";

    public static void main(String[] args) {
        Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT);

        // 生产者：向消息队列中添加消息
        addMessage(jedis, "Hello, Redis Stream!");
        addMessage(jedis, "This is another message.");

        // 消费者：从消息队列中读取消息
        readMessages(jedis);

        // 关闭连接
        jedis.close();
    }

    // 向消息队列中添加消息
    private static void addMessage(Jedis jedis, String message) {
        // 使用XADD命令将消息添加到Stream中
        String id = jedis.xadd(STREAM_NAME, new XAddParams().maxLenApproximately(100), "message", message);
        System.out.println("Added message with ID: " + id);
    }

    // 从消息队列中读取消息
    private static void readMessages(Jedis jedis) {
        // 使用XRANGE命令读取Stream中的所有消息
        var messages = jedis.xrange(STREAM_NAME, "-", "+");

        System.out.println("Reading messages from the queue:");
        for (var message : messages) {
            System.out.println("Message ID: " + message.getId());
            System.out.println("Message Content: " + message.getValues().get("message"));
        }
    }
}

解释说明

1. Redis连接:

   • 使用Jedis客户端连接到Redis服务器，指定主机和端口。

2. 生产者:

   • addMessage方法使用XADD命令将消息添加到Redis Stream中。XADD命令的第一个参数是Stream的名称，第二个参数是消息的内容。
   • maxLenApproximately(100)限制Stream的最大长度为100条记录（近似值）。

3. 消费者:

   • readMessages方法使用XRANGE命令读取Stream中的所有消息。XRANGE命令的第三个和第四个参数分别是起始ID和结束ID，这里使用-和+表示读取所有消息。
   • 遍历返回的消息列表并打印每条消息的ID和内容。

4. 关闭连接:

   • 最后不要忘记关闭Jedis连接以释放资源。

这个示例展示了如何使用Redis Stream实现简单的消息队列功能。