Java案例怎么实现消息队列？

wen java案例 2026-06-07 79

Java案例怎么实现消息队列？实战全流程解析

📖 目录导读

为什么Java项目需要消息队列？
消息队列的核心概念与选型
基于Java实现消息队列的三种经典方案
使用Redis List实现轻量级消息队列（含代码）
整合Apache Kafka实现高吞吐消息队列（含代码）
基于RabbitMQ的AMQP消息队列（含代码）
常见坑与性能优化问答

为什么Java项目需要消息队列？

在分布式系统或高并发场景中,直接调用API会导致服务雪崩，用户下单后需要发送短信、记录日志、更新积分，如果同步执行，响应时间会被最慢的操作拖垮。

Java案例怎么实现消息队列？

消息队列（Message Queue） 的核心价值在于：

解耦：生产者与消费者独立部署
削峰填谷：缓冲突发流量
异步处理：提升系统吞吐量

案例场景：假设你的订单系统每秒需处理1000个请求，后端数据库只能承受200 QPS，引入MQ后，请求先写入队列，消费者按能力拉取，系统瞬间平稳。

消息队列的核心概念与选型

在实现MQ前,必须理解三个核心角色：

Producer：发送消息的Java服务
Broker：消息中间件（如Kafka、Redis）
Consumer：接收并处理消息的服务

选型建议（基于Java生态）： | 方案 | 适用场景 | 优势 | 劣势 | |------------|------------------------|------------------------|--------------------| | Redis List | 轻量级、临时小流量 | 极简部署，无需独立中间件 | 无持久化保障 | | Kafka | 大数据流、日志采集 | 超高吞吐（百万级/秒） | 运维复杂，延迟高 | | RabbitMQ | 事务性业务、可靠投递 | 消息确认机制成熟 | 吞吐低于Kafka |

基于Java实现消息队列的三种经典方案

下面分别用Spring Boot + Redis、Kafka、RabbitMQ演示核心代码，已去除冗余注释，保留关键逻辑。

使用Redis List实现轻量级MQ

核心原理：利用LPUSH生产消息，BRPOP阻塞消费。

// 消息生产者
@Component
public class RedisProducer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    public void send(String queueKey, String message) {
        redisTemplate.opsForList().leftPush(queueKey, message);
        log.info("消息已推入队列: {}", queueKey);
    }
}
// 消息消费者（持续监听）
@Component
public class RedisConsumer implements CommandLineRunner {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Override
    public void run(String... args) {
        new Thread(() -> {
            while (true) {
                try {
                    String message = redisTemplate.opsForList()
                        .rightPop("order_queue", 5, TimeUnit.SECONDS);
                    if (message != null) {
                        System.out.println("消费消息: " + message);
                        // 业务处理：更新订单状态、写日志等
                    }
                } catch (Exception e) {
                    log.error("消费异常", e);
                }
            }
        }).start();
    }
}

优缺点：零依赖，但消息可能丢失（Redis宕机），适合测试环境或非关键业务。

整合Apache Kafka实现高吞吐MQ

步骤：引入依赖 → 配置生产者/消费者 → 发送/接收

pom.xml：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.8.0</version>
</dependency>

Kafka生产者：

@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendOrderMessage(String orderId) {
        kafkaTemplate.send("order-topic", orderId);
        System.out.println("订单消息投递至Kafka: " + orderId);
    }
}

Kafka消费者（配合@KafkaListener）：

@Component
public class KafkaConsumer {
    @KafkaListener(topics = "order-topic", groupId = "order-group")
    public void handleOrder(String orderId) {
        System.out.println("Kafka消费订单: " + orderId);
        // 调用积分服务、短信服务等
    }
}

性能要点：batch.size和linger.ms控制吞吐与延迟的平衡。

基于RabbitMQ的AMQP消息队列

步骤：配置交换机、队列、绑定关系。

生产者代码：

@Service
public class RabbitMQProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send(String message) {
        rabbitTemplate.convertAndSend("order.exchange", "order.routingKey", message);
        log.info("RabbitMQ消息已发送");
    }
}

消费者代码（支持死信队列与手动ACK）：

@Component
@RabbitListener(queues = "order.queue")
public class RabbitMQConsumer {
    @RabbitHandler
    public void process(String msg, Channel channel, Message message) {
        try {
            System.out.println("消费者收到: " + msg);
            // 业务处理
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 拒绝并重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

可靠性增强：设置mandatory=true监听消息丢失，配合死信队列兜底。