Topics(动态路由)

news2024/11/26 13:43:15

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列中。只不过Topic类型Exchange可以让队列在绑定路由时可以使用通配符。

*：匹配不多不少刚好一个单词。
#：匹配一个或多个词。
举例：
audit.#可以匹配audit.x.y或者dudit.x。
audit.*只能匹配audit.x.

1.开发生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        String exchange = "topicexchange";
        //将通道声明指定交换机 参数1：指明交换机名称  参数2：交换机的类型 fanout广播类型
        channel.exchangeDeclare(exchange,"topic");
        //发送消息
        String routingKey = "user.save.okk";
        channel.basicPublish(exchange,routingKey,null,("这是topicsexchange发布的"+routingKey+"消息").getBytes());
        //释放资源
        RabbitMqUtil.closeConnectionAndChannel(channel,connection);
    }
}

2.开发消费者(user.*)

public class Consumer1 {
    public static void main(String[] args) throws IOException {
        //获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        String exchange = "topicexchange";
        //通道绑定交换机
        channel.exchangeDeclare(exchange,"topic");
        //临时队列
        String queueName = channel.queueDeclare().getQueue();
        //基于RoutingKey绑定交换机和队列
        channel.queueBind(queueName,exchange,"user.*");
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumer1得到："+new String(body));
            }
        });
        //注意这里不能关闭通道和连接，因为要一直监听
    }
}

3.开发消费者(user.#)

public class Consumer2 {
    public static void main(String[] args) throws IOException {
        //获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        String exchange = "topicexchange";
        //通道绑定交换机
        channel.exchangeDeclare(exchange,"topic");
        //临时队列
        String queueName = channel.queueDeclare().getQueue();
        //基于RoutingKey绑定交换机和队列
        channel.queueBind(queueName,exchange,"user.#");
        channel.basicConsume(queueName,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumer2得到："+new String(body));
            }
        });
        //注意这里不能关闭通道和连接，因为要一直监听
    }
}