RabbitMQ的Direct Exchange模式实现的消息发布案例(示例代码)

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQProducer {
    private final static String EXCHANGE_NAME = "direct_message_exchange";
    private final static String EXCHANGE_TYPE = "direct";
    public static void main(String[] args) {
        // 1. 创建连接工厂，设置连接参数
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672); // RabbitMQ默认端口
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            // 2. 声明交换机 (direct类型，持久化)
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true);
            // 3. 声明队列 (持久化，非独占，连接断开时不自动删除)
            channel.queueDeclare("queue5", true, false, false, null);
            channel.queueDeclare("queue6", true, false, false, null);
            channel.queueDeclare("queue7", true, false, false, null);
            // 4. 绑定队列到交换机，设置路由键
            channel.queueBind("queue5", EXCHANGE_NAME, "order");
            channel.queueBind("queue6", EXCHANGE_NAME, "order");
            channel.queueBind("queue7", EXCHANGE_NAME, "course");
            // 5. 准备要发送的消息
            String message = "你好，学相伴：www.kuangstudy.com";
            // 6. 向交换机发送消息，使用路由键 "course"
            channel.basicPublish(EXCHANGE_NAME, "course", null, message.getBytes("UTF-8"));
            System.out.println("消息发送成功！");
        } catch (Exception ex) {
            // 捕获异常并打印堆栈信息
            ex.printStackTrace();
            System.out.println("消息发送出现异常...");
        } finally {
            // 在try-with-resources中，不再需要显式关闭连接和通道
            // 会自动关闭连接和通道
        }
    }
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
public class RabbitMQConsumer {
    private final static String QUEUE_NAME = "queue7"; // 与生产者的绑定一致
    private final static String EXCHANGE_NAME = "direct_message_exchange";
    private final static String EXCHANGE_TYPE = "direct";
    public static void main(String[] args) {
        // 1. 创建连接工厂，设置连接参数
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672); // RabbitMQ默认端口
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            // 2. 声明交换机和队列，与生产者保持一致
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true);
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            // 3. 绑定队列到交换机，路由键为"course"
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "course");
            System.out.println(" [*] 等待接收消息...");
            // 4. 定义接收消息的回调函数
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] 接收到的消息: '" + message + "'");
                // 这里可以添加进一步的消息处理逻辑
            };
            // 5. 开始消费消息 (自动应答)
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        } catch (Exception ex) {
            // 捕获异常并打印堆栈信息
            ex.printStackTrace();
            System.out.println("消费者运行中出现异常...");
        }
    }
}