Direct exchange

  Preface

We built a simple logging system . We can broadcast log messages to many recipients . In this section, we will add some special features - For example, we only let a consumer subscribe to some of the published news . For example, we only store serious error messages in the log file ( To save disk space ), At the same time, all log messages can still be printed on the console .

Binding is the bridge between switch and queue . You can understand that ：

The queue is only interested in the messages of the switch to which it is bound . Binding parameters ：routingKey It can also be called binding key, To create a binding, we use code :channel.queueBind(queueName, EXCHANGE_NAME, "routingKey"); The meaning after binding is determined by its exchange type .

Direct exchange Introduce

We want programs that write log messages to disk to receive only serious errors (errros), Instead of storing warnings (warning) Or information (info) Log messages avoid wasting disk space .Fanout This type of exchange does not give us much flexibility - It can only broadcast unconsciously , Here we will use direct This type to replace , This type of work is , The message only goes to its bound routingKey The queue to .

 

In the picture above , We can see X Bound to two queues , The binding type is direct. queue Q1 The binding key is orange, queue Q2 The binding key has two : A binding key is black, The other binding key is green.

In this case of binding , The producer publishes a message to exchange On , The binding key is orange The message will be published to the queue

Q1. The binding key is blackgreen And messages will be published to the queue Q2, Messages of other message types will be discarded .

  Multiple bindings

Of course, if exchange The binding type of is direct, But it binds to multiple queues key If they are the same , In this case, although the binding type is direct But it behaves like fanout It's kind of like , It's like broadcasting , As shown in the figure below . 

 

Combat code  

 

Tool class ：

 

public class untils {
    public static Channel getChannel() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.231.132");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

consumer 1：

public class ReceiveLogsDirect01 {
    private  static  final  String EXCHANG_NAME="direct_logs";
    public static void main(String[] args) throws  Exception{
        Channel channel = untils.getChannel();
        /**
         *  Declaration switch 
         */
        channel.exchangeDeclare(EXCHANG_NAME, BuiltinExchangeType.DIRECT);
        String queueName="disk";
        channel.queueDeclare(queueName,false,false,false,null);
        channel.queueBind(queueName,EXCHANG_NAME,"error");
        System.out.println(" Preparing to receive messages ....");
        DeliverCallback deliverCallback=(Consumer,delivert)->
        {
            String s = new String(delivert.getBody());
            System.out.println(" Error message --->："+s);
        };
        channel.basicConsume(queueName,true,deliverCallback,consumerTag ->{} );
    }
}

  consumer 2：

public class ReceiveLogsDirect02 {
        private  static  final  String EXCHANG_NAME="direct_logs";
        public static void main(String[] args) throws  Exception{
            Channel channel = untils.getChannel();
            /**
             *  Declaration switch 
             */
            channel.exchangeDeclare(EXCHANG_NAME, BuiltinExchangeType.DIRECT);
            String queueName="console";
            channel.queueDeclare(queueName,false,false,false,null);
            channel.queueBind(queueName,EXCHANG_NAME,"info");
            channel.queueBind(queueName,EXCHANG_NAME,"warning");
            System.out.println(" Preparing to receive messages ....");
            DeliverCallback deliverCallback=(Consumer, delivert)->
            {
                String s = new String(delivert.getBody());
                System.out.println("info warning news --->："+s);
            };
            channel.basicConsume(queueName,true,deliverCallback,consumerTag ->{} );
        }
}

producer ：

 

public class EmitLogDirect {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] args)throws  Exception {
        Channel channel = untils.getChannel();
        // Create multiple bindingKeyMap
        Map<String,String> bindingKeyMap=new HashMap<>();
        bindingKeyMap.put("info"," Ordinary info news ");
        bindingKeyMap.put("warning"," Warning warning news ");
        bindingKeyMap.put("error"," error error news ");
        //debug No consumers receive this message , So I lost 
        bindingKeyMap.put("debug"," debugging debug news ");
        for (Map.Entry<String,String> bindingKeyEntry:bindingKeyMap.entrySet())
        {
            String bindingKey=bindingKeyEntry.getKey();
            String message=bindingKeyEntry.getValue();
            channel.basicPublish(EXCHANGE_NAME,bindingKey,null,message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" The producer sends a message "+message);
        }
    }
}

  result ：