docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: / # 虛擬主機(jī)（默認(rèn)/）
    publisher-confirm-type: correlated # 開啟生產(chǎn)者確認(rèn)機(jī)制
    publisher-returns: true # 開啟消息回退機(jī)制
    listener:
      simple:
        acknowledge-mode: manual # 消費(fèi)者手動(dòng)確認(rèn)消息
        concurrency: 2 # 消費(fèi)者核心線程數(shù)
        max-concurrency: 5 # 消費(fèi)者最大線程數(shù)

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    // 交換機(jī)名稱（Topic交換機(jī)，支持模糊路由，最常用）
    public static final String TOPIC_EXCHANGE = "order_exchange";
    // 隊(duì)列名稱（訂單通知隊(duì)列）
    public static final String ORDER_QUEUE = "order_queue";
    // 路由鍵（匹配規(guī)則：order.* 匹配 order.create、order.cancel 等）
    public static final String ROUTING_KEY = "order.#";
    // 1. 聲明Topic交換機(jī)
    @Bean
    public TopicExchange topicExchange() {
        // durable=true：交換機(jī)持久化，重啟RabbitMQ不丟失
        return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE).durable(true).build();
    }
    // 2. 聲明隊(duì)列
    @Bean
    public Queue orderQueue() {
        // durable=true：隊(duì)列持久化；exclusive=false：不排他；autoDelete=false：不自動(dòng)刪除
        return QueueBuilder.durable(ORDER_QUEUE).build();
    }
    // 3. 綁定交換機(jī)與隊(duì)列（指定路由鍵）
    @Bean
    public Binding bindingExchangeQueue(TopicExchange topicExchange, Queue orderQueue) {
        return BindingBuilder.bind(orderQueue).to(topicExchange).with(ROUTING_KEY);
    }
}

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.UUID;
@Component
public class OrderProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;
    // 發(fā)送訂單創(chuàng)建消息
    public void sendOrderCreateMsg(Long orderId, String userId) {
        // 1. 構(gòu)建消息內(nèi)容
        String msg = String.format("用戶%s創(chuàng)建訂單：%s", userId, orderId);
        // 2. 消息ID（用于確認(rèn)機(jī)制，追蹤消息）
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 3. 發(fā)送消息（交換機(jī)、路由鍵、消息內(nèi)容、消息ID）
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.TOPIC_EXCHANGE,
                "order.create", // 具體路由鍵（匹配 order.#）
                msg,
                correlationData
        );
    }
    // 4. 生產(chǎn)者確認(rèn)回調(diào)（確認(rèn)消息是否到達(dá)交換機(jī)）
    @Resource
    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        // 消息到達(dá)交換機(jī)回調(diào)
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("消息到達(dá)交換機(jī)，消息ID：" + correlationData.getId());
            } else {
                System.out.println("消息未到達(dá)交換機(jī)，原因：" + cause);
                // 消息投遞失敗，可重試或記錄日志
            }
        });
        // 消息無法路由到隊(duì)列回調(diào)（回退機(jī)制）
        rabbitTemplate.setReturnsCallback(returned -> {
            System.out.println("消息無法路由，路由鍵：" + returned.getRoutingKey() + "，原因：" + returned.getReplyText());
        });
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
public class OrderConsumer {
    // 監(jiān)聽訂單隊(duì)列
    @RabbitListener(queues = RabbitMQConfig.ORDER_QUEUE)
    public void consumeOrderMsg(String msg, Channel channel, Message message) throws IOException {
        try {
            // 1. 處理業(yè)務(wù)邏輯（如更新訂單狀態(tài)、發(fā)送短信通知）
            System.out.println("接收訂單消息：" + msg);
            // 2. 手動(dòng)確認(rèn)消息（multiple=false：只確認(rèn)當(dāng)前消息；true：確認(rèn)所有未確認(rèn)消息）
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 3. 消息消費(fèi)失敗，拒絕消息并重回隊(duì)列（或死信隊(duì)列）
            // requeue=true：重回隊(duì)列；false：不重回隊(duì)列（需配置死信隊(duì)列處理）
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            System.out.println("消息消費(fèi)失敗，已重回隊(duì)列：" + msg);
        }
    }
}

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
@RestController
public class OrderController {
    @Resource
    private OrderProducer orderProducer;
    @GetMapping("/order/create/{orderId}/{userId}")
    public String createOrder(@PathVariable Long orderId, @PathVariable String userId) {
        orderProducer.sendOrderCreateMsg(orderId, userId);
        return "訂單創(chuàng)建消息已發(fā)送";
    }
}

# 啟動(dòng)ZooKeeper（Kafka依賴ZooKeeper管理元數(shù)據(jù)）
docker run -d --name zookeeper -p 2181:2181 confluentinc/cp-zookeeper:latest
# 啟動(dòng)Kafka
docker run -d --name kafka -p 9092:9092 \
-e KAFKA_ZOOKEEPER_CONNECT=localhost:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
confluentinc/cp-kafka:latest

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

spring:
  kafka:
    bootstrap-servers: localhost:9092 # Kafka服務(wù)地址
    # 生產(chǎn)者配置
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: 1 # 消息確認(rèn)機(jī)制（1：領(lǐng)導(dǎo)者分區(qū)確認(rèn)；all：所有副本確認(rèn)）
      retries: 3 # 消息發(fā)送失敗重試次數(shù)
    # 消費(fèi)者配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: order-group # 消費(fèi)者組ID（同一組內(nèi)消費(fèi)者負(fù)載均衡消費(fèi)）
      auto-offset-reset: earliest # 無偏移量時(shí)，從最早消息開始消費(fèi)
      enable-auto-commit: false # 關(guān)閉自動(dòng)提交偏移量，手動(dòng)提交
    # 監(jiān)聽配置
    listener:
      ack-mode: manual_immediate # 手動(dòng)提交偏移量

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import javax.annotation.Resource;
@Component
public class KafkaOrderProducer {
    // 主題名稱（Kafka無需提前聲明，發(fā)送消息時(shí)自動(dòng)創(chuàng)建）
    public static final String ORDER_TOPIC = "order_topic";
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    // 異步發(fā)送訂單消息
    public void sendOrderMsg(Long orderId, String userId) {
        String msg = String.format("用戶%s創(chuàng)建訂單：%s", userId, orderId);
        // 發(fā)送消息（主題、消息鍵、消息內(nèi)容）
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(ORDER_TOPIC, orderId.toString(), msg);
        // 回調(diào)函數(shù)：處理發(fā)送結(jié)果
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("消息發(fā)送成功，分區(qū)：" + result.getRecordMetadata().partition());
            }
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("消息發(fā)送失敗，原因：" + ex.getMessage());
                // 失敗重試邏輯
            }
        });
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class KafkaOrderConsumer {
    // 監(jiān)聽訂單主題，手動(dòng)提交偏移量
    @KafkaListener(topics = KafkaOrderProducer.ORDER_TOPIC, groupId = "order-group")
    public void consumeOrderMsg(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            // 1. 獲取消息內(nèi)容
            String key = record.key();
            String msg = record.value();
            System.out.println("接收Kafka消息，訂單ID：" + key + "，內(nèi)容：" + msg);
            // 2. 處理業(yè)務(wù)邏輯
            // 3. 手動(dòng)提交偏移量（確認(rèn)消息消費(fèi)成功）
            acknowledgment.acknowledge();
        } catch (Exception e) {
            System.out.println("消息消費(fèi)失?。? + e.getMessage());
            // 消費(fèi)失敗可記錄日志，人工介入處理
        }
    }
}

@Resource
private KafkaOrderProducer kafkaOrderProducer;
@GetMapping("/kafka/order/create/{orderId}/{userId}")
public String createOrderByKafka(@PathVariable Long orderId, @PathVariable String userId) {
    kafkaOrderProducer.sendOrderMsg(orderId, userId);
    return "Kafka訂單消息已發(fā)送";
}

// 基于Redis實(shí)現(xiàn)冪等（消費(fèi)前檢查消息是否已處理）
public void consumeWithIdempotent(String msgId, String msg) {
    String key = "msg:processed:" + msgId;
    Boolean isProcessed = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 24, TimeUnit.HOURS);
    if (Boolean.TRUE.equals(isProcessed)) {
        // 未處理，執(zhí)行業(yè)務(wù)邏輯
        System.out.println("處理消息：" + msg);
    } else {
        // 已處理，直接跳過
        System.out.println("消息已重復(fù)消費(fèi)，跳過：" + msg);
    }
}