第三章 消息发送与消费 v1.0

第一部分 Spring Boot集成

一、添加依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.2</version>
</dependency>

二、配置文件

rocketmq:
  name-server: localhost:9876
  producer:
    group: my-producer-group
    send-message-timeout: 3000
    retry-times-when-send-failed: 2

第二部分 消息发送

一、同步发送

同步发送会等待Broker返回发送结果，可靠性高，适用于重要消息。

@Autowired
private RocketMQTemplate rocketMQTemplate;

public void sendSyncMessage() {
    Message<String> message = MessageBuilder.withPayload("Hello RocketMQ").build();
    SendResult result = rocketMQTemplate.syncSend("topic-name", message);
    System.out.println("发送结果: " + result.getSendStatus());
}

二、异步发送

异步发送通过回调处理结果，性能高，适用于对响应时间敏感的场景。

public void sendAsyncMessage() {
    Message<String> message = MessageBuilder.withPayload("Hello RocketMQ").build();
    rocketMQTemplate.asyncSend("topic-name", message, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("发送成功");
        }
        @Override
        public void onException(Throwable e) {
            System.out.println("发送失败: " + e.getMessage());
        }
    });
}

三、单向发送

单向发送不等待Broker响应，性能最高，适用于日志等不重要的消息。

public void sendOneWayMessage() {
    Message<String> message = MessageBuilder.withPayload("Hello RocketMQ").build();
    rocketMQTemplate.sendOneWay("topic-name", message);
}

四、顺序消息

public void sendOrderlyMessage() {
    Message<String> message = MessageBuilder.withPayload("Hello RocketMQ").build();
    // 使用orderId作为分区Key，保证同一订单的消息顺序消费
    rocketMQTemplate.syncSendOrderly("topic-name", message, "orderId");
}

五、延时消息

public void sendDelayedMessage() {
    Message<String> message = MessageBuilder.withPayload("Hello RocketMQ").build();
    // 延时3秒投递
    rocketMQTemplate.syncSend("topic-name", message, 3000);
}

六、批量消息

public void sendBatchMessage() {
    List<Message> messages = new ArrayList<>();
    messages.add(MessageBuilder.withPayload("Message1").build());
    messages.add(MessageBuilder.withPayload("Message2").build());
    messages.add(MessageBuilder.withPayload("Message3").build());
    
    rocketMQTemplate.syncSend("topic-name", messages);
}

第三部分 消息消费

一、基本消费

@Component
@RocketMQMessageListener(
    topic = "topic-name",
    consumerGroup = "my-consumer-group"
)
public class MyConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息: " + message);
    }
}

二、消费Tag过滤

@Component
@RocketMQMessageListener(
    topic = "topic-name",
    consumerGroup = "my-consumer-group",
    selectorExpression = "tag1 || tag2"  // 只消费tag1和tag2的消息
)
public class TagConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息: " + message);
    }
}

三、顺序消费

@Component
@RocketMQMessageListener(
    topic = "topic-name",
    consumerGroup = "my-consumer-group",
    consumeMode = ConsumeMode.ORDERLY  // 顺序消费模式
)
public class OrderlyConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        System.out.println("顺序消费: " + message);
    }
}

四、广播消费

@Component
@RocketMQMessageListener(
    topic = "topic-name",
    consumerGroup = "my-consumer-group",
    messageModel = MessageModel.BROADCASTING  // 广播消费模式
)
public class BroadcastConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        System.out.println("广播消费: " + message);
    }
}

五、消费重试与死信队列

消费失败时，RocketMQ会自动重试。超过重试次数后，消息会被投递到死信队列（DLQ）。

@Component
@RocketMQMessageListener(
    topic = "topic-name",
    consumerGroup = "my-consumer-group"
)
public class RetryConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        try {
            // 业务处理
            processMessage(message);
        } catch (Exception e) {
            // 抛出异常触发重试
            throw new RuntimeException("消费失败", e);
        }
    }
}

第四部分 消息查询

一、按MessageId查询

public MessageExt queryByMessageId(String messageId) throws Exception {
    return rocketMQTemplate.getProducer().queryMessage("topic-name", messageId).getMessageExt();
}

二、按Key查询

public List<MessageExt> queryByKey(String key) throws Exception {
    QueryResult result = rocketMQTemplate.getProducer().queryMessage("topic-name", key, 32, 0, System.currentTimeMillis());
    return result.getMessageList();
}