使用Redis實(shí)現(xiàn)輕量級消息隊列

更新時間：2025年11月14日 10:08:48 作者：moxiaoran5753

使用消息中間件如RabbitMQ或kafka雖然好,但也給服務(wù)器帶來很大的內(nèi)存開銷,當(dāng)系統(tǒng)的業(yè)務(wù)量,并發(fā)量不高時,考慮到服務(wù)器和維護(hù)成本,可考慮使用Redis實(shí)現(xiàn)一個輕量級的消息隊列,下面就來詳細(xì)的介紹一下

使用消息中間件如RabbitMQ或kafka雖然好，但也給服務(wù)器帶來很大的內(nèi)存開銷，當(dāng)系統(tǒng)的業(yè)務(wù)量，并發(fā)量不高時，考慮到服務(wù)器和維護(hù)成本，可考慮使用Redis實(shí)現(xiàn)一個輕量級的消息隊列，實(shí)現(xiàn)事件監(jiān)聽的效果。下面介紹下Redis實(shí)現(xiàn)消息隊列的三種形式。

方式一 Redis Pub/Sub（適用于廣播通知）

Redis Pub/Sub 適用于 實(shí)時消息推送，但不支持消息持久化，如果消費(fèi)者掉線，消息會丟失。

(1) 發(fā)布消息（生產(chǎn)者）

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderService {
    private final StringRedisTemplate redisTemplate;

    public OrderService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void createOrder(Long orderId) {
        System.out.println("訂單創(chuàng)建成功: " + orderId);

        // 發(fā)布消息
        redisTemplate.convertAndSend("order.channel", orderId.toString());
    }
}

(2) 訂閱消息（消費(fèi)者）

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;

@Service
public class NotificationService implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String orderId = message.toString();
        System.out.println("【通知服務(wù)】收到訂單創(chuàng)建消息：" + orderId);
    }
}

(3) 注冊 Redis 監(jiān)聽器

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
public class RedisPubSubConfig {

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                   MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, new PatternTopic("order.channel"));
        return container;
    }

    @Bean
    public MessageListenerAdapter listenerAdapter(NotificationService receiver) {
        return new MessageListenerAdapter(receiver, "onMessage");
    }
}

缺點(diǎn)

無持久化，消費(fèi)者掉線后無法重新獲取消息。
不支持消費(fèi)組，多個消費(fèi)者同時訂閱時，所有都會收到消息（無法負(fù)載均衡）。

方式二：Redis List（適用于任務(wù)隊列）

使用 Redis List（LPUSH + RPOP）可以實(shí)現(xiàn)簡單的任務(wù)隊列，適用于任務(wù)異步處理，但不支持回溯消費(fèi)。

(1) 生產(chǎn)者（推送任務(wù)）

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderService {
    private final StringRedisTemplate redisTemplate;

    public OrderService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void createOrder(Long orderId) {
        System.out.println("訂單創(chuàng)建成功: " + orderId);

        // 推送到隊列
        redisTemplate.opsForList().leftPush("order.queue", orderId.toString());
    }
}

(2) 消費(fèi)者（輪詢獲取任務(wù)）

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class NotificationService {
    private final StringRedisTemplate redisTemplate;

    public NotificationService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Scheduled(fixedDelay = 5000) // 每5秒輪詢一次
    public void processOrderQueue() {
        String orderId = redisTemplate.opsForList().rightPop("order.queue");
        if (orderId != null) {
            System.out.println("【通知服務(wù)】處理訂單：" + orderId);
        }
    }
}

要想消費(fèi)者能監(jiān)聽到消息并進(jìn)行處理，需要在方法上添加@Scheduled注解，同時在服務(wù)啟動類中添加@EnableScheduling注解，或者在配置類添加

import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableScheduling
public class SchedulingConfig {
}

缺點(diǎn)

無消費(fèi)組，多個消費(fèi)者時需要自行分配任務(wù)，可能會造成任務(wù)重復(fù)消費(fèi)或丟失。
無持久化保障，如果任務(wù)未處理完，Redis 發(fā)生故障，任務(wù)可能會丟失。

方式三：Redis Stream（推薦，支持持久化 + 消費(fèi)組）

Redis Stream 是 Redis 6.0 之后的特性，類似于 Kafka，支持持久化、消費(fèi)組、多消費(fèi)者模式。

(1) 生產(chǎn)者（推送事件）

import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class OrderService {
    private final StringRedisTemplate redisTemplate;

    public OrderService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void createOrder(Long orderId) {
        System.out.println("訂單創(chuàng)建成功: " + orderId);

        // 推送到 Redis Stream
        ObjectRecord<String, String> record = StreamRecords.newRecord()
                .ofObject(orderId.toString())
                .withStreamKey("order.stream");
        redisTemplate.opsForStream().add(record);
    }
}

(2) 消費(fèi)者（監(jiān)聽事件）

import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;

@Service
public class NotificationService implements StreamListener<String, MapRecord<String, String, String>> {

    private final StringRedisTemplate redisTemplate;
    private final StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer;

    public NotificationService(StringRedisTemplate redisTemplate,
                               StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer) {
        this.redisTemplate = redisTemplate;
        this.listenerContainer = listenerContainer;
    }

    @PostConstruct
    public void startListening() {
        listenerContainer.receive(StreamOffset.fromStart("order.stream"), this);
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String orderId = message.getValue().values().iterator().next();
        System.out.println("【通知服務(wù)】訂單 " + orderId + " 創(chuàng)建成功！");
    }
}

優(yōu)點(diǎn)

持久化存儲，即使 Redis 重啟，消息不會丟失。
支持消費(fèi)組，多個消費(fèi)者可以負(fù)載均衡地消費(fèi)消息。
支持回溯，可以讀取歷史消息。

總結(jié)

方案	適用場景	優(yōu)點(diǎn)	缺點(diǎn)
Pub/Sub	即時消息通知	低延遲	無持久化，消費(fèi)者掉線丟消息
List	簡單任務(wù)隊列	輕量級	無消費(fèi)組，任務(wù)可能丟失
Stream	高級事件流處理	持久化、消費(fèi)組	復(fù)雜度較高