direct直连模式

左手的ㄟ右手 2022-06-11 07:29 243阅读 0赞

如果我们需要接收指定的消息，可以通过直连类型的交换机实现，队列1、2将仅接受key1的消息，而队列3、4将接收key2的消息。

![1240][]

生产者:

public class Producer {
        private static final String EXCHANGE_NAME = "aaa";
        // 路由关键字
        private static final String[] routingKeys = new String[]{"a" ,"b", "c"};
    
        public static void main(String[] argv) throws Exception {
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    //		声明交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    //		发送消息
            for(String severity :routingKeys){
                String message = "Send the message level:" + severity;
                channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
                System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
            }
            channel.close();
                connection.close();
        }
    }

消费者1:

public class Consumer01 {
            // 交换器名称
            private static final String EXCHANGE_NAME = "aaa";
            // 路由关键字
            private static final String[] routingKeys = new String[]{"a" ,"b", "c"};
            public static void main(String[] argv) throws Exception {
                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost("localhost");
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
    //		声明交换器
                channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    //		获取匿名队列名称
                String queueName = channel.queueDeclare().getQueue();
    //		根据路由关键字进行多重绑定
                for (String severity : routingKeys) {
                    channel.queueBind(queueName, EXCHANGE_NAME, severity);
                    System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity);
                }
                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(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
                    }
                };
                channel.basicConsume(queueName, true, consumer);
            }
    }

消费者2

public class Consumer02 {
        // 交换器名称
        private static final String EXCHANGE_NAME = "aaa";
        // 路由关键字
        private static final String[] routingKeys = new String[]{"c"};
        public static void main(String[] argv) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    //		声明交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    //		获取匿名队列名称
            String queueName = channel.queueDeclare().getQueue();
    //		根据路由关键字进行多重绑定
            for (String severity : routingKeys) {
                channel.queueBind(queueName, EXCHANGE_NAME, severity);
                System.out.println("ReceiveLogsDirect1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity);
            }
            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(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
                }
            };
            channel.basicConsume(queueName, true, consumer);
        }
    }

这里生产者产生的消息分别放到了两个队列中，队列1中的消息被消费了并不影响另外一个队列中的消息。

[1240]: /images/20220611/54c2d978b24240f796352a67c5f67c88.png