RabbitMQ：Topics主题/通配符模式

Posted 2022-10-10 不断前进的皮卡丘

✨ RabbitMQ：Topics主题/通配符模式

• 1.基本介绍
• 2.生产者
• 3.消费者
• 4.测试

📃个人主页:不断前进的皮卡丘
🌞博客描述:梦想也许遥不可及，但重要的是追梦的过程，用博客记录自己的成长，记录自己一步一步向上攀登的印记
🔥个人专栏:消息中间件

1.基本介绍

• Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符
• Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
• 通配符规则:
  • #:匹配0个或者多个词
  • *:刚好可以匹配一个词

2.生产者

public class Producer 
    public static String TOPIC_EXCHANGE = "topic_exchange";
    public static String TOPIC_QUEUE_1 = "topic_queue_1";
    public static String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) 
        try 
            Channel channel = ConnectUtil.getChannel();
            //声明交换机(交换机名称，交换机类型)
            channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);
            //声明队列
            channel.queueDeclare(TOPIC_QUEUE_1,true,false,false,null);
            channel.queueDeclare(TOPIC_QUEUE_2,true,false,false,null);
            //把交换机和队列1进行绑定
            channel.queueBind(TOPIC_QUEUE_1,TOPIC_EXCHANGE,"#.error");
            //把交换机和队列2进行绑定
            channel.queueBind(TOPIC_QUEUE_2,TOPIC_EXCHANGE,"order.*");
            channel.queueBind(TOPIC_QUEUE_2,TOPIC_EXCHANGE,"*.orange.*");
            channel.queueBind(TOPIC_QUEUE_2,TOPIC_EXCHANGE,"*.*");
            //发送消息
            String msg="日志信息:调用了xxx方法，日志级别是error";
             channel.basicPublish(TOPIC_EXCHANGE,"error",null,msg.getBytes());
            System.out.println("消息发送成功");
         catch (IOException e) 
            e.printStackTrace();
         catch (TimeoutException e) 
            e.printStackTrace();
        


    

3.消费者

消费者1

public class Consumer1 
    public static void main(String[] args) 

        try 
            //获取信道对象
            Channel channel = ConnectUtil.getChannel();
            //消费消息
            DefaultConsumer consumer=new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    System.out.println("消费者1接收到消息:"+new String(body,"UTF-8"));
                    System.out.println("消费者1把日志信息保存到数据库");
                
            ;
            channel.basicConsume(Producer.TOPIC_QUEUE_1,true,consumer);



         catch (IOException e) 
            e.printStackTrace();
         catch (TimeoutException e) 
            e.printStackTrace();
        
    

消费者2

public class Consumer2 
    public static void main(String[] args) 

        try 
            //获取信道对象
            Channel channel = ConnectUtil.getChannel();
            //消费消息
            DefaultConsumer consumer=new DefaultConsumer(channel)
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                    System.out.println("消费者2接收到消息:"+new String(body,"UTF-8"));
                    System.out.println("消费者2把日志信息保存到数据库");
                
            ;
            channel.basicConsume(Producer.TOPIC_QUEUE_2,true,consumer);



         catch (IOException e) 
            e.printStackTrace();
         catch (TimeoutException e) 
            e.printStackTrace();
        
    

4.测试