Use of message queues

3.1 RabbitMQ At the heart of ：

There are introductions on the core official website , Yes connecnton,channel And so on. , What's going on ,who care？

overall , Our focus on business implementation is ：1） How was the message delivered .2） How do consumers consume news .3） Whether the message is delivered reliably .4） Message delivery method .5） The life cycle of a message .6） Message queue lifecycle

3.2  How the message was delivered ？（ Remember a little , Producer message delivery is switch oriented ）

RabbitMQ It is aimed at delivering messages to the switch . The switch may be bound with many queues , How does the switch deliver messages to these queues ？

First, let's talk about the advantages of the switch design ：1） This obviously relies on the design idea of the switch at the data link layer . In addition to the clear hierarchy , It can also improve link utilization （ It may be a little like ）.

2） At the code level ： If there is no switch , You must at least maintain a very large routing table , Then the message is delivered correctly from the routing table , With the interactive machine , This way

The table will be split into multiple switches , The effect is needless to say .

   3） Then there is a high degree of decoupling , Different switches can have different routing rules , Without a switch ......

stay RabbitMQ, The switch has 4 Two delivery methods , Enumeration class BuiltinExchangeType Of 4 Enumeration variables ：

    DIRECT: All messages will be retrieved first ROUTE_KEY, And then deliver it to ROUTE_KEY In the bound queue （if(msg.routekey.equals(queue.routekey))）.

FANOUT: In this mode , Don't check messages at all ROUTE_KEY, Directly deliver to all queues owned by the switch .

TOPIC,HEADERS Take a look at what the official website says , I don't want to code anymore ^_^||

To sum up, a function sends the message :channel.basicPublish(excange_name,route_key,false,bs,"test".getBytes()); You can check this on the official website API

3.3 How do consumers consume news （ Remember a little , Consumer consumption messages are message queue oriented , This is a little different from the generator ）

Not yet TCP Those things that connect the heartbeat for a long time , That's the one API：channel.basicConsume(QUEUE_AUTODELETE, true, consumer);consumer yes Consumer An instance of a class ,

You can handle the callback interface directly ok 了

3.4 Whether the message delivery is reliable

It's obviously reliable , Unless you queue messages , Declare as a non persistent pattern , You restart the machine again . This will lose the message . And he has a response mechanism , You can set the mode of consumer consumption messages ,

Go to answer manually .channel.basicConsume（？,autoACk,?） Of autoAck Parameter setting

3.5 The life cycle of a message

Once the consumer answers , The identity message has been consumed , The message is recycled .

3.6 Queue lifecycle

channel.queueDeclare(QUEUE_NAME,false,false,true,null);

The second parameter is set to true, Will persist messages to disk , The fourth parameter is set to true It means that if there is no message and no connection, the queue will be deleted , You can check the details API

Four 、 An example
4.1 Producer code ：

Import dependent packages or dependencies by yourself

 

Copy code
  ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("username");
        factory.setPort(5672);// Note that the port here is different from that of the management plug-in
        factory.setPassword("pwd");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // Make a statement dirent Mode switch
        channel.exchangeDeclare("exchange_name",BuiltinExchangeType.DIRECT,true);
        // Declare a non persistent auto delete queue
        channel.queueDeclare("queue_name",false,false,true,null);// Delete the queue if it is no longer in use zhe
        // Bind to switch
        channel.queueBind("queue_name","exchange_name","route_key");
        // Declare a message header
        Map<String,Object> header=new HashMap<>();
        AMQP.BasicProperties.Builder b= new AMQP.BasicProperties.Builder();
        header.put("charset","utf-8");
        b.headers(header);
        AMQP.BasicProperties bp=b.build();
        // Send the message
       channel.basicPublish("exchange_name","route_key",false,bp,"test3".getBytes());
Copy code
4.2 Consumer code

ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("localhost");
  factory.setUsername("username");
  factory.setPort(5672);// Note that the port here is different from that of the management plug-in
  factory.setPassword("pwd");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  // Make a statement dirent Mode switch
  channel.exchangeDeclare("exchange_name",BuiltinExchangeType.DIRECT,true);
  // Declare a non persistent auto delete queue
  channel.queueDeclare("queue_name",false,false,true,null);// Delete the queue if it is no longer in use zhe
  // Bind to switch
  channel.queueBind("queue_name","exchange_name","route_key");
  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 '" + message + "'");
      }
  };
  channel.basicConsume("queue_name", true, consumer);