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;
}