别再瞎装 RabbitMQ 了！从 0 到 1 部署到 Spring Boot 全链路实战指南(生产级坑全填平)

# 1. 安装匹配版本的 erlang 依赖
yum install -y erlang-25.3.2.7
# 2. 安装 rabbitmq 服务包
rpm -ivh rabbitmq-server-3.12.12-1.el7.noarch.rpm
# 3. 启动服务 + 设置开机自启
systemctl start rabbitmq-server
systemctl enable rabbitmq-server

# ① 开启 web 管理控制台（运维必备，不开等于瞎子）
rabbitmq-plugins enable rabbitmq_management
# ② 删除默认 guest 用户（默认只允许 localhost 访问，远程连接直接报错）
rabbitmqctl delete_user guest
# ③ 创建管理员用户
rabbitmqctl add_user order_admin yourstrongpassword
rabbitmqctl set_user_tags order_admin administrator
rabbitmqctl set_permissions -p / order_admin ".*" ".*" ".*"
# ④ 创建业务 virtual host（资源隔离，不同项目的交换机和队列互不干扰）
rabbitmqctl add_vhost /order_vhost
rabbitmqctl set_permissions -p /order_vhost order_admin ".*" ".*" ".*"

<dependency>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-amqp</artifactid>
</dependency>

spring:
  rabbitmq:
    host: 192.168.1.100
    port: 5672
    username: order_admin
    password: yourstrongpassword
    virtual-host: /order_vhost
    # 开启发送端确认（消息到达 exchange 后有回调）
    publisher-confirm-type: correlated
    # 开启发送端回退（路由到 queue 失败可感知）
    publisher-returns: true
    # 消费端配置
    listener:
      simple:
        acknowledge-mode: manual   # 手动 ack，保证消费可靠性
        prefetch: 10               # 每次预取 10 条，配合限流防打爆

@configuration
public class rabbitorderconfig {
    // 订单交换机
    @bean
    public directexchange orderexchange() {
        return new directexchange("order.exchange", true, false);
    }
    // 订单队列 —— 注意这里绑定了死信交换机
    @bean
    public queue orderqueue() {
        map<string, object> args = new hashmap<>();
        args.put("x-dead-letter-exchange", "order.dlx.exchange");     // 死信交换机
        args.put("x-dead-letter-routing-key", "order.dlx.routingkey"); // 死信路由 key
        args.put("x-message-ttl", 600000);                             // 消息 ttl 10 分钟
        return new queue("order.queue", true, false, false, args);
    }
    // 绑定关系
    @bean
    public binding orderbinding() {
        return bindingbuilder.bind(orderqueue()).to(orderexchange()).with("order.create");
    }
    // ===== 死信交换机 & 死信队列 =====
    @bean
    public directexchange orderdlxexchange() {
        return new directexchange("order.dlx.exchange", true, false);
    }
    @bean
    public queue orderdlxqueue() {
        return new queue("order.dlx.queue", true);
    }
    @bean
    public binding orderdlxbinding() {
        return bindingbuilder.bind(orderdlxqueue()).to(orderdlxexchange()).with("order.dlx.routingkey");
    }
}

@service
public class ordermessagesender {
    @autowired
    private rabbittemplate rabbittemplate;
    @autowired
    private msglogrepository msglogrepository;
    /**
     * 发送订单消息
     * 核心思路：先落库 → 再发送 → 回调更新状态 → 失败定时补偿
     */
    public void sendordermessage(order order) {
        // 1. 消息落库，status=0（待确认）
        msglog msglog = new msglog();
        msglog.setmsgid(uuid.randomuuid().tostring());
        msglog.setcontent(json.tojsonstring(order));
        msglog.setstatus(0);
        msglogrepository.save(msglog);
        // 2. 携带 msgid 发送，用于回调匹配
        correlationdata correlationdata = new correlationdata(msglog.getmsgid());
        rabbittemplate.convertandsend("order.exchange", "order.create", order, correlationdata);
    }
    // 统一设置 confirm 和 return 回调
    @postconstruct
    public void initrabbittemplate() {
        // confirmcallback：消息是否到达 exchange
        rabbittemplate.setconfirmcallback((correlationdata, ack, cause) -> {
            string msgid = correlationdata.getid();
            if (ack) {
                msglogrepository.updatestatus(msgid, 1); // 成功
            } else {
                msglogrepository.updatestatus(msgid, 2); // 失败，等待补偿
                log.error("消息发送到 exchange 失败, msgid: {}", msgid);
            }
        });
        // returncallback：消息路由不到 queue 时触发
        rabbittemplate.setreturncallback((message, replycode, replytext, exchange, routingkey) -> {
            string msgid = message.getmessageproperties().getcorrelationid();
            msglogrepository.updatestatus(msgid, 2);
            log.error("消息路由失败, msgid: {}, replytext: {}", msgid, replytext);
        });
    }
}

@component
public class orderconsumer {
    @autowired
    private msglogrepository msglogrepository;
    @autowired
    private businessservice businessservice;
    @rabbitlistener(queues = "order.queue")
    public void handleorder(message message, channel channel) throws ioexception {
        string msgid = message.getmessageproperties().getcorrelationid();
        order order = json.parseobject(message.getbody(), order.class);
        try {
            // ① 幂等判断：这条消息是否已经处理过了？
            if (msglogrepository.ismsgprocessed(msgid)) {
                channel.basicack(message.getmessageproperties().getdeliverytag(), false);
                return; // 已处理，直接 ack 丢弃
            }
            // ② 核心业务逻辑：发短信、加积分、写日志……
            businessservice.process(order);
            // ③ 标记已处理 + 手动 ack
            msglogrepository.markprocessed(msgid);
            channel.basicack(message.getmessageproperties().getdeliverytag(), false);
        } catch (exception e) {
            log.error("消费异常, msgid: {}", msgid, e);
            // 拒绝且不放回队列 → 消息进入死信队列，由独立任务处理
            channel.basicnack(message.getmessageproperties().getdeliverytag(), false, false);
        }
    }
}

<dependency>
    <groupid>com.rabbitmq</groupid>
    <artifactid>amqp-client</artifactid>
    <version>5.20.0</version>
</dependency>

public class orderproducer {
    private static final string exchange = "order_direct_exchange";
    private static final string queue = "order_create_queue";
    private static final string routing_key = "order.create";
    public static void main(string[] args) throws exception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("192.168.1.100");
        factory.setport(5672);
        factory.setusername("order_admin");
        factory.setpassword("yourstrongpassword");
        factory.setvirtualhost("/");
        try (connection conn = factory.newconnection();
             channel channel = conn.createchannel()) {
            // 声明交换机、队列（durable=true 持久化，broker 重启不丢失）
            channel.exchangedeclare(exchange, builtinexchangetype.direct, true);
            channel.queuedeclare(queue, true, false, false, null);
            channel.queuebind(queue, exchange, routing_key);
            // 开启生产者发布确认
            channel.confirmselect();
            string msg = "订单id:10086, 用户id:6666";
            // persistent_text_plain：消息持久化落盘
            channel.basicpublish(exchange, routing_key,
                    messageproperties.persistent_text_plain,
                    msg.getbytes(standardcharsets.utf_8));
            // 同步确认（生产环境推荐异步 confirmlistener 批量确认，性能更高）
            if (channel.waitforconfirms()) {
                system.out.println("消息投递成功");
            } else {
                system.out.println("消息投递失败，触发重试逻辑");
            }
        }
    }
}

public class orderconsumer {
    private static final string queue = "order_create_queue";
    public static void main(string[] args) throws exception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("192.168.1.100");
        factory.setusername("order_admin");
        factory.setpassword("yourstrongpassword");
        connection conn = factory.newconnection();
        channel channel = conn.createchannel();
        // 每次只拉 1 条消息，处理完再拉下一条，防止服务被打垮
        channel.basicqos(1);
        // autoack=false：关闭自动确认，手动 ack 保证消费可靠性
        channel.basicconsume(queue, false, (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), standardcharsets.utf_8);
            try {
                system.out.println("处理订单：" + message);
                // 业务处理成功，手动确认
                channel.basicack(delivery.getenvelope().getdeliverytag(), false);
            } catch (exception e) {
                // 消费失败：nack 后消息重回队列
                // 生产建议：requeue=false 转死信队列，避免无限重试
                channel.basicnack(delivery.getenvelope().getdeliverytag(), false, true);
            }
        }, system.out::println);
    }
}

# 全局策略：所有队列开启镜像，同步到集群所有节点
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

# 开启 prometheus 监控插件
rabbitmq-plugins enable rabbitmq_prometheus