哪些网站是做外贸生意的,职教集团网站建设方案,wordpress 虾米音乐插件,网站建设方维1、 订阅模型-Direct
• 有选择性的接收消息 • 在订阅模式中#xff0c;生产者发布消息#xff0c;所有消费者都可以获取所有消息。 • 在路由模式中#xff0c;我们将添加一个功能 - 我们将只能订阅一部分消息。 例如#xff0c;我们只能将重要的错误消息引导到日志文件…1、 订阅模型-Direct
• 有选择性的接收消息 • 在订阅模式中生产者发布消息所有消费者都可以获取所有消息。 • 在路由模式中我们将添加一个功能 - 我们将只能订阅一部分消息。 例如我们只能将重要的错误消息引导到日志文件以节省磁盘空间同时仍然能够在控制台上打印所有日志消息。 • 但是在某些场景下我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。 • 在Direct模型下队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key • 消息的发送方在向Exchange发送消息时也必须指定消息的routing key。
• P生产者向Exchange发送消息发送消息时会指定一个routing key。 • XExchange交换机接收生产者的消息然后把消息递交给 与routing key完全匹配的队列 • C1消费者其所在队列指定了需要routing key 为 error 的消息 • C2消费者其所在队列指定了需要routing key 为 info、error、warning 的消息
1.1、生产者
此处我们模拟商品的增删改发送消息的RoutingKey分别是insert、update、delete
public class Send {private final static String EXCHANGE_NAME direct_exchange_test;public static void main(String[] argv) throws Exception {// 获取到连接Connection connection ConnectionUtil.getConnection();// 获取通道Channel channel connection.createChannel();// 声明exchange指定类型为directchannel.exchangeDeclare(EXCHANGE_NAME, direct);// 消息内容String message 商品新增了 id 1001;// 发送消息并且指定routing key 为insert ,代表新增商品channel.basicPublish(EXCHANGE_NAME, insert, null, message.getBytes());System.out.println( [商品服务] Sent message );channel.close();connection.close();}
}1.2、消费者1
我们此处假设消费者1只接收两种类型的消息更新商品和删除商品。
public class Recv {private final static String QUEUE_NAME direct_exchange_queue_1;private final static String EXCHANGE_NAME direct_exchange_test;public static void main(String[] argv) throws Exception {// 获取到连接Connection connection ConnectionUtil.getConnection();// 获取通道Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机同时指定需要订阅的routing key。假设此处需要update和delete消息channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, update);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, delete);// 定义队列的消费者DefaultConsumer consumer new DefaultConsumer(channel) {// 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg new String(body);System.out.println( [消费者1] received : msg !);}};// 监听队列自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}1.3、 消费者2
我们此处假设消费者2接收所有类型的消息新增商品更新商品和删除商品。
public class Recv2 {private final static String QUEUE_NAME direct_exchange_queue_2;private final static String EXCHANGE_NAME direct_exchange_test;public static void main(String[] argv) throws Exception {// 获取到连接Connection connection ConnectionUtil.getConnection();// 获取通道Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换机同时指定需要订阅的routing key。订阅 insert、update、deletechannel.queueBind(QUEUE_NAME, EXCHANGE_NAME, insert);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, update);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, delete);// 定义队列的消费者DefaultConsumer consumer new DefaultConsumer(channel) {// 获取消息并且处理这个方法类似事件监听如果有消息的时候会被自动调用Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg new String(body);System.out.println( [消费者2] received : msg !);}};// 监听队列自动ACKchannel.basicConsume(QUEUE_NAME, true, consumer);}
}1.4、测试
我们分别发送增、删、改的RoutingKey发现结果
