RabbitMQ消息模型之Direct消息模型

news2024/9/21 20:31:39

Direct消息模型

* 路由模型：
*    一个交换机可以绑定多个队列
*    生产者给交换机发送消息时，需要指定消息的路由键
*    消费者绑定队列到交换机时，需要指定所需要消费的信息的路由键
*    交换机会根据消息的路由键将消息转发到对应的队列

*    缺点：
*      当消息很多的时候，需要指定的路由键也会很多，究极复杂。

生产者

package com.example.demo02.mq.direct;

import com.example.demo02.mq.util.ConnectionUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;

/**
 * @author Allen
 * 4/11/2024 9:30 AM
 * @version 1.0
 * @description: 路由模型发送者
 *
 * 路由模型：
 *    一个交换机可以绑定多个队列
 *    生产者给交换机发送消息时，需要指定消息的路由键
 *    消费者绑定队列到交换机时，需要指定所需要消费的信息的路由键
 *    交换机会根据消息的路由键将消息转发到对应的队列

 *    缺点：
 *      当消息很多的时候，需要指定的路由键也会很多，究极复杂。
 */
public class DirectSender {
    public static void main(String[] args) throws Exception {
        // 1：创建连接
        Connection connection = ConnectionUtils.getConnection();
        // 2：创建通道
        Channel channel = connection.createChannel();
        // 3：声明交换机   参数1：交换机名称 参数2：交换机类型 参数3：是否持久化
        channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT, false);

        // 6：发送消息
        String msg1 = "{To DirectReceiver1: orderId:1001}";
        String msg2 = "{To DirectReceiver2: orderId:1002}";
        // 参数1：交换机 参数2：路由键(与消费者相匹配) 参数3：其他参数 参数4：消息内容
        channel.basicPublish("direct.exchange","order.save",null,msg1.getBytes());
        channel.basicPublish("direct.exchange","order.update",null,msg2.getBytes());
        // 7：关闭通道
        channel.close();
        // 8：关闭连接
        connection.close();
    }
}

消费者1

package com.example.demo02.mq.direct;

import com.example.demo02.mq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @author Allen
 * 4/11/2024 9:44 AM
 * @version 1.0
 * @description: 路由模型接收者1
 */
public class DirectReceiver1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT, false);

        channel.queueDeclare("direct.queue1", false, false, false, null);

        channel.queueBind("direct.queue1","direct.exchange","order.save");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println("DirectReceiver1接收到的新增订单消息是：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("direct.queue1",false,consumer);
    }
}

消费者2

package com.example.demo02.mq.direct;

import com.example.demo02.mq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @author Allen
 * 4/11/2024 9:44 AM
 * @version 1.0
 * @description: 路由模型接收者2
 */
public class DirectReceiver2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtils.getConnection();

        Channel channel = connection.createChannel();

        channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT, false);

        channel.queueDeclare("direct.queue2", false, false, false, null);

        channel.queueBind("direct.queue2","direct.exchange","order.update");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                System.out.println("DirectReceiver2接收到的修改订单消息：" + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("direct.queue2",false,consumer);
    }
}