✨ RabbitMQ:路由模式
📃个人主页:不断前进的皮卡丘
🌞博客描述:梦想也许遥不可及,但重要的是追梦的过程,用博客记录自己的成长,记录自己一步一步向上攀登的印记
🔥个人专栏:消息中间件
1.基本介绍
在路由工作模式中,我们需要配置一个类型为direct的交换机,并且需要指定不同的路由键(routing key),把对应的消息从交换机路由到不同的消息队列进行存储,由消费者进行消费。
- P:生产者,向交换机发送消息的时候,会指定一个routing key
- X:Exchange(交换机),接收生产者的消息,然后把消息传递给和routing key完全匹配的队列
- C1:消费者,它所在队列指定了需要routing key为error的信息
- C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息
路由模式的特点
队列和交换机的绑定是需要指定routing key的,不可以随意绑定
消息的发送方向交换机发送消息的时候,也需要指定消息的routing key
交换机不再把消息交给每一个绑定的队列,而是根据消息的routing key来进行判断,只有队列的routing key和消息的routing key完全一样才会接收到消息。
2.生产者
publicclassProducer{publicstaticString DIRECT_EXCHANGE =" direct_exchange";publicstaticString DIRECT_QUEUE_1 ="direct_queue_1";publicstaticString DIRECT_QUEUE_2 ="direct_queue_2";publicstaticvoidmain(String[] args){try{Channel channel =ConnectUtil.getChannel();//声明交换机(交换机名称,交换机类型)
channel.exchangeDeclare(DIRECT_EXCHANGE,BuiltinExchangeType.DIRECT);//声明队列
channel.queueDeclare(DIRECT_QUEUE_1,true,false,false,null);
channel.queueDeclare(DIRECT_QUEUE_2,true,false,false,null);//把交换机和队列1进行绑定
channel.queueBind(DIRECT_QUEUE_1,DIRECT_EXCHANGE,"error");//把交换机和队列2进行绑定
channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHANGE,"info");
channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHANGE,"error");
channel.queueBind(DIRECT_QUEUE_2,DIRECT_EXCHANGE,"warning");//发送消息String msg="日志信息:调用了xxx方法,日志级别是info";
channel.basicPublish(DIRECT_EXCHANGE,"info",null,msg.getBytes());System.out.println("消息发送成功");}catch(IOException e){
e.printStackTrace();}catch(TimeoutException e){
e.printStackTrace();}}}
3.消费者
消费者1
publicclassConsumer1{publicstaticvoidmain(String[] args){try{//获取信道对象Channel channel =ConnectUtil.getChannel();//消费消息DefaultConsumer consumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println("消费者1接收到消息:"+newString(body,"UTF-8"));System.out.println("消费者1把日志信息保存到数据库");}};
channel.basicConsume(Producer.DIRECT_QUEUE_1,true,consumer);}catch(IOException e){
e.printStackTrace();}catch(TimeoutException e){
e.printStackTrace();}}}
消费者2
publicclassConsumer2{publicstaticvoidmain(String[] args){try{//获取信道对象Channel channel =ConnectUtil.getChannel();//消费消息DefaultConsumer consumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{System.out.println("消费者2接收到消息:"+newString(body,"UTF-8"));System.out.println("消费者2把日志信息输出到控制台");}};
channel.basicConsume(Producer.DIRECT_QUEUE_2,true,consumer);}catch(IOException e){
e.printStackTrace();}catch(TimeoutException e){
e.printStackTrace();}}}
4.测试
第一次测试,发送日志级别为info的信息
第二次测试,发送日志级别为error的信息
版权归原作者 不断前进的皮卡丘 所有, 如有侵权,请联系我们删除。