RabbitMQ-主题模式Topic

分手后的思念是犯贱 2022-05-30 11:40 176阅读 0赞

### 文章目录 ###

*  RabbitMQ-主题模式Topic
 *   *   *   *  Topic Exchange
             *   *  生产者
                 *  消费者1
                 *  消费者2

# RabbitMQ-主题模式Topic #

#### Topic Exchange ####

>  将路由和某个模式匹配，\# 匹配一个或者多个，\* 匹配一个。例如 Good.insert Good.delete，则Good.\#都能获得  
> ![这里写图片描述][70]  
> ![这里写图片描述][70 1]

>  商品 : 发布 删除 修改 查询…

##### 生产者 #####

package com.ithzk.rabbitmq.topic;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Send { 
    
        private final static String EXCHANGER_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            Channel channel = connection.createChannel();
    
            // 声明交换机
            channel.exchangeDeclare(EXCHANGER_NAME,"topic");
    
            String msg = "goods......";
    
            //发送goods.add 消费者1和消费者2都可以收到消息
            //发送goods.delete 消费者2收到消息
            String routingKey = "goods.delete";
            channel.basicPublish(EXCHANGER_NAME,routingKey,null,msg.getBytes());
    
            System.out.println("Send msg"+msg);
    
            channel.close();
            connection.close();
        }
    
    }

##### 消费者1 #####

package com.ithzk.rabbitmq.topic;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv1 { 
    
        private static final String QUEUE_NAME="test_queue_topic_one";
        private final static String EXCHANGER_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            String routingKey = "goods.add";
            //绑定队列到交换机 转发器
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer 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] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[1] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 1 start]");
    
    
        }
    
    
    }

##### 消费者2 #####

package com.ithzk.rabbitmq.topic;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv2 { 
    
        private static final String QUEUE_NAME="test_queue_topic_two";
        private final static String EXCHANGER_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            //绑定队列到交换机 转发器
            String routingKey = "goods.#";
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer 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] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[2] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 2 start]");
    
    
        }
    
    
    }

[70]: /images/20220530/6a619566eba4429f8d38878b6a3cb5c8.png
[70 1]: /images/20220530/c54866702f1645cdac7bda74839a5800.png