IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java消息发送案例怎么编写

wen java案例 22

高效掌握Java消息发送案例：从入门到实战的完整指南

目录导读

  1. 为什么需要消息发送机制？
  2. 主流消息中间件对比与选型
  3. Java消息发送核心概念解析
  4. 基于RabbitMQ的实战案例（含代码）
  5. 基于Kafka的高吞吐量案例
  6. 常见问题与最佳实践
  7. 总结与进阶建议

问答前置：什么是消息发送机制？

问：消息发送机制解决了什么问题？
答：消息发送（Message Queuing）是一种异步通信模式，主要用于解耦系统组件、缓解瞬时请求压力、实现流量削峰填谷，电商订单系统创建订单后，需要通知物流、库存、积分等多个子系统，若使用同步调用，任何一个子系统故障都会导致主流程失败，而消息中间件能将这些任务异步处理，提升系统稳定性和扩展性。

Java消息发送案例怎么编写

问：Java开发者需要掌握哪些消息中间件？
答：目前工业界主流的有RabbitMQ、Apache Kafka、RocketMQ、ActiveMQ等，其中RabbitMQ适合中小规模、对可靠性要求高的业务；Kafka适合海量日志、数据流处理场景；RocketMQ是阿里开源，在金融和电商领域表现优秀。

为什么需要消息发送机制？

在微服务架构中,服务间通信通常采用同步RPC或异步消息，同步调用会导致调用方阻塞等待，且强耦合，而消息发送机制具有以下关键优势：

  1. 解耦：生产者无需关心消费者是谁，只需将消息投递到中间件。
  2. 异步：生产者不阻塞，可立即返回，提升系统响应速度。
  3. 削峰：将突发流量暂存在消息队列中，下游按能力消费。
  4. 冗余：消息可持久化，防止数据丢失。

主流消息中间件对比与选型

特性 RabbitMQ Apache Kafka RocketMQ
协议 AMQP 自定义TCP 自定义TCP
吞吐量 万级/s 百万级/s 十万级/s
消息顺序 单队列内顺序 分区内顺序 分区内顺序
消息确认 支持 通过偏移量控制 支持
管理界面 自带 需第三方 自带

选型建议：若团队对延迟敏感且需要灵活路由，选RabbitMQ；若处理日志、埋点等高吞吐数据，选Kafka；若业务场景复杂且需要事务消息，选RocketMQ。

Java消息发送核心概念解析

无论使用哪种中间件,核心概念都类似：

  • Producer（生产者）：负责创建并发送消息。
  • Broker（代理）：消息中间件服务器，存储和转发消息。
  • Queue（队列）：消息存储的容器。
  • Consumer（消费者）：从队列中拉取或订阅消息。
  • Exchange与Routing（交换机与路由）：RabbitMQ特有，通过路由规则分发消息。

一个典型的流程：Producer发送消息到Exchange → Exchange根据Routing Key将消息路由到Queue → Consumer从Queue拉取消息。

基于RabbitMQ的实战案例（完整代码）

1 环境准备

<!-- Maven依赖 -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.18.0</version>
</dependency>

2 发送消息代码（生产者）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class MessageProducer {
    private final static String QUEUE_NAME = "demo_queue";
    public static void main(String[] args) throws Exception {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        // 2. 创建连接与通道
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 3. 声明队列（若不存在则创建）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 4. 发送消息
            String message = "Hello, RabbitMQ! 当前时间: " + System.currentTimeMillis();
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] 发送消息: '" + message + "'");
        }
    }
}

3 接收消息代码（消费者）

import com.rabbitmq.client.*;
public class MessageConsumer {
    private final static String QUEUE_NAME = "demo_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] 等待消息...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] 收到消息: '" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
}

注意事项：生产环境需配置连接池、消息持久化、手动ACK（将autoAck设为false）以防止消息丢失。

基于Kafka的高吞吐量案例

1 环境依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.1</version>
</dependency>

2 生产者代码

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaMsgProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // 所有副本确认
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        String topic = "demo-topic";
        String key = "order-001";
        String value = "用户下单成功，金额: 199.00";
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(topic, key, value);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("发送成功: offset=" + metadata.offset() + 
                                   ", partition=" + metadata.partition());
            } else {
                System.err.println("发送失败: " + exception.getMessage());
            }
        });
        producer.close();
    }
}

3 消费者代码

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaMsgConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("demo-topic"));
        while (true) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("消费消息: key=%s, value=%s, offset=%d%n",
                                  record.key(), record.value(), record.offset());
            }
            consumer.commitSync(); // 手动提交偏移量
        }
    }
}

常见问题与最佳实践

Q1: 消息发送后消费者没收到怎么办？
A: 检查网络连接、队列是否存在、消费者是否订阅正确队列，Kafka还需检查消费者组偏移量。

Q2: 如何保证消息不丢失？
A: 生产者设置acks=all或mandatory=true；消息持久化（RabbitMQ设置MessageProperties.PERSISTENT_TEXT_PLAIN）；消费者使用手动ACK。

Q3: 消息重复消费怎么办？
A: 消费端做幂等处理，例如使用数据库唯一索引、Redis去重或业务主键判断。

Q4: 生产环境中使用什么连接方式？
A: 建议使用连接池（如PooledRabbitMQ或KafkaClient的Producer池），避免每次请求都创建新连接。

总结与进阶建议

本文通过Java编写了两个主流消息中间件的发送案例：RabbitMQ适合中小型项目，配置直观，管理方便；Kafka适合大数据流场景，吞吐量极高，初学者可先从RabbitMQ入手，理解JMS/AMQP协议，再过渡到Kafka。

进阶学习方向：

  • 了解Spring Boot集成（spring-boot-starter-amqp / spring-kafka）
  • 掌握消息幂等性设计模式
  • 深入学习消息顺序性与事务消息
  • 调研云原生产品（如Pulsar）的原理

通过动手运行本文案例,你将快速掌握消息发送机制的实战要点，在实际项目中，建议结合业务场景选择合适中间件，并做好监控与告警（如消息积压量、消费延迟等），确保系统稳定运行。

上一篇Java消息接收案例如何实现

下一篇Java RabbitMQ案例如何Java对接

相关文章

抱歉，评论功能暂时关闭!