<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>4.4.3</version> <!-- 建議使用最新穩(wěn)定版 -->
</dependency>

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
public class ListMQProducer {
    // 隊列key命名規(guī)范：業(yè)務域:組件類型:數(shù)據(jù)結構:具體業(yè)務
    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()) {
            // 模擬發(fā)送10條訂單消息
            for (int i = 1; i <= 10; i++) {
                // 消息內(nèi)容格式：業(yè)務標識_序號_時間戳
                String message = String.format("order_%d_%d", i, System.currentTimeMillis());
                // LPUSH命令將消息放入隊列頭部
                long queueLength = jedis.lpush(QUEUE_KEY, message);
                System.out.printf("生產(chǎn)者發(fā)送消息：%s，當前隊列長度：%d%n", message, queueLength);
                // 模擬業(yè)務處理間隔
                Thread.sleep(500); 
            }
        } catch (Exception e) {
            System.err.println("生產(chǎn)者異常：" + 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命令參數(shù)：
                // 1. 超時時間3秒(避免空輪詢消耗CPU)
                // 2. 可以監(jiān)聽多個隊列
                List<String> messages = jedis.brpop(3, QUEUE_KEY);
                if (messages != null) {
                    // BRPOP返回結果格式：
                    // 第一個元素是隊列key
                    // 第二個元素是消息內(nèi)容
                    String message = messages.get(1);
                    System.out.println("消費者接收消息：" + message);
                    // 業(yè)務處理邏輯示例
                    processMessage(message);
                } else {
                    System.out.println("隊列暫無消息，繼續(xù)等待...");
                }
            } 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 {
        // 模擬業(yè)務處理
        System.out.println("處理消息：" + message);
        // 解析消息內(nèi)容
        String[] parts = message.split("_");
        String orderId = parts[1];
        // 模擬業(yè)務處理耗時
        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;
// 優(yōu)化后的消費者處理邏輯
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. 處理業(yè)務邏輯
        processMessage(message);
        // 3. 處理成功，從待確認隊列刪除
        jedis.lrem(CONFIRM_QUEUE_KEY, 1, message);
    } catch (Exception e) {
        System.err.println("處理消息失?。? + message);
        // 4. 檢查重試次數(shù)
        long retryCount = jedis.incr("retry:" + message);
        if (retryCount <= MAX_RETRY) {
            // 放回主隊列重試
            jedis.lpush(QUEUE_KEY, message);
        } else {
            // 超過重試次數(shù)，放入死信隊列
            jedis.rpush(DEAD_QUEUE_KEY, message);
        }
        // 無論重試還是加入死信隊列，都要從待確認隊列刪除
        jedis.lrem(CONFIRM_QUEUE_KEY, 1, message);
    }
}

// 定時檢查待確認隊列(每分鐘執(zhí)行)
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);
            }
        }
    }
}

// 生產(chǎn)者批量發(fā)送
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(); // 批量提交
}

# 監(jiān)控隊列長度
redis-cli llen redis:mq:list:order

# 監(jiān)控Redis內(nèi)存
redis-cli info memory

import redis.clients.jedis.Jedis;

public class PubSubProducer {
    // 定義頻道名稱，采用命名空間方式避免沖突
    private static final String CHANNEL_KEY = "redis:mq:pubsub:news"; 
    // 創(chuàng)建Redis連接實例
    private static final Jedis jedis = new Jedis("localhost", 6379);

    public static void main(String[] args) throws InterruptedException {
        // 模擬發(fā)布3條新聞消息，實際應用中可接入實時數(shù)據(jù)源
        String[] news = {
            "Redis 7.2版本發(fā)布，新增Stream增強功能",
            "基于Redis的消息隊列在電商場景的實踐",
            "Redis Cluster集群部署最佳實踐"
        };

        // 循環(huán)發(fā)布消息
        for (String msg : news) {
            // 發(fā)布消息并獲取接收者數(shù)量
            long receiverCount = jedis.publish(CHANNEL_KEY, msg);
            System.out.println(String.format(
                "【生產(chǎn)者】發(fā)布消息：%s，當前訂閱者數(shù)量：%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) {
        // 創(chuàng)建自定義的消息處理器
        JedisPubSub pubSub = new JedisPubSub() {
            // 接收到消息時的回調(diào)方法
            @Override
            public void onMessage(String channel, String message) {
                System.out.println(String.format(
                    "【消費者1】接收到新消息（頻道：%s）：%s", 
                    channel, message));
                // 此處可添加業(yè)務處理邏輯
                // 例如：解析消息內(nèi)容、寫入數(shù)據(jù)庫、觸發(fā)其他操作等
            }

            // 成功訂閱頻道時的回調(diào)
            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                System.out.println(String.format(
                    "【消費者1】成功訂閱頻道：%s，當前訂閱總數(shù)：%d", 
                    channel, subscribedChannels));
            }
            
            // 可添加其他回調(diào)方法如onUnsubscribe、onPSubscribe等
        };

        System.out.println("【消費者1】啟動并開始監(jiān)聽...");
        // 開始訂閱（該方法會阻塞當前線程）
        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();
    // 業(yè)務代碼...
}

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();
            // 添加監(jiān)控埋點
            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 -> {
            // 檢查重試次數(shù)
            if (getRetryCount(entry.getID()) > MAX_RETRY) {
                // 轉(zhuǎn)移到死信隊列
                jedis.xadd(DEAD_STREAM_KEY, StreamEntryID.NEW_ENTRY, entry.getFields());
                jedis.xack(STREAM_KEY, GROUP_NAME, entry.getID());
                logger.warn("消息轉(zhuǎn)入死信隊列: {}", entry.getID());
                // 發(fā)送告警通知
                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, ">"));

// 批量確認減少網(wǎng)絡開銷
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());
    });
}

// 訂單服務（生產(chǎn)者）發(fā)送消息 - 增強版
public void createOrder(Order order) {
    // 1. 數(shù)據(jù)庫事務確保數(shù)據(jù)一致性
    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. 構建消息體（添加時間戳和業(yè)務標識）
        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. 發(fā)送消息（添加重試機制）
        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("消息發(fā)送失敗", e);
                }
                Thread.sleep(1000 * retryTimes);
            }
        }
        transactionManager.commit(status);
    } catch (Exception e) {
        transactionManager.rollback(status);
        throw new BusinessException("訂單創(chuàng)建失敗", e);
    }
}

// 通知服務（消費者）完整實現(xiàn)
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區(qū)分消費者
                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. 發(fā)送短信（帶熔斷機制）
                    boolean smsSent = circuitBreaker.execute(() -> 
                        smsService.send(userId, "訂單通知", "您的訂單" + orderId + "已創(chuàng)建成功"));
                    // 2. 發(fā)送APP推送
                    boolean pushSent = pushService.send(userId, "訂單創(chuàng)建通知", 
                        Map.of("orderId", orderId, "type", "order_created"));
                    if (smsSent || pushSent) {
                        // 至少一個通知發(fā)送成功才確認消息
                        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); // 異常休眠避免循環(huán)異常
        }
    }
}
private void initConsumerGroup(String streamKey, String groupName) {
    try {
        jedis.xgroupCreate(streamKey, groupName, StreamEntryID.LAST_ENTRY, true);
    } catch (RedisBusyException e) {
        log.info("消費者組已存在: {}", groupName);
    }
}

// 庫存服務（消費者）完整實現(xiàn) 
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狀態(tài)
                    }
                }
            }
        } 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;
}