rabbitmq(6)路由模式

叁歲伎倆 2022-05-29 14:08 195阅读 0赞

### 1、模型 ###

路由模式为升级版的订阅模式，增加了消费者选择性接收消息的功能。  
每个消费者可以绑定多个routingKey，生产者在发送消息时指定routingKey致使只有绑定了该routingKey的消费者才能接收到此消息。  
![这里写图片描述][20180325162858408]

### 2.代码 ###

#### 2.1、生产者 ####

#### 2.2、消费者1 ####

private final static String EXCHANGE_NAME = "test_direct_exchange";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.1.7");
        factory.setUsername("root");
        factory.setPassword("123456");
    
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();
    
        channel.queueBind(queueName, EXCHANGE_NAME, "key1");
    
        System.out.println(" [1] Waiting for messages. ");
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [1] Received '" + message + "'");
    
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, consumer);
    }

#### 2.2、消费者2 ####

private final static String EXCHANGE_NAME = "test_direct_exchange";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.1.7");
        factory.setUsername("root");
        factory.setPassword("123456");
    
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();
    
        // 一个消费者可以绑定多个routingKey
        channel.queueBind(queueName, EXCHANGE_NAME, "key1");
        channel.queueBind(queueName, EXCHANGE_NAME, "key2");
        channel.queueBind(queueName, EXCHANGE_NAME, "key3");
    
        System.out.println(" [2] Waiting for messages.");
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [2] Received '" + message + "'");
    
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, consumer);
    }

[20180325162858408]: /images/20220529/0e722d77af684e26b188cac1a6a16afa.png