消息隊列的使用

3.1 RabbitMQ的核心：

核心官網有介紹，說的connecnton,channel之類的，到底怎麼樣，who care？

總體來看，我們關注業務實現是：1）消息怎麼投遞的。2）消費者怎麼消費消息。3）消息是否是可靠投遞。4）消息投遞方式。5）消息的生命周期。6）消息隊列生命周期

3.2  消息是怎麼投遞的？（記住一點，生產者消息投遞都是面向交換機的）

RabbitMQ 是面向交換機投遞消息的。交換機可能綁定有許多隊列，交換機如何將消息投遞給這些隊列呢？

首先說一下面向交換機的設計的優勢：1）這明顯借助了數據鏈路層那個交換機的設計思想。除了層級分明以外，還能從分提高鏈路利用率（可能有點抽像）。

2）從代碼層面來看：如果沒有交換機，你至少得維護一個十分龐大的路由錶，然後從路由錶正確投遞消息，有了交互機，這裏路

由錶就會被拆分到多個交換機裏面，效果不必多說。

   3）然後就是高度的解耦，不同的交換機可有不同的路由規則，要是沒有交換機。。。。。。

在RabbitMQ，交換機有4種投遞方式，就是枚舉類BuiltinExchangeType的4個枚舉變量：

    DIRECT:會將所有消息先取消息的ROUTE_KEY，然後投遞到與ROUTE_KEY綁定的隊列裏面（if(msg.routekey.equals(queue.routekey))）。

FANOUT:此種模式下，根本不檢查消息的ROUTE_KEY,直接投送到交換機所擁有的所有隊列裏面。

TOPIC,HEADERS自行看一下官網怎麼說的，不想碼字了^_^||

總結起來就一個函數就把消息發出去了:channel.basicPublish(excange_name,route_key,false,bs,"test".getBytes());可以去官網查一下這個API

3.3 消費者怎麼消費消息（記住一點，消費者消費消息是面向消息隊列的，這與生成者有點不一樣）

還不是就是TCP長連接心跳的那些事，就是這麼一個API：channel.basicConsume(QUEUE_AUTODELETE, true, consumer);consumer是Consumer類的一個實例，

你直接去處理回調接口就ok了

3.4 消息傳遞是否可靠

很明顯是可靠的，除非你將消息隊列，聲明成非持久模式，這事你又重啟了機器。這會丟失消息的。還有就是他有應答機制，你可以通過設置消費者消費消息的模式，

去手動應答。channel.basicConsume（？，autoACk,?）的autoAck參數設置

3.5 消息的生命周期

一旦受到消費者應答，標識消息已被消費，則消息被回收掉。

3.6 隊列生命周期

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

第二個參數設置為true，會將消息持久化到磁盤，第四個參數設置為true錶示沒有消息並且沒有連接則删除改隊列，詳情可以查一下API

四、一個示例
4.1 生產者代碼：

自行導入相關依賴包或相關依賴

 

複制代碼
  ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("username");
        factory.setPort(5672);//注意這裏的端口與管理插件的端口不一樣
        factory.setPassword("pwd");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //聲明一個dirent模式的交換機
        channel.exchangeDeclare("exchange_name",BuiltinExchangeType.DIRECT,true);
        //聲明一個非持久化自動删除的隊列
        channel.queueDeclare("queue_name",false,false,true,null);//如果該隊列不在被使用就删除他 zhe
        //將綁定到改交換機
        channel.queueBind("queue_name","exchange_name","route_key");
        //聲明一個消息頭部
        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();
        //將消息發出去
       channel.basicPublish("exchange_name","route_key",false,bp,"test3".getBytes());
複制代碼
4.2 消費者代碼

ConnectionFactory factory = new ConnectionFactory();
  factory.setHost("localhost");
  factory.setUsername("username");
  factory.setPort(5672);//注意這裏的端口與管理插件的端口不一樣
  factory.setPassword("pwd");
  Connection connection = factory.newConnection();
  Channel channel = connection.createChannel();
  //聲明一個dirent模式的交換機
  channel.exchangeDeclare("exchange_name",BuiltinExchangeType.DIRECT,true);
  //聲明一個非持久化自動删除的隊列
  channel.queueDeclare("queue_name",false,false,true,null);//如果該隊列不在被使用就删除他 zhe
  //將綁定到改交換機
  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);