MQ 消息队列面试题大全

// RocketMQ 批量发送
List<Message> messages = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
    messages.add(new Message("topic", "tag", ("message" + i).getBytes()));
}
SendResult result = producer.send(messages);

// Kafka 批量消费
List<ConsumerRecord<String, String>> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    // 批量处理逻辑
    processMessage(record);
}

// 批量消息发送
List<Message> messages = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
    Message msg = new Message("TopicTest", "TagA",
        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    messages.add(msg);
}
SendResult sendResult = producer.send(messages);

// 批量消费配置
consumer.setConsumeMessageBatchMaxSize(32);
consumer.setPullBatchSize(32);

// 设置延迟级别
message.setDelayTimeLevel(3); // 延迟10秒

// 设置 acks=all，等待所有副本确认
props.put("acks", "all");
// 设置重试次数
props.put("retries", Integer.MAX_VALUE);
// 设置重试间隔
props.put("retry.backoff.ms", 100);

// 禁用自动提交
props.put("enable.auto.commit", "false");
// 手动提交偏移量
consumer.commitSync();

// 声明死信交换器
channel.exchangeDeclare("dlx", "direct");

// 声明死信队列
channel.queueDeclare("dlq", false, false, false, null);

// 绑定死信队列
channel.queueBind("dlq", "dlx", "dlq");

// 主队列配置死信
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx");
args.put("x-dead-letter-routing-key", "dlq");
channel.queueDeclare("main_queue", false, false, false, args);

boolean durable = true;
channel.queueDeclare("durable_queue", durable, false, false, null);

AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;
channel.basicPublish("", "durable_queue", props, message.getBytes());

channel.exchangeDeclare("durable_exchange", "direct", true);

// 连接池配置
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");

// 连接池参数
factory.setRequestedHeartBeat(60);
factory.setConnectionTimeout(60000);
factory.setRequestedChannelMax(0);

// 通道复用
Channel channel = connection.createChannel();

// 批量确认
channel.basicQos(100); // 预取数量
channel.basicAck(deliveryTag, true); // 批量确认

// 队列参数优化
Map<String, Object> args = new HashMap<>();
args.put("x-max-length", 10000); // 队列最大长度
args.put("x-message-ttl", 60000); // 消息TTL
args.put("x-expires", 300000); // 队列TTL

// 消息加密
public class MessageEncryption {
    private static final String ALGORITHM = "AES";
    private static final String TRANSFORMATION = "AES/CBC/PKCS5Padding";

    public byte[] encrypt(String message, String key) throws Exception {
        Cipher cipher = Cipher.getInstance(TRANSFORMATION);
        SecretKeySpec secretKey = new SecretKeySpec(key.getBytes(), ALGORITHM);
        cipher.init(Cipher.ENCRYPT_MODE, secretKey);
        return cipher.doFinal(message.getBytes());
    }
}

// 订单流程消息设计
public class OrderMessage {
    private String orderId;
    private String userId;
    private OrderStatus status;
    private Long timestamp;
    private Map<String, Object> data;
}

// 消息路由策略
@Component
public class OrderMessageRouter {
    public void routeOrderMessage(OrderMessage message) {
        switch (message.getStatus()) {
            case CREATED:
                // 发送到库存扣减队列
                inventoryService.deductStock(message);
                break;
            case PAID:
                // 发送到发货队列
                shippingService.prepareShipment(message);
                break;
            case SHIPPED:
                // 发送到物流跟踪队列
                logisticsService.trackShipment(message);
                break;
        }
    }
}

// 转账事务消息
@Service
@Transactional
public class TransferService {

    public void transfer(TransferRequest request) {
        // 1. 扣减转出账户
        accountService.deduct(request.getFromAccount(), request.getAmount());

        // 2. 发送事务消息
        TransactionMessage message = new TransactionMessage();
        message.setTransferId(request.getTransferId());
        message.setToAccount(request.getToAccount());
        message.setAmount(request.getAmount());

        // 3. 发送半消息
        rocketMQTemplate.sendMessageInTransaction("transfer-topic", message, request);
    }

    @RocketMQTransactionListener
    public class TransferTransactionListener implements RocketMQLocalTransactionListener {

        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                // 执行本地事务：增加转入账户余额
                TransferRequest request = (TransferRequest) arg;
                accountService.add(request.getToAccount(), request.getAmount());
                return RocketMQLocalTransactionState.COMMIT;
            } catch (Exception e) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 回查本地事务状态
            return checkTransferStatus(msg);
        }
    }
}

// 日志消息模型
public class LogMessage {
    private String appId;
    private String level;
    private String message;
    private String traceId;
    private Long timestamp;
    private Map<String, String> tags;
}

// 日志收集器
@Component
public class LogCollector {

    @Autowired
    private KafkaTemplate<String, LogMessage> kafkaTemplate;

    public void collectLog(LogMessage logMessage) {
        // 根据日志级别路由到不同Topic
        String topic = getTopicByLevel(logMessage.getLevel());

        // 异步发送到Kafka
        kafkaTemplate.send(topic, logMessage.getAppId(), logMessage);
    }

    private String getTopicByLevel(String level) {
        switch (level.toUpperCase()) {
            case "ERROR":
                return "error-logs";
            case "WARN":
                return "warn-logs";
            case "INFO":
                return "info-logs";
            default:
                return "debug-logs";
        }
    }
}

// 日志处理器
@Component
public class LogProcessor {

    @KafkaListener(topics = "error-logs")
    public void processErrorLogs(LogMessage logMessage) {
        // 错误日志处理：告警、统计、存储
        alertService.sendAlert(logMessage);
        statisticsService.recordError(logMessage);
        storageService.saveLog(logMessage);
    }

    @KafkaListener(topics = "info-logs")
    public void processInfoLogs(LogMessage logMessage) {
        // 信息日志处理：统计、存储
        statisticsService.recordInfo(logMessage);
        storageService.saveLog(logMessage);
    }
}

// 用户行为消息
public class UserBehaviorMessage {
    private String userId;
    private String itemId;
    private String action; // view, click, purchase
    private Long timestamp;
    private Map<String, Object> context;
}

// 行为收集器
@Component
public class BehaviorCollector {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void collectBehavior(UserBehaviorMessage behavior) {
        // 发送到行为分析队列
        rocketMQTemplate.convertAndSend("user-behavior", behavior);

        // 发送到实时推荐队列
        rocketMQTemplate.convertAndSend("realtime-recommendation", behavior);
    }
}

// 实时推荐引擎
@Component
public class RealtimeRecommendationEngine {

    @RocketMQMessageListener(topic = "realtime-recommendation")
    public class RealtimeRecommendationListener implements RocketMQListener<UserBehaviorMessage> {

        @Override
        public void onMessage(UserBehaviorMessage behavior) {
            // 1. 更新用户画像
            userProfileService.updateProfile(behavior);

            // 2. 计算实时推荐
            List<String> recommendations = calculateRecommendations(behavior);

            // 3. 发送推荐结果
            RecommendationMessage recMessage = new RecommendationMessage();
            recMessage.setUserId(behavior.getUserId());
            recMessage.setRecommendations(recommendations);
            recMessage.setTimestamp(System.currentTimeMillis());

            rocketMQTemplate.convertAndSend("recommendation-result", recMessage);
        }

        private List<String> calculateRecommendations(UserBehaviorMessage behavior) {
            // 基于协同过滤、内容推荐等算法计算推荐结果
            return recommendationAlgorithm.calculate(behavior);
        }
    }
}

// 领域事件
public class DomainEvent {
    private String eventId;
    private String eventType;
    private String aggregateId;
    private Object payload;
    private Long timestamp;
    private String source;
}

// 事件发布器
@Component
public class EventPublisher {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void publishEvent(DomainEvent event) {
        // 发送到事件总线
        rabbitTemplate.convertAndSend("event-bus", event.getEventType(), event);

        // 记录事件发布日志
        eventLogService.logEvent(event);
    }
}

// 事件处理器
@Component
public class EventHandler {

    @RabbitListener(queues = "user-service-events")
    public void handleUserEvent(DomainEvent event) {
        switch (event.getEventType()) {
            case "USER_REGISTERED":
                handleUserRegistered(event);
                break;
            case "USER_UPDATED":
                handleUserUpdated(event);
                break;
            case "USER_DELETED":
                handleUserDeleted(event);
                break;
        }
    }

    private void handleUserRegistered(DomainEvent event) {
        // 发送欢迎邮件
        emailService.sendWelcomeEmail(event.getPayload());

        // 初始化用户偏好
        preferenceService.initUserPreference(event.getPayload());

        // 记录用户注册统计
        statisticsService.recordUserRegistration(event.getPayload());
    }
}

// 数据消息模型
public class DataMessage {
    private String dataId;
    private String dataType;
    private String source;
    private Object data;
    private Long timestamp;
    private Map<String, String> metadata;
}

// 数据采集器
@Component
public class DataCollector {

    @Autowired
    private KafkaTemplate<String, DataMessage> kafkaTemplate;

    public void collectData(DataMessage dataMessage) {
        // 数据预处理
        DataMessage processedData = preprocessData(dataMessage);

        // 发送到数据管道
        kafkaTemplate.send("data-pipeline", processedData.getDataType(), processedData);
    }

    private DataMessage preprocessData(DataMessage data) {
        // 数据清洗、验证、格式化
        return dataProcessor.process(data);
    }
}

// 数据处理器
@Component
public class DataProcessor {

    @KafkaListener(topics = "data-pipeline")
    public void processData(DataMessage dataMessage) {
        // 根据数据类型进行不同处理
        switch (dataMessage.getDataType()) {
            case "USER_BEHAVIOR":
                processUserBehavior(dataMessage);
                break;
            case "BUSINESS_METRICS":
                processBusinessMetrics(dataMessage);
                break;
            case "SYSTEM_LOGS":
                processSystemLogs(dataMessage);
                break;
        }
    }

    private void processUserBehavior(DataMessage data) {
        // 用户行为数据处理
        userBehaviorService.process(data);

        // 发送到分析队列
        kafkaTemplate.send("analytics-queue", data);
    }
}