<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-consumer-group
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

@Service
public class OrderMessageProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public static final String TOPIC = "order-topic";
    public void sendOrderMessage(OrderMessage orderMessage) {
        String key = orderMessage.getOrderId();
        String value = JSON.toJSONString(orderMessage);
        kafkaTemplate.send(TOPIC, key, value, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    log.error("消息發(fā)送失敗，訂單號(hào)：{}，錯(cuò)誤信息：{}", 
                              orderMessage.getOrderId(), exception.getMessage());
                } else {
                    log.info("消息發(fā)送成功，訂單號(hào)：{}，分區(qū)：{}，偏移量：{}", 
                             orderMessage.getOrderId(), 
                             metadata.partition(), 
                             metadata.offset());
                }
            }
        });
    }
}

@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(6);
        factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

@Component
public class OrderKafkaListener {
    @Autowired
    private OrderService orderService;
    @KafkaListener(
        topics = OrderMessageProducer.TOPIC,
        groupId = "${spring.kafka.consumer.group-id}"
    )
    public void consumeOrderMessage(ConsumerRecord<String, String> record, 
                                    Acknowledgment acknowledgment) {
        try {
            String value = record.value();
            OrderMessage orderMessage = JSON.parseObject(value, OrderMessage.class);
            log.info("收到訂單消息，分區(qū)：{}，偏移量：{}，訂單號(hào)：{}", 
                     record.partition(), record.offset(), orderMessage.getOrderId());
            orderService.processOrder(orderMessage);
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("訂單消息處理失敗，錯(cuò)誤信息：{}", e.getMessage());
            throw e;
        }
    }
}

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

rocketmq:
  name-server: localhost:9876
  producer:
    group: order-producer-group

@Service
public class OrderMessageProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public static final String TOPIC = "order-topic";
    public void sendOrderMessage(OrderMessage orderMessage) {
        String keys = orderMessage.getOrderId();
        String tags = orderMessage.getOrderType();
        rocketMQTemplate.asyncSend(
            TOPIC + ":order",
            MessageBuilder.withPayload(orderMessage)
                          .setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
                          .build(),
            new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("順序消息發(fā)送成功，訂單號(hào)：{}，隊(duì)列ID：{}", 
                             orderMessage.getOrderId(), 
                             sendResult.getMessageQueue().getQueueId());
                }
                @Override
                public void onException(Throwable e) {
                    log.error("順序消息發(fā)送失敗，訂單號(hào)：{}", orderMessage.getOrderId());
                }
            },
            3000,
            keys,
            tags
        );
    }
}

@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer-group",
    tag = "order",
    messageModel = MessageModel.CLUSTERING,
    consumeMode = ConsumeMode.ORDERLY
)
public class OrderMessageConsumer implements RocketMQListener<OrderMessage> {
    @Autowired
    private OrderService orderService;
    @Override
    public void onMessage(OrderMessage orderMessage) {
        try {
            log.info("收到順序消息，訂單號(hào)：{}", orderMessage.getOrderId());
            orderService.processOrder(orderMessage);
        } catch (Exception e) {
            log.error("訂單消息處理失敗，訂單號(hào)：{}，錯(cuò)誤信息：{}", 
                      orderMessage.getOrderId(), e.getMessage());
            throw new RuntimeException("處理失敗", e);
        }
    }
}

@Configuration
public class RabbitMQConfig {
    public static final String ORDER_QUEUE = "order.queue";
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_ROUTING_KEY = "order.create";
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable(ORDER_QUEUE)
                .withArgument("x-single-active-consumer", true)
                .build();
    }
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(ORDER_EXCHANGE);
    }
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
                .to(orderExchange())
                .with(ORDER_ROUTING_KEY);
    }
}

@Component
public class OrderMessageListener {
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void handleOrderMessage(OrderMessage orderMessage, Channel channel,
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("收到訂單消息，訂單號(hào)：{}", orderMessage.getOrderId());
            orderService.processOrder(orderMessage);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("訂單消息處理失敗，訂單號(hào)：{}，錯(cuò)誤信息：{}", 
                      orderMessage.getOrderId(), e.getMessage());
            channel.basicNack(deliveryTag, false, true);
        }
    }
}

public class PartitionKeyStrategy {     public static String forOrderOperations(String orderId) {         return "order:" + orderId;     }     public static String forUserOperations(String userId) {         return "user:" + userId;     }     public static String forBusinessFlow(String businessType, String businessId) {         return businessType + ":" + businessId;     } }

@Service
public class IdempotentOrderService {
    @Autowired
    private OrderMapper orderMapper;
    public void processOrder(IdempotentOrderMessage message) {
        try {
            Order order = Order.builder()
                    .orderId(message.getOrderId())
                    .amount(message.getAmount())
                    .status(message.getStatus())
                    .build();
            orderMapper.insertSelective(order);
        } catch (DuplicateKeyException e) {
            log.info("訂單已存在，跳過處理，訂單號(hào)：{}", message.getOrderId());
        }
    }
}

@Service
public class StateMachineOrderService {
    public void processOrderStateChange(StateChangeMessage message) {
        Order order = orderMapper.selectByOrderId(message.getOrderId());
        if (!canTransition(order.getStatus(), message.getNewStatus())) {
            throw new IllegalStateException("狀態(tài)轉(zhuǎn)換非法");
        }
        Order updated = Order.builder()
                .id(order.getId())
                .status(message.getNewStatus())
                .version(order.getVersion())
                .build();
        int rows = orderMapper.updateByVersion(updated);
        if (rows == 0) {
            throw new OptimisticLockException("版本沖突");
        }
    }
    private boolean canTransition(OrderStatus from, OrderStatus to) {
        return TRANSITIONS.get(from).contains(to);
    }
}

@Service
public class OrderSequentialConsumer {
    private static final int MAX_RETRY_COUNT = 3;
    private static final long RETRY_INTERVAL = 5000;
    @Autowired
    private OrderService orderService;
    @Autowired
    private DeadLetterService deadLetterService;
    public void consumeOrderMessage(OrderMessage message, int retryCount) {
        try {
            orderService.processOrder(message);
        } catch (TemporaryException e) {
            if (retryCount < MAX_RETRY_COUNT) {
                log.warn("臨時(shí)性錯(cuò)誤，將在{}ms后重試，訂單號(hào)：{}", 
                         RETRY_INTERVAL, message.getOrderId());
                throw new RetryableException(RETRY_INTERVAL);
            }
            moveToDeadLetter(message, "臨時(shí)錯(cuò)誤超過最大重試次數(shù)");
        } catch (BusinessException e) {
            log.error("業(yè)務(wù)錯(cuò)誤，無法重試，訂單號(hào)：{}，錯(cuò)誤信息：{}", 
                      message.getOrderId(), e.getMessage());
            moveToDeadLetter(message, "業(yè)務(wù)錯(cuò)誤：" + e.getMessage());
        } catch (Exception e) {
            log.error("未知錯(cuò)誤，訂單號(hào)：{}，錯(cuò)誤信息：{}", 
                      message.getOrderId(), e.getMessage());
            moveToDeadLetter(message, "系統(tǒng)錯(cuò)誤：" + e.getMessage());
        }
    }
    private void moveToDeadLetter(OrderMessage message, String reason) {
        deadLetterService.saveDeadLetter(message, reason);
        deadLetterService.sendAlert(message, reason);
    }
}

public class UnifiedPartitionKeyStrategy {
    public static String forUserOperations(String userId, String operationType) {
        return "user:" + userId + ":operation:" + operationType;
    }
    public static String forPaymentCallback(String orderId, String callbackType) {
        return "order:" + orderId + ":callback:" + callbackType;
    }
}

@Data
public class SequencedMessage {
    private String messageId;
    private String sequenceId;
    private String payload;
    private long timestamp;
}
@Service
public class SequencedMessageProcessor {
    private Map<String, TreeMap<Long, SequencedMessage>> userMessages = new ConcurrentHashMap<>();
    public void processMessage(SequencedMessage message) {
        String userId = extractUserId(message);
        userMessages.computeIfAbsent(userId, k -> new TreeMap<>())
                   .put(Long.parseLong(message.getSequenceId()), message);
        TreeMap<Long, SequencedMessage> messages = userMessages.get(userId);
        processInOrder(messages, userId);
    }
    private void processInOrder(TreeMap<Long, SequencedMessage> messages, String userId) {
        while (!messages.isEmpty()) {
            Map.Entry<Long, SequencedMessage> entry = messages.firstEntry();
            SequencedMessage message = entry.getValue();
            if (!isNextSequence(userId, Long.parseLong(message.getSequenceId()))) {
                break;
            }
            doProcess(message);
            messages.remove(entry.getKey());
            updateLastProcessedSequence(userId, entry.getKey());
        }
    }
}

spring:
  kafka:
    consumer:
      concurrency: 8