RocketMQ 和 RabbitMQ 的比较以及 RocketMQ 的使用

@Slf4j
@Service
public class MessageProducerService {

    // RocketMQ消息主题
    public static final String TOPIC_UNICAST = "topic-unicast";
    public static final String TOPIC_BROADCAST = "topic-broadcast";
    public static final String TOPIC_RETRY = "topic-retry";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送单播消息（点对点）
     * 单播消息会被消费组中的某一个消费者消费
     */
    public void sendUnicastMessage(MessageEvent message) {
        rocketMQTemplate.convertAndSend(TOPIC_UNICAST, message);
        log.info("Unicast message sent: {}", message);
    }

    /**
     * 发送广播消息
     * 广播消息会被所有订阅该主题的消费者消费
     */
    public void sendBroadcastMessage(MessageEvent message) {
        rocketMQTemplate.convertAndSend(TOPIC_BROADCAST, message);
        log.info("Broadcast message sent: {}", message);
    }

    /**
     * 发送需要重试的消息
     * 使用异步发送方式，并在回调中处理发送结果
     */
    public void sendRetryMessage(MessageEvent message) {
        rocketMQTemplate.asyncSend(TOPIC_RETRY, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Retry message sent successfully: {}, result: {}", message, sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("Failed to send retry message: {}, error: {}", message, throwable.getMessage());
            }
        });
    }
}

@Slf4j
@Service
public class MessageConsumerService {

    // RocketMQ消息主题
    public static final String TOPIC_UNICAST = "topic-unicast";
    public static final String TOPIC_BROADCAST = "topic-broadcast";
    public static final String TOPIC_RETRY = "topic-retry";
    /**
     * 单播消息消费者
     * consumeMode默认为CONCURRENTLY（并发消费）
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_UNICAST,
            consumerGroup = "unicast-consumer-group"
    )
    public class UnicastMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {

            log.info("Received unicast message: {}", message);
        }
    }

    /**
     * 广播消息消费者
     * messageModel设置为BROADCASTING表示广播模式
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_BROADCAST,
            consumerGroup = "broadcast-consumer-group",
            messageModel = MessageModel.BROADCASTING
    )
    public class BroadcastMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {
            log.info("Received broadcast message: {}", message);
        }
    }

    /**
     * 重试消息消费者
     * 配置了重试次数和重试间隔
     */
    @Service
    @RocketMQMessageListener(
            topic = TOPIC_RETRY,
            consumerGroup = "retry-consumer-group",
            maxReconsumeTimes = 3,    // 最大重试次数
            delayLevelWhenNextConsume = 2  // 重试间隔级别
    )
    public class RetryMessageListener implements RocketMQListener<MessageEvent> {
        @Override
        public void onMessage(MessageEvent message) {
            try {
                // 模拟处理失败的情况
                if (message.getContent().contains("error")) {
                    throw new RuntimeException("Processing failed, will retry");
                }
                log.info("Received retry message: {}", message);
            } catch (Exception e) {
                log.error("Error processing message: {}, error: {}", message, e.getMessage());
                throw e; // 抛出异常触发重试机制
            }
        }
    }
}

@RestController
@RequestMapping("/api/messages")
public class MessageController {

    @Autowired
    private MessageProducerService producerService;

    @PostMapping("/unicast")
    public String sendUnicastMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendUnicastMessage(message);
        return "Unicast message sent successfully";
    }

    @PostMapping("/broadcast")
    public String sendBroadcastMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendBroadcastMessage(message);
        return "Broadcast message sent successfully";
    }

    @PostMapping("/retry")
    public String sendRetryMessage(@RequestParam String content) {
        MessageEvent message = new MessageEvent()
                .setId(UUID.randomUUID().toString())
                .setContent(content)
                .setTimestamp(System.currentTimeMillis());
        producerService.sendRetryMessage(message);
        return "Retry message sent successfully";
    }
}