Redis 实现消息队列实际案例

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.4.3</version> <!-- 建议使用最新稳定版 -->
</dependency>

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
public class ListMQProducer {
    // 队列key命名规范：业务域:组件类型:数据结构:具体业务
    private static final String QUEUE_KEY = "redis:mq:list:order"; 
    // 使用连接池提高性能
    private static final JedisPool jedisPool = new JedisPool(
        new JedisPoolConfig(),
        "localhost", 
        6379,
        2000,  // 连接超时时间
        null   // 密码
    );
    public static void main(String[] args) throws InterruptedException {
        try (Jedis jedis = jedisPool.getResource()) {
            // 模拟发送10条订单消息
            for (int i = 1; i <= 10; i++) {
                // 消息内容格式：业务标识_序号_时间戳
                String message = String.format("order_%d_%d", i, System.currentTimeMillis());
                // LPUSH命令将消息放入队列头部
                long queueLength = jedis.lpush(QUEUE_KEY, message);
                System.out.printf("生产者发送消息：%s，当前队列长度：%d%n", message, queueLength);
                // 模拟业务处理间隔
                Thread.sleep(500); 
            }
        } catch (Exception e) {
            System.err.println("生产者异常：" + e.getMessage());
        } finally {
            jedisPool.close();
        }
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import java.util.List;
public class ListMQConsumer {
    private static final String QUEUE_KEY = "redis:mq:list:order";
    private static final JedisPool jedisPool = new JedisPool(
        new JedisPoolConfig(),
        "localhost",
        6379
    );
    public static void main(String[] args) {
        System.out.println("消费者启动，等待接收消息...");
        while (true) {
            try (Jedis jedis = jedisPool.getResource()) {
                // BRPOP命令参数：
                // 1. 超时时间3秒(避免空轮询消耗CPU)
                // 2. 可以监听多个队列
                List<String> messages = jedis.brpop(3, QUEUE_KEY);
                if (messages != null) {
                    // BRPOP返回结果格式：
                    // 第一个元素是队列key
                    // 第二个元素是消息内容
                    String message = messages.get(1);
                    System.out.println("消费者接收消息：" + message);
                    // 业务处理逻辑示例
                    processMessage(message);
                } else {
                    System.out.println("队列暂无消息，继续等待...");
                }
            } catch (Exception e) {
                System.err.println("消费者处理消息异常：" + e.getMessage());
                // 异常处理策略：
                // 1. 记录错误日志
                // 2. 重试机制
                // 3. 告警通知
                try {
                    Thread.sleep(5000); // 出错后暂停5秒
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
    private static void processMessage(String message) throws InterruptedException {
        // 模拟业务处理
        System.out.println("处理消息：" + message);
        // 解析消息内容
        String[] parts = message.split("_");
        String orderId = parts[1];
        // 模拟业务处理耗时
        Thread.sleep(1000);
        System.out.println("订单" + orderId + "处理完成");
    }
}

private static final String CONFIRM_QUEUE_KEY = "redis:mq:list:order:confirm";
private static final String DEAD_QUEUE_KEY = "redis:mq:list:order:dead";
private static final int MAX_RETRY = 3;
// 优化后的消费者处理逻辑
List<String> messages = jedis.brpop(3, QUEUE_KEY);
if (messages != null) {
    String message = messages.get(1);
    // 1. 将消息移到待确认队列(使用RPUSH保持顺序)
    jedis.rpush(CONFIRM_QUEUE_KEY, message);
    try {
        // 2. 处理业务逻辑
        processMessage(message);
        // 3. 处理成功，从待确认队列删除
        jedis.lrem(CONFIRM_QUEUE_KEY, 1, message);
    } catch (Exception e) {
        System.err.println("处理消息失败：" + message);
        // 4. 检查重试次数
        long retryCount = jedis.incr("retry:" + message);
        if (retryCount <= MAX_RETRY) {
            // 放回主队列重试
            jedis.lpush(QUEUE_KEY, message);
        } else {
            // 超过重试次数，放入死信队列
            jedis.rpush(DEAD_QUEUE_KEY, message);
        }
        // 无论重试还是加入死信队列，都要从待确认队列删除
        jedis.lrem(CONFIRM_QUEUE_KEY, 1, message);
    }
}

// 定时检查待确认队列(每分钟执行)
public void checkConfirmQueue() {
    try (Jedis jedis = jedisPool.getResource()) {
        // 获取待确认队列所有消息
        List<String> pendingMessages = jedis.lrange(CONFIRM_QUEUE_KEY, 0, -1);
        for (String message : pendingMessages) {
            // 检查消息滞留时间
            long createTime = Long.parseLong(message.split("_")[2]);
            long currentTime = System.currentTimeMillis();
            long delay = currentTime - createTime;
            // 超过30秒未处理则重试
            if (delay > 30000) {
                jedis.lrem(CONFIRM_QUEUE_KEY, 1, message);
                jedis.lpush(QUEUE_KEY, message);
                System.out.println("消息超时重试：" + message);
            }
        }
    }
}

// 生产者批量发送
jedis.lpush(QUEUE_KEY, "msg1", "msg2", "msg3");
// 消费者批量获取(非阻塞)
List<String> batch = jedis.rpop(QUEUE_KEY, 10); // 获取最多10条

try (Pipeline p = jedis.pipelined()) {
    p.lpush(QUEUE_KEY, "msg1");
    p.lpush(QUEUE_KEY, "msg2");
    p.sync(); // 批量提交
}

# 监控队列长度
redis-cli llen redis:mq:list:order

# 监控Redis内存
redis-cli info memory

import redis.clients.jedis.Jedis;

public class PubSubProducer {
    // 定义频道名称，采用命名空间方式避免冲突
    private static final String CHANNEL_KEY = "redis:mq:pubsub:news"; 
    // 创建Redis连接实例
    private static final Jedis jedis = new Jedis("localhost", 6379);

    public static void main(String[] args) throws InterruptedException {
        // 模拟发布3条新闻消息，实际应用中可接入实时数据源
        String[] news = {
            "Redis 7.2版本发布，新增Stream增强功能",
            "基于Redis的消息队列在电商场景的实践",
            "Redis Cluster集群部署最佳实践"
        };

        // 循环发布消息
        for (String msg : news) {
            // 发布消息并获取接收者数量
            long receiverCount = jedis.publish(CHANNEL_KEY, msg);
            System.out.println(String.format(
                "【生产者】发布消息：%s，当前订阅者数量：%d", 
                msg, receiverCount));
            // 模拟消息间隔
            Thread.sleep(1000);
        }
        
        // 关闭连接
        jedis.close();
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class PubSubConsumer {
    private static final String CHANNEL_KEY = "redis:mq:pubsub:news";
    private static final Jedis jedis = new Jedis("localhost", 6379);

    public static void main(String[] args) {
        // 创建自定义的消息处理器
        JedisPubSub pubSub = new JedisPubSub() {
            // 接收到消息时的回调方法
            @Override
            public void onMessage(String channel, String message) {
                System.out.println(String.format(
                    "【消费者1】接收到新消息（频道：%s）：%s", 
                    channel, message));
                // 此处可添加业务处理逻辑
                // 例如：解析消息内容、写入数据库、触发其他操作等
            }

            // 成功订阅频道时的回调
            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                System.out.println(String.format(
                    "【消费者1】成功订阅频道：%s，当前订阅总数：%d", 
                    channel, subscribedChannels));
            }
            
            // 可添加其他回调方法如onUnsubscribe、onPSubscribe等
        };

        System.out.println("【消费者1】启动并开始监听...");
        // 开始订阅（该方法会阻塞当前线程）
        jedis.subscribe(pubSub, CHANNEL_KEY);
        
        // 注意：在实际应用中，通常会将订阅逻辑放在独立线程中
        // 以避免阻塞主线程
    }
}

// Maven依赖
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.3.1</version>
</dependency>
// 连接配置
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxTotal(10);
try (JedisPool pool = new JedisPool(config, "localhost", 6379)) {
    Jedis jedis = pool.getResource();
    // 业务代码...
}

public class EnhancedProducer {
    private static final String[] ORDER_STATUS = {"PENDING", "PAID", "SHIPPED", "COMPLETED"};
    public void sendOrderEvent(Order order) {
        try (Jedis jedis = pool.getResource()) {
            Map<String, String> fields = new HashMap<>();
            fields.put("order_id", order.getId());
            fields.put("user_id", order.getUserId());
            fields.put("amount", order.getAmount().toString());
            fields.put("status", ORDER_STATUS[0]);
            fields.put("create_time", Instant.now().toString());
            // 使用事务保证原子性
            Transaction t = jedis.multi();
            t.xadd(STREAM_KEY, StreamEntryID.NEW_ENTRY, fields);
            t.sadd("order:ids", order.getId()); // 记录订单ID集合
            t.exec();
            // 添加监控埋点
            Metrics.counter("mq.produce.count").increment();
        }
    }
}

public class ReliableConsumer implements Runnable {
    private static final int MAX_RETRY = 3;
    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                Map<String, List<StreamEntry>> messages = jedis.xreadGroup(
                    GROUP_NAME, consumerName,
                    new StreamParams().count(1).block(2000),
                    new StreamOffset(STREAM_KEY, ">")
                );
                if (messages != null) {
                    messages.forEach((stream, entries) -> {
                        entries.forEach(entry -> {
                            processWithRetry(entry);
                        });
                    });
                }
            } catch (Exception e) {
                logger.error("消费异常", e);
                sleep(1000);
            }
        }
    }
    private void processWithRetry(StreamEntry entry) {
        int retryCount = getRetryCount(entry.getID());
        if (retryCount >= MAX_RETRY) {
            moveToDeadLetter(entry);
            return;
        }
        try {
            Order order = parseOrder(entry.getFields());
            orderService.process(order);
            jedis.xack(STREAM_KEY, GROUP_NAME, entry.getID());
        } catch (Exception e) {
            logger.warn("处理失败准备重试", e);
            sleep(1000 * (retryCount + 1));
        }
    }
}

public class DeadLetterMonitor {
    public void checkPendingMessages() {
        // 获取所有超时未确认的消息
        List<StreamEntry> pending = getPendingMessages(TIMEOUT_MS);
        pending.forEach(entry -> {
            // 检查重试次数
            if (getRetryCount(entry.getID()) > MAX_RETRY) {
                // 转移到死信队列
                jedis.xadd(DEAD_STREAM_KEY, StreamEntryID.NEW_ENTRY, entry.getFields());
                jedis.xack(STREAM_KEY, GROUP_NAME, entry.getID());
                logger.warn("消息转入死信队列: {}", entry.getID());
                // 发送告警通知
                alertService.notifyAdmin(entry);
            }
        });
    }
    public void reprocessDeadLetters() {
        // 从死信队列重新处理
        List<StreamEntry> deadMessages = jedis.xrange(DEAD_STREAM_KEY, "-", "+");
        deadMessages.forEach(entry -> {
            try {
                manualProcess(entry.getFields());
                jedis.xdel(DEAD_STREAM_KEY, entry.getID());
            } catch (Exception e) {
                logger.error("死信处理失败", e);
            }
        });
    }
}

// 批量消费提高吞吐量
jedis.xreadGroup(GROUP_NAME, consumerName, 
    new StreamParams().count(100).block(1000),
    new StreamOffset(STREAM_KEY, ">"));

// 批量确认减少网络开销
jedis.xack(STREAM_KEY, GROUP_NAME, id1, id2, id3);

// 消费者崩溃后的恢复处理
public void recoverConsumer(String failedConsumer) {
    List<PendingEntry> pendings = jedis.xpending(
        STREAM_KEY, GROUP_NAME, "-", "+", 100, failedConsumer);
    pendings.forEach(pending -> {
        jedis.xclaim(STREAM_KEY, GROUP_NAME, currentConsumer, 
            TIMEOUT_MS, pending.getIdAsString());
    });
}

// 订单服务（生产者）发送消息 - 增强版
public void createOrder(Order order) {
    // 1. 数据库事务确保数据一致性
    TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
    try {
        // 1.1 保存主订单
        orderMapper.insert(order);
        // 1.2 保存订单明细
        order.getItems().forEach(item -> {
            item.setOrderId(order.getId());
            orderItemMapper.insert(item);
        });
        // 2. 构建消息体（添加时间戳和业务标识）
        Map<String, String> message = new HashMap<>();
        message.put("order_id", order.getId());
        message.put("user_id", order.getUserId());
        message.put("product_id", order.getProductId()); 
        message.put("quantity", order.getQuantity() + "");
        message.put("create_time", System.currentTimeMillis() + "");
        message.put("biz_type", "NORMAL_ORDER");
        // 3. 发送消息（添加重试机制）
        int retryTimes = 0;
        while (retryTimes < 3) {
            try {
                jedis.xadd("redis:mq:stream:order_create", null, message);
                break;
            } catch (Exception e) {
                retryTimes++;
                if (retryTimes == 3) {
                    throw new RuntimeException("消息发送失败", e);
                }
                Thread.sleep(1000 * retryTimes);
            }
        }
        transactionManager.commit(status);
    } catch (Exception e) {
        transactionManager.rollback(status);
        throw new BusinessException("订单创建失败", e);
    }
}

// 通知服务（消费者）完整实现
public void handleNotification() {
    // 初始化消费者组（幂等操作）
    initConsumerGroup("redis:mq:stream:order_create", "order_group");
    while (!Thread.currentThread().isInterrupted()) {
        try {
            Map<String, List<StreamEntry>> messages = jedis.xreadGroup(
                "order_group", 
                "notify_consumer_" + instanceId, // 使用实例ID区分消费者
                1, 
                5000, 
                false,
                Map.of("redis:mq:stream:order_create", StreamEntryID.UNRECEIVED_ENTRY)
            );
            if (messages != null && !messages.isEmpty()) {
                for (StreamEntry entry : messages.get("redis:mq:stream:order_create")) {
                    Map<String, String> content = entry.getFields();
                    String userId = content.get("user_id");
                    String orderId = content.get("order_id");
                    // 1. 发送短信（带熔断机制）
                    boolean smsSent = circuitBreaker.execute(() -> 
                        smsService.send(userId, "订单通知", "您的订单" + orderId + "已创建成功"));
                    // 2. 发送APP推送
                    boolean pushSent = pushService.send(userId, "订单创建通知", 
                        Map.of("orderId", orderId, "type", "order_created"));
                    if (smsSent || pushSent) {
                        // 至少一个通知发送成功才确认消息
                        jedis.xack("redis:mq:stream:order_create", "order_group", entry.getID());
                        monitorService.recordSuccess("order_notify");
                    } else {
                        monitorService.recordFailure("order_notify");
                    }
                }
            }
        } catch (Exception e) {
            log.error("通知处理异常", e);
            monitorService.recordError("order_notify", e);
            Thread.sleep(5000); // 异常休眠避免循环异常
        }
    }
}
private void initConsumerGroup(String streamKey, String groupName) {
    try {
        jedis.xgroupCreate(streamKey, groupName, StreamEntryID.LAST_ENTRY, true);
    } catch (RedisBusyException e) {
        log.info("消费者组已存在: {}", groupName);
    }
}

// 库存服务（消费者）完整实现 
public void handleInventory() {
    // 初始化消费者组
    initConsumerGroup("redis:mq:stream:order_create", "order_group");
    while (!Thread.currentThread().isInterrupted()) {
        try {
            Map<String, List<StreamEntry>> messages = jedis.xreadGroup(
                "order_group",
                "inventory_consumer_" + instanceId,
                1,
                5000,
                false,
                Map.of("redis:mq:stream:order_create", StreamEntryID.UNRECEIVED_ENTRY)
            );
            if (messages != null && !messages.isEmpty()) {
                for (StreamEntry entry : messages.get("redis:mq:stream:order_create")) {
                    Map<String, String> content = entry.getFields();
                    String productId = content.get("product_id");
                    int quantity = Integer.parseInt(content.get("quantity"));
                    String orderId = content.get("order_id");
                    // 1. 扣减库存（带事务）
                    boolean success = inventoryService.deductWithLog(
                        productId, 
                        quantity,
                        orderId,
                        "ORDER_DEDUCTION"
                    );
                    if (success) {
                        // 2. 确认消息
                        jedis.xack("redis:mq:stream:order_create", "order_group", entry.getID());
                        monitorService.recordSuccess("inventory_deduct");
                    } else {
                        // 3. 记录失败日志
                        log.warn("库存扣减失败 orderId={}, productId={}", orderId, productId);
                        monitorService.recordFailure("inventory_deduct");
                        // 不确认消息，让其进入Pending状态
                    }
                }
            }
        } catch (Exception e) {
            log.error("库存处理异常", e);
            monitorService.recordError("inventory_deduct", e);
            Thread.sleep(5000);
        }
    }
}
// 库存扣减服务方法
@Transactional
public boolean deductWithLog(String productId, int quantity, String bizId, String bizType) {
    // 1. 扣减库存
    int affected = inventoryMapper.deductWithVersion(
        productId, 
        quantity,
        getCurrentVersion(productId)
    );
    if (affected == 0) {
        return false;
    }
    // 2. 记录操作流水
    InventoryLog log = new InventoryLog();
    log.setLogId(UUID.randomUUID().toString());
    log.setProductId(productId);
    log.setChangedAmount(-quantity);
    log.setBizId(bizId);
    log.setBizType(bizType);
    log.setRemarks("订单扣减");
    inventoryLogMapper.insert(log);
    return true;
}