RabbitMQ：Topics主题/通配符模式-灵析社区

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.测试

点赞 评论 收藏 目录

阅读量：2012

点赞量：0

收藏量:0