RabbitMQ 核心概念与六种常用模式解析

安装 Erlang

# 更新软件包
root@VM-0-7-ubuntu:~# apt-get update
# 安装 erlang
root@VM-0-7-ubuntu:~# apt-get install erlang
# 查看 erlang 版本，输入 "erl" 会启动 Erlang 运行时系统
root@VM-0-7-ubuntu:~# erl
# 输入 "halt()." 立即停止 Erlang 运行时系统
1> halt().

2.2 访问

• 5672 端口是 RabbitMQ 的默认 AMQP 协议端口。生产者和消费者通过此端口与 RabbitMQ 交互。
• 15672 端口是 RabbitMQ 管理插件的默认 HTTP 端口。通过此端口可以访问 RabbitMQ 的 Web 管理界面，提供图形化的管理功能。

添加管理员用户

# 添加用户
rabbitmqctl add_user admin admin
# 用户授权
rabbitmqctl set_user_tags admin administrator

2.3 工作流程

第一阶段：连接建立 建立 TCP 连接：生产者应用程序首先与 Broker 建立 TCP 连接，默认端口为 5672。连接需要指定虚拟主机、用户名和密码。 创建信道：在已建立的 TCP 连接上创建一个或多个信道。信道是虚拟连接，共享同一个 TCP 连接。每个信道有独立的通信流，相互隔离。

第二阶段：资源声明 声明交换机：指定名称、类型和属性。 声明队列：指定名称和属性。 创建绑定：将队列绑定到交换机，并指定绑定键 (Binding Key)。

第三阶段：消息发送 发送消息到交换机：将消息发布到指定的交换机。消息包含：交换机名称、路由键、消息体、消息属性。 交换机路由消息：交换机接收消息后，根据其类型和绑定规则进行路由。

第四阶段：路由处理 成功路由：队列接收并存储消息，等待消费者拉取或推送消息给消费者。

3. RabbitMQ 六种工作模式

3.1 Simple

核心特征：

• 最简单的消息队列模式
• 一个生产者，一个消费者，一个队列

工作流程：

代码实现：

/**
 * 生产者
 */
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机，使用内置交换机

        // 4. 声明队列
        /* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
           queue：队列名称 durable：可持久化 exclusive：是否独占 autoDelete：是否自动删除 arguments：参数，暂不介绍 */
        channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);

        // 5. 发送消息
        for (int i = 0; i < 10; i++) {
            channel.basicPublish("", Constants.WORK_QUEUE, null, ("Hello RabbitMQ!" + i).getBytes());
        }
        System.out.println("消息发送成功");

        // 6. 资源释放
        channel.close();
        connection.close();
    }
}

/**
 * 消费者
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机，使用内置交换机

        // 4. 声明队列
        /* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
           queue：队列名称 durable：可持久化 exclusive：是否独占 autoDelete：是否自动删除 arguments：参数，暂不介绍 */
        channel.queueDeclare("first", true, false, false, null);

        // 5. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.WORK_QUEUE, true, defaultConsumer);
        Thread.sleep(2000);

        // 6. 资源释放
        channel.close();
        connection.close();
    }
}

3.2 Work Queues

核心特征：

• 多个消费者竞争消费同一个队列
• 轮询分发 (默认)：每个消费者依次接收消息

工作流程：

代码实现：

/**
 * 生产者
 */
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机，使用内置交换机

        // 4. 声明队列
        channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);

        // 5. 发送消息
        for (int i = 0; i < 10; i++) {
            channel.basicPublish("", Constants.WORK_QUEUE, null, ("Hello RabbitMQ!:" + i).getBytes());
        }
        System.out.println("消息发送成功");

        // 6. 资源释放
        channel.close();
        connection.close();
    }
}

/**
 * 消费者 A
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerA {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机，使用内置交换机

        // 4. 声明队列
        channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);

        // 5. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("ConsumerA 接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.WORK_QUEUE, true, defaultConsumer);
        Thread.sleep(2000);

        // 6. 资源释放
        // channel.close();
        // connection.close();
    }
}

/**
 * 消费者 B
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerB {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机，使用内置交换机

        // 4. 声明队列
        channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);

        // 5. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("ConsumerB 接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.WORK_QUEUE, true, defaultConsumer);
        Thread.sleep(2000);

        // 6. 资源释放
        // channel.close();
        // connection.close();
    }
}

3.3 Publish/Subscribe

核心特征：

• 一条消息被多个消费者接收
• 消息广播到所有绑定队列

工作流程：

代码实现：

/**
 * 生产者
 */
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        // BuiltinExchangeType.FANOUT: 广播类型，将消息交给所有绑定到交换机的队列
        channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);

        // 4. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);
        channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);

        // 5. 交换机和队列绑定
        channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, "");
        channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, "");

        // 6. 发送消息
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(Constants.FANOUT_EXCHANGE, "", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        System.out.println("消息发送成功");

        // 7. 资源释放
        channel.close();
        connection.close();
    }
}

/**
 * 消费者 A
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerA {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.FANOUT_QUEUE1, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

/**
 * 消费者 B
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerB {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.FANOUT_QUEUE2, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

3.4 Routing

核心特征：

• 根据路由键选择性接收消息
• 精确匹配路由键

工作流程：

代码实现：

/**
 * 生产者
 */
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        // BuiltinExchangeType.DIRECT: 定向类型，把消息交给符合指定 binding key 的队列
        channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);

        // 4. 声明队列
        channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);
        channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);

        // 5. 交换机和队列绑定
        channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, "a");
        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "a");
        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "b");
        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "c");

        // 6. 发送消息
        for (int i = 0; i < 3; i++) {
            channel.basicPublish(Constants.DIRECT_EXCHANGE, "a", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        for (int i = 0; i < 3; i++) {
            channel.basicPublish(Constants.DIRECT_EXCHANGE, "b", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        for (int i = 0; i < 3; i++) {
            channel.basicPublish(Constants.DIRECT_EXCHANGE, "c", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        System.out.println("消息发送成功");

        // 7. 资源释放
        channel.close();
        connection.close();
    }
}

/**
 * 消费者 A
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerA {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.DIRECT_QUEUE1, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

/**
 * 消费者 B
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerB {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.DIRECT_QUEUE2, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

3.5 Topics

核心特征：

• 根据路由键选择性接收消息
• 支持通配符匹配
• 通配符规则
  • *(星号)：匹配一个单词
  • #(井号)：匹配零个或多个单词
  • 单词用点号分隔，如 stock.usd.nyse

工作流程：

代码实现：与 Routing 模式基本一致

/**
 * 生产者
 */
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        // BuiltinExchangeType.TOPIC: 通配符类型，把消息交给符合 routing pattern(路由模式) 的队列
        channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);

        // 4. 声明队列
        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);
        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);

        // 5. 交换机和队列绑定
        channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, "*.a.*");
        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "*.*.b");
        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "c.#");

        // 6. 发送消息
        for (int i = 0; i < 3; i++) {
            // 转发到 queue1
            channel.basicPublish(Constants.TOPIC_EXCHANGE, "aa.a.f", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        for (int i = 0; i < 3; i++) {
            // 转发到 queue1 和 queue2
            channel.basicPublish(Constants.TOPIC_EXCHANGE, "aa.a.b", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        for (int i = 0; i < 3; i++) {
            // 转发到 queue2
            channel.basicPublish(Constants.TOPIC_EXCHANGE, "c.a.a.a", null, ("Hello RabbitMQ!" + i).getBytes());
        }
        System.out.println("消息发送成功");

        // 7. 资源释放
        channel.close();
        connection.close();
    }
}

/**
 * 消费者 A
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerA {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.TOPIC_QUEUE1, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

/**
 * 消费者 B
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerB {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);

        // 4. 消费消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
            }
        };
        channel.basicConsume(Constants.TOPIC_QUEUE2, true, defaultConsumer);
        Thread.sleep(2000);
    }
}

3.6 RPC

RPC(Remote Procedure Call，远程过程调用)：通过网络从远程计算机上请求服务并同步等待服务端的响应。

核心特征：

• 请求/响应模式
• 实现远程过程调用，使用回调队列和关联 ID

工作流程：

• 当客户端启动时，它创建一个独占回调队列。
• 对于 RPC 请求，客户端发送一个具有两个属性的消息：reply_to，它被设置为回调队列，correlation_id，它被设置为每个请求的唯一值。
• 请求被发送到 rpc_queue 队列。
• RPC 工作者 (又名：服务器) 正在等待队列上的请求。当出现请求时，它完成任务，并使用 replyTo 字段中的队列将带有结果的消息发送回客户机。
• 客户端在应答队列上等待数据。当消息出现时，它会检查 correlationId 属性。如果与请求的值匹配，则将响应返回给应用程序。

代码实现：

/**
 * 客户端
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class RpcClient {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);
        channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);

        // 4. 发送请求
        String correlationID = UUID.randomUUID().toString();
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().correlationId(correlationID).replyTo(Constants.RPC_RESPONSE_QUEUE).build();
        channel.basicPublish("", Constants.RPC_REQUEST_QUEUE, properties, ("Hello RabbitMQ!").getBytes());

        // 5. 接收响应
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                // TODO
                System.out.println("接收到消息:" + new String(body));
                if (correlationID.equals(properties.getCorrelationId())) {
                    if (!response.offer(new String(body))) {
                        System.out.println("队列满");
                    }
                }
            }
        };
        channel.basicConsume(Constants.RPC_RESPONSE_QUEUE, true, defaultConsumer);
        System.out.println("[RPC Client 响应结果]:" + response.take());
    }
}

/**
 * 服务器
 */
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class RpcServer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();

        // 2. 开启信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);
        channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);
        channel.basicQos(1);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // TODO
                String request = new String(body, StandardCharsets.UTF_8);
                System.out.println("接收到消息:" + request);
                System.out.println("针对请求:" + request + ",响应成功");
                AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();

                // 5. 发送响应
                channel.basicPublish("", Constants.RPC_RESPONSE_QUEUE, basicProperties, request.getBytes());

                // 手动确认
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };

        // 4. 接收请求
        channel.basicConsume(Constants.RPC_REQUEST_QUEUE, false, defaultConsumer);
    }
}