RabbitMQ消息模型之Direct消息模型

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){@Overridepublic 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){@Overridepublic 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);}
}