Java使用Redis实现消息队列

近期刷Java面试题刷到了“如何使用Redis实现消息队列”，解答如下：
一般使用 list 结构作为队列， rpush 生产消息， lpop 消费消息。当 lpop 没有消息的时候，要适当sleep 一会再重试。若不使用sleep，则可以用指令blpop（该指令在没有消息的时候，它会阻塞住直到消息到来）

目录

引入Jedis
指令简介
rpush
blpop

Java使用Redis实现消息队列

引入Jedis

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>3.7.0</version>
</dependency>

指令简介

rpush

rpush操作是向Redis服务器中的一个列表（List）中添加一个元素。这个操作通常用于向队列中添加任务或者数据。
语法：rpush key value1 [value2] [value3] …
示例：rpush my_list “task1” “task2” “task3”

blpop

blpop操作是用于移除并获取Redis服务器列表中的第一个元素，在该元素存在时会阻塞列表直到超时或发现可弹出的元素。
语法：blpop key1 [key2 …] timeout
示例：blpop my_list 5 （移除my_list中第一个元素且最多阻塞等待5秒钟）
多个列表可以这样写 blpop mylist1 mylist2 10

Java使用Redis实现消息队列

模拟生产者

import redis.clients.jedis.Jedis;

public class Producer {
    private Jedis jedis;
    private String queueName;

    public Producer(String queueName) {
        this.jedis = new Jedis("localhost", 6379);
        this.queueName = queueName;
    }

    public void produce(String message) {
        jedis.rpush(queueName, message);
    }
}

模拟消费者

import redis.clients.jedis.Jedis;
import java.util.List;

public class Consumer {
    private Jedis jedis;
    private String queueName;

    public Consumer(String queueName) {
        this.jedis = new Jedis("localhost", 6379);
        this.queueName = queueName;
    }

    public String consume() {
    	//参数1是阻塞时长，0代表不限时
        List<String> result = jedis.blpop(0, queueName);
        return result.get(1);
    }
}

生产者端测试

public class MainP {
    public static void main(String[] args) {

        Producer producer = new Producer("queue");

        Thread producerThread = new Thread(() -> {
            producer.produce("hello world");
        });

        producerThread.start();
    }
}

消费者端测试

public class MainC {

    public static void main(String[] args) {
        Consumer consumer = new Consumer("queue");

        Thread consumerThread = new Thread(() -> {
        	//当没有消息的时候，此处会自动堵塞等待消息
            while (true){
                String message = consumer.consume();
                System.out.println("消费消息: " + message);
            }
        });

        consumerThread.start();
    }
}