第123集RocketMQ消息队列搭建与开发运维实战

package com.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

/**
 * 消息生产者
 * @author 运维实战
 */
@Slf4j
@Component
public class MessageProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    /**
     * 发送同步消息
     */
    public void sendSyncMessage(String topic, String message) {
        try {
            log.info("发送同步消息: topic={}, message={}", topic, message);
            
            // 发送同步消息
            rocketMQTemplate.syncSend(topic, message);
            
            log.info("同步消息发送成功: topic={}", topic);
            
        } catch (Exception e) {
            log.error("同步消息发送失败: topic={}", topic, e);
            throw new RuntimeException("同步消息发送失败", e);
        }
    }
    
    /**
     * 发送异步消息
     */
    public void sendAsyncMessage(String topic, String message) {
        try {
            log.info("发送异步消息: topic={}, message={}", topic, message);
            
            // 发送异步消息
            rocketMQTemplate.asyncSend(topic, message, new org.apache.rocketmq.client.producer.SendCallback() {
                @Override
                public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
                    log.info("异步消息发送成功: topic={}, msgId={}", topic, sendResult.getMsgId());
                }
                
                @Override
                public void onException(Throwable e) {
                    log.error("异步消息发送失败: topic={}", topic, e);
                }
            });
            
        } catch (Exception e) {
            log.error("异步消息发送失败: topic={}", topic, e);
            throw new RuntimeException("异步消息发送失败", e);
        }
    }
    
    /**
     * 发送单向消息
     */
    public void sendOneWayMessage(String topic, String message) {
        try {
            log.info("发送单向消息: topic={}, message={}", topic, message);
            
            // 发送单向消息
            rocketMQTemplate.sendOneWay(topic, message);
            
            log.info("单向消息发送成功: topic={}", topic);
            
        } catch (Exception e) {
            log.error("单向消息发送失败: topic={}", topic, e);
            throw new RuntimeException("单向消息发送失败", e);
        }
    }
    
    /**
     * 发送延迟消息
     */
    public void sendDelayMessage(String topic, String message, int delayLevel) {
        try {
            log.info("发送延迟消息: topic={}, message={}, delayLevel={}", topic, message, delayLevel);
            
            // 发送延迟消息
            rocketMQTemplate.syncSend(topic, message, 3000, delayLevel);
            
            log.info("延迟消息发送成功: topic={}, delayLevel={}", topic, delayLevel);
            
        } catch (Exception e) {
            log.error("延迟消息发送失败: topic={}", topic, e);
            throw new RuntimeException("延迟消息发送失败", e);
        }
    }
    
    /**
     * 发送顺序消息
     */
    public void sendOrderMessage(String topic, String message, String orderKey) {
        try {
            log.info("发送顺序消息: topic={}, message={}, orderKey={}", topic, message, orderKey);
            
            // 发送顺序消息
            rocketMQTemplate.syncSendOrderly(topic, message, orderKey);
            
            log.info("顺序消息发送成功: topic={}, orderKey={}", topic, orderKey);
            
        } catch (Exception e) {
            log.error("顺序消息发送失败: topic={}", topic, e);
            throw new RuntimeException("顺序消息发送失败", e);
        }
    }
    
    /**
     * 发送事务消息
     */
    public void sendTransactionMessage(String topic, String message) {
        try {
            log.info("发送事务消息: topic={}, message={}", topic, message);
            
            // 发送事务消息
            rocketMQTemplate.sendMessageInTransaction(topic, message, null);
            
            log.info("事务消息发送成功: topic={}", topic);
            
        } catch (Exception e) {
            log.error("事务消息发送失败: topic={}", topic, e);
            throw new RuntimeException("事务消息发送失败", e);
        }
    }
    
    /**
     * 发送批量消息
     */
    public void sendBatchMessage(String topic, String[] messages) {
        try {
            log.info("发送批量消息: topic={}, count={}", topic, messages.length);
            
            // 发送批量消息
            for (String message : messages) {
                rocketMQTemplate.syncSend(topic, message);
            }
            
            log.info("批量消息发送成功: topic={}, count={}", topic, messages.length);
            
        } catch (Exception e) {
            log.error("批量消息发送失败: topic={}", topic, e);
            throw new RuntimeException("批量消息发送失败", e);
        }
    }
    
    /**
     * 发送带标签的消息
     */
    public void sendTagMessage(String topic, String tag, String message) {
        try {
            log.info("发送带标签的消息: topic={}, tag={}, message={}", topic, tag, message);
            
            // 发送带标签的消息
            rocketMQTemplate.syncSend(topic + ":" + tag, message);
            
            log.info("带标签的消息发送成功: topic={}, tag={}", topic, tag);
            
        } catch (Exception e) {
            log.error("带标签的消息发送失败: topic={}", topic, e);
            throw new RuntimeException("带标签的消息发送失败", e);
        }
    }
}

package com.rocketmq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * 消息消费者
 * @author 运维实战
 */
@Slf4j
@Component
public class MessageConsumer {
    
    /**
     * 消费普通消息
     */
    @RocketMQMessageListener(
        topic = "demo-topic",
        consumerGroup = "demo-consumer-group"
    )
    public class DemoMessageConsumer implements RocketMQListener<String> {
        
        @Override
        public void onMessage(String message) {
            try {
                log.info("消费普通消息: {}", message);
                
                // 处理业务逻辑
                processMessage(message);
                
                log.info("普通消息处理完成: {}", message);
                
            } catch (Exception e) {
                log.error("普通消息处理失败: {}", message, e);
                throw new RuntimeException("普通消息处理失败", e);
            }
        }
    }
    
    /**
     * 消费延迟消息
     */
    @RocketMQMessageListener(
        topic = "delay-topic",
        consumerGroup = "delay-consumer-group"
    )
    public class DelayMessageConsumer implements RocketMQListener<String> {
        
        @Override
        public void onMessage(String message) {
            try {
                log.info("消费延迟消息: {}", message);
                
                // 处理延迟消息
                processDelayMessage(message);
                
                log.info("延迟消息处理完成: {}", message);
                
            } catch (Exception e) {
                log.error("延迟消息处理失败: {}", message, e);
                throw new RuntimeException("延迟消息处理失败", e);
            }
        }
    }
    
    /**
     * 消费顺序消息
     */
    @RocketMQMessageListener(
        topic = "order-topic",
        consumerGroup = "order-consumer-group",
        consumeMode = org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING,
        messageModel = org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING,
        consumeOrderly = true
    )
    public class OrderMessageConsumer implements RocketMQListener<String> {
        
        @Override
        public void onMessage(String message) {
            try {
                log.info("消费顺序消息: {}", message);
                
                // 处理顺序消息
                processOrderMessage(message);
                
                log.info("顺序消息处理完成: {}", message);
                
            } catch (Exception e) {
                log.error("顺序消息处理失败: {}", message, e);
                throw new RuntimeException("顺序消息处理失败", e);
            }
        }
    }
    
    /**
     * 消费事务消息
     */
    @RocketMQMessageListener(
        topic = "transaction-topic",
        consumerGroup = "transaction-consumer-group"
    )
    public class TransactionMessageConsumer implements RocketMQListener<String> {
        
        @Override
        public void onMessage(String message) {
            try {
                log.info("消费事务消息: {}", message);
                
                // 处理事务消息
                processTransactionMessage(message);
                
                log.info("事务消息处理完成: {}", message);
                
            } catch (Exception e) {
                log.error("事务消息处理失败: {}", message, e);
                throw new RuntimeException("事务消息处理失败", e);
            }
        }
    }
    
    /**
     * 消费带标签的消息
     */
    @RocketMQMessageListener(
        topic = "tag-topic",
        consumerGroup = "tag-consumer-group",
        selectorExpression = "tag1 || tag2"
    )
    public class TagMessageConsumer implements RocketMQListener<String> {
        
        @Override
        public void onMessage(String message) {
            try {
                log.info("消费带标签的消息: {}", message);
                
                // 处理带标签的消息
                processTagMessage(message);
                
                log.info("带标签的消息处理完成: {}", message);
                
            } catch (Exception e) {
                log.error("带标签的消息处理失败: {}", message, e);
                throw new RuntimeException("带标签的消息处理失败", e);
            }
        }
    }
    
    /**
     * 处理普通消息
     */
    private void processMessage(String message) {
        // 模拟业务处理
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("普通消息处理完成: {}", message);
    }
    
    /**
     * 处理延迟消息
     */
    private void processDelayMessage(String message) {
        // 模拟延迟消息处理
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("延迟消息处理完成: {}", message);
    }
    
    /**
     * 处理顺序消息
     */
    private void processOrderMessage(String message) {
        // 模拟顺序消息处理
        try {
            Thread.sleep(150);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("顺序消息处理完成: {}", message);
    }
    
    /**
     * 处理事务消息
     */
    private void processTransactionMessage(String message) {
        // 模拟事务消息处理
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("事务消息处理完成: {}", message);
    }
    
    /**
     * 处理带标签的消息
     */
    private void processTagMessage(String message) {
        // 模拟带标签的消息处理
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("带标签的消息处理完成: {}", message);
    }
}

package com.rocketmq.controller;

import com.rocketmq.producer.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

/**
 * RocketMQ控制器
 * @author 运维实战
 */
@Slf4j
@RestController
@RequestMapping("/api/rocketmq")
public class RocketMQController {
    
    @Autowired
    private MessageProducer messageProducer;
    
    /**
     * 发送同步消息
     */
    @PostMapping("/send/sync")
    public ResponseEntity<Map<String, Object>> sendSyncMessage(
            @RequestParam String topic,
            @RequestParam String message) {
        try {
            messageProducer.sendSyncMessage(topic, message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "同步消息发送成功");
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送同步消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送异步消息
     */
    @PostMapping("/send/async")
    public ResponseEntity<Map<String, Object>> sendAsyncMessage(
            @RequestParam String topic,
            @RequestParam String message) {
        try {
            messageProducer.sendAsyncMessage(topic, message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "异步消息发送成功");
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送异步消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送单向消息
     */
    @PostMapping("/send/oneway")
    public ResponseEntity<Map<String, Object>> sendOneWayMessage(
            @RequestParam String topic,
            @RequestParam String message) {
        try {
            messageProducer.sendOneWayMessage(topic, message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "单向消息发送成功");
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送单向消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送延迟消息
     */
    @PostMapping("/send/delay")
    public ResponseEntity<Map<String, Object>> sendDelayMessage(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam(defaultValue = "1") int delayLevel) {
        try {
            messageProducer.sendDelayMessage(topic, message, delayLevel);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "延迟消息发送成功");
            result.put("delayLevel", delayLevel);
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送延迟消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送顺序消息
     */
    @PostMapping("/send/order")
    public ResponseEntity<Map<String, Object>> sendOrderMessage(
            @RequestParam String topic,
            @RequestParam String message,
            @RequestParam String orderKey) {
        try {
            messageProducer.sendOrderMessage(topic, message, orderKey);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "顺序消息发送成功");
            result.put("orderKey", orderKey);
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送顺序消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送事务消息
     */
    @PostMapping("/send/transaction")
    public ResponseEntity<Map<String, Object>> sendTransactionMessage(
            @RequestParam String topic,
            @RequestParam String message) {
        try {
            messageProducer.sendTransactionMessage(topic, message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "事务消息发送成功");
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送事务消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送批量消息
     */
    @PostMapping("/send/batch")
    public ResponseEntity<Map<String, Object>> sendBatchMessage(
            @RequestParam String topic,
            @RequestBody String[] messages) {
        try {
            messageProducer.sendBatchMessage(topic, messages);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "批量消息发送成功");
            result.put("count", messages.length);
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送批量消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
    
    /**
     * 发送带标签的消息
     */
    @PostMapping("/send/tag")
    public ResponseEntity<Map<String, Object>> sendTagMessage(
            @RequestParam String topic,
            @RequestParam String tag,
            @RequestParam String message) {
        try {
            messageProducer.sendTagMessage(topic, tag, message);
            
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put("message", "带标签的消息发送成功");
            result.put("tag", tag);
            result.put("timestamp", LocalDateTime.now());
            
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            log.error("发送带标签的消息失败", e);
            return ResponseEntity.internalServerError().build();
        }
    }
}