Topic路由模式

原本理解的不够彻底，程序总是不太对。所以查询了资料，关于这种模式的意思做了仔细的解读，复制的文字。

一：介绍

1.模式

2.知识点

　　其中，#可以匹配一个或者多个字符

　　其中，*可以匹配一个字符

3.仔细解读上面的图

在上图例子中，我们发送描述动物的消息。消息会转发给包含3个单词（2个小数点）的路由键绑定的队列中。绑定键中的第一个单词描述的是速度，第二个是颜色，第三个是物种：“速度.颜色.物种”。

我们创建3个绑定：Q1绑定键是“*.orange.*”，Q2绑定键是“*.*.rabbit”，Q3绑定键是“lazy.#”。这些绑定可以概括为：Q1只对橙色的动物感兴趣。Q2则是关注兔子和所有懒的动物。

路由键为“quick.orange.rabbit”的消息会被路由到2个队列中去。而“lazy.orange.elephant”的消息同样会发往2个队列。另外“quick.orange.fox” 仅仅发往第一个队列，而"lazy.brown.fox"则只发往第二个队列。“quick.brown.fox”则所有的绑定键都不匹配而被丢弃。

如果我们违反约定，发送了只带1个或者4个标识符的选择键，像“orange”或者“quick.orange.male.rabbit”，会发生什么呢？这些消息都不匹配任何绑定，所以将被丢弃。

另外，“lazy.orange.male.rabbit”，尽管有4个标识符，但是仍然匹配最后一个绑定键，所以会发送到第二个队列中。

二：程序

1.生产者

 package com.mq.topic;

 import com.mq.utils.ConnectionUtil;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 public class TopicSend {

     private static final String EXCHANGE_NAME="test_exchange_topic";

     public static void main(String[] args)throws Exception{

         Connection connection= ConnectionUtil.getConnection();

         Channel channel=connection.createChannel();

         channel.exchangeDeclare(EXCHANGE_NAME,"topic");

         String msg="hello topic";

         //重点是第二个参数routingKey

         String routingKey="商品";

         channel.basicPublish(EXCHANGE_NAME,"good.query.pro",null,msg.getBytes());

         System.out.println("msg send:"+msg);

         channel.close();

         connection.close();

     }

 }

2.消费者一

 package com.mq.topic;

 import com.mq.utils.ConnectionUtil;

 import com.rabbitmq.client.*;

 import java.io.IOException;

 public class TopicReceive1 {

     private static final String EXCHANGE_NAME="test_exchange_topic";

     private static final String QUEUE_NAME="test_queue_topic_1";

     public static void main(String[] args)throws Exception{

         Connection connection= ConnectionUtil.getConnection();

         final Channel channel=connection.createChannel();

         channel.queueDeclare(QUEUE_NAME,false,false,false,null);

         //good.query.pro

         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"good.#");

         channel.basicQos(1);

         Consumer consumer=new DefaultConsumer(channel){

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                 String msg = new String(body, "utf-8");

                 System.out.println("[1] receive:" + msg);

                 try {

                     Thread.sleep(200);

                 }catch (Exception e){

                     e.printStackTrace();

                 }finally {

                     System.out.println("[done]");

                     channel.basicAck(envelope.getDeliveryTag(),false);

                 }

             }

         };

         boolean autoAck=false;

         channel.basicConsume(QUEUE_NAME,autoAck,consumer);

     }

 }

3.消费者二

 package com.mq.topic;

 import com.mq.utils.ConnectionUtil;

 import com.rabbitmq.client.*;

 import java.io.IOException;

 public class TopicReceive2 {

     private static final String EXCHANGE_NAME="test_exchange_topic";

     private static final String QUEUE_NAME="test_queue_topic_2";

     public static void main(String[] args)throws Exception{

         Connection connection= ConnectionUtil.getConnection();

         final Channel channel=connection.createChannel();

         channel.queueDeclare(QUEUE_NAME,false,false,false,null);

         //good.query.pro

         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"good.add.*");  //不可以

         channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"good.query.*");//可以

         channel.basicQos(1);

         Consumer consumer=new DefaultConsumer(channel){

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                 String msg = new String(body, "utf-8");

                 System.out.println("[2] receive:" + msg);

                 try {

                     Thread.sleep(200);

                 }catch (Exception e){

                     e.printStackTrace();

                 }finally {

                     System.out.println("[done]");

                     channel.basicAck(envelope.getDeliveryTag(),false);

                 }

             }

         };

         boolean autoAck=false;

         channel.basicConsume(QUEUE_NAME,autoAck,consumer);

     }

 }

巴特西

Topic路由模式

最新文章

热门文章