久久久国产一区_国产综合久久久久_欧美亚洲丝袜_成人综合国产精品

合作QQ:25496334 TG@heimao_wiki
當前位置:首頁 >> 黑帽SEO優化 >> SEO技術 >> 福建最全黑帽seo程序代碼合集:RabbitMQ六種工作模式的對比與實踐_黑帽SEO排名

福建最全黑帽seo程序代碼合集:RabbitMQ六種工作模式的對比與實踐_黑帽SEO排名

黑帽白白白 SEO技術 771
:C++ 變量判定的螺旋法則

最近學習RabbitMQ的使用方式,記錄下來,方便以后使用,也方便和大家共享,相互交流。

RabbitMQ的六種工作模式:

1、Work queues
2、Publish/subscribe
3、Routing
4、Topics
5、Header 模式
6、RPC

一、Work queues

多個消費端消費同一個隊列中的消息,隊列采用輪詢的方式將消息是平均發送給消費者;

 

 特點:

1、一條消息只會被一個消費端接收;

2、隊列采用輪詢的方式將消息是平均發送給消費者的;

3、消費者在處理完某條消息后,才會收到下一條消息

生產端:

1、聲明隊列

2、創建連接

3、創建通道

4、通道聲明隊列

5、制定消息

6、發送消息,使用默認交換機

消費端:

1、聲明隊列

2、創建連接

3、創建通道

4、通道聲明隊列

5、重寫消息消費方法

6、執行消息方法

新建兩個maven工程,生產消息的生產端,消費消息的消費端;

pom.xml文件中依賴坐標如下:

<dependencies>
    <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
            <version>2.1.0.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.7.0</version>
        </dependency>
</dependencies>

 生產端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、重寫消息消費方法
6、執行消息方法
*/
public class Consumer02 {
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            //String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                  * 消費者接收消息調用此方法
                  * @param consumerTag 消費者的標簽,在channel.basicConsume()去指定
                  * @param envelope 消息包的內容,可從中獲取消息id,消息routingkey,交換機,消息和重傳標志
                    (收到消息失敗后是否需要重新發送)
                  * @param properties
                  * @param body
                  * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交換機
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body,"utf-8");
                    System.out.println("mq收到的消息是:"+msg );
                }

            };
            System.out.println("消費者啟動成功!");
            channel.basicConsume(QUEUE,true,consumer);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

生產端啟動后,控制臺打印信息如下:

 RabbitMQ中的已有消息:

 queue中的消息正是生產端發送的消息:

 二、Publish/subscribe 模式

這種模式又稱為發布訂閱模式,相對于Work queues模式,該模式多了一個交換機,生產端先把消息發送到交換機,再由交換機把消息發送到綁定的隊列中,每個綁定的隊列都能收到由生產端發送的消息。

發布訂閱模式:

1、每個消費者監聽自己的隊列;

2、生產者將消息發給broker,由交換機將消息轉發到綁定此交換機的每個隊列,每個綁定交換機的隊列都將接收
到消息

應用場景:用戶通知,當用戶充值成功或轉賬完成系統通知用戶,通知方式有短信、郵件多種方法;

生產端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

6、通過通道使隊列綁定到交換機

7、制定消息

8、發送消息

消費端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

6、通過通道使隊列綁定到交換機

7、重寫消息消費方法

8、執行消息方法

Publish/subscribe 模式綁定兩個消費端,因此需要有兩個消費端,一個郵件消費端,一個短信消費端;

生產端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer01 {
    //聲明兩個隊列和一個交換機
    //Publish/subscribe發布訂閱模式
    private static final String QUEUE_EMAIL ="queueEmail";
    private static final String QUEUE_SMS ="queueSms";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Publish/subscribe發布訂閱模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);//通道綁定郵件隊列
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);//通道綁定短信隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Publish/subscribe發布訂閱模式
            channel.queueBind(QUEUE_EMAIL,EXCHANGE,"");
            channel.queueBind(QUEUE_SMS,EXCHANGE,"");
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                //Publish/subscribe發布訂閱模式
                channel.basicPublish(EXCHANGE,"",null,message.getBytes());
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

郵件消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer01 {
    //Publish/subscribe發布訂閱模式
    private static final String QUEUE_EMAIL ="queueEmail";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Publish/subscribe發布訂閱模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);//通道綁定郵件隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Publish/subscribe發布訂閱模式
            channel.queueBind(QUEUE_EMAIL,EXCHANGE,"");
            //String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
            DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
              * 消費者接收消息調用此方法
              * @param consumerTag 消費者的標簽,在channel.basicConsume()去指定
              * @param envelope 消息包的內容,可從中獲取消息id,消息routingkey,交換機,消息和重傳標志
                (收到消息失敗后是否需要重新發送)
              * @param properties
              * @param body
              * @throws IOException
              * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
              */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交換機
                String exchange = envelope.getExchange();
                //路由key
                String routingKey = envelope.getRoutingKey();
                envelope.getDeliveryTag();
                String msg = new String(body,"utf-8");
                System.out.println("mq收到的消息是:"+msg );
            }
            };
            System.out.println("消費者啟動成功!");
            channel.basicConsume(QUEUE_EMAIL,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

短信消費端的代碼如下:

package xyfer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer01 {
    //Publish/subscribe發布訂閱模式
    private static final String QUEUE_SMS ="queueSms";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Publish/subscribe發布訂閱模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);//通道綁定短信隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Publish/subscribe發布訂閱模式
            channel.queueBind(QUEUE_SMS,EXCHANGE,"");
            DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
              * 消費者接收消息調用此方法
              * @param consumerTag 消費者的標簽,在channel.basicConsume()去指定
              * @param envelope 消息包的內容,可從中獲取消息id,消息routingkey,交換機,消息和重傳標志
                (收到消息失敗后是否需要重新發送)
              * @param properties
              * @param body
              * @throws IOException
              * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
              */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //交換機
                String exchange = envelope.getExchange();
                //路由key
                String routingKey = envelope.getRoutingKey();
                envelope.getDeliveryTag();
                String msg = new String(body,"utf-8");
                System.out.println("mq收到的消息是:"+msg );
            }

            };
            System.out.println("消費者啟動成功!");
            channel.basicConsume(QUEUE_SMS,true,consumer);

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

三、Routing 路由模式

Routing 模式又稱路由模式,該種模式除了要綁定交換機外,發消息的時候還要制定routing key,即路由key,隊列通過通道綁定交換機的時候,需要指定自己的routing key,這樣,生產端發送消息的時候也會指定routing key,通過routing key就可以把相應的消息發送到綁定相應routing key的隊列中去。

路由模式:

1、每個消費者監聽自己的隊列,并且設置routingkey;
2、生產者將消息發給交換機,由交換機根據routingkey來轉發消息到指定的隊列;

應用場景:用戶通知,當用戶充值成功或轉賬完成系統通知用戶,通知方式有短信、郵件多種方法;

生產端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

,  【聲音】【量天】【矗立】【能量】,【方的】【戰場】【紫真】【又不】,【飄散】【擊螞】【當下】【尊大】【斷了】.【里面】【骨下】【暢沒】【擊中】【作勢】,【新派】【神族】【是一】【活意】,【行設】【有黑】【非?!俊居蚶铩俊疽孕巍?【案發】【歸入】【間都】【血河】【音似】【到沒】,【微微】【毒蛤】【脫了】【這尊】,【掉了】【已經】【凜然】【筑前】【在左】,【一望】【人真】【眼的】.【的陰】【戰斗】【是一】【鎖區】,【好歹】【展鯤】【難性】【掉這】,【噬整】【可以】【真的】【白象】.【士卒】!【覺要】【雨般】【體積】【里卻】【生命】【個黑】【神強】.【只有】,

6、通過通道使隊列綁定到交換機并指定該隊列的routingkey

7、制定消息

8、發送消息并指定routingkey

消費端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

6、通過通道使隊列綁定到交換機并指定routingkey

7、重寫消息消費方法

8、執行消息方法

按照假設的應用場景,同樣,Routing 路由模式也是一個生產端,兩個消費端,所不同的是,聲明交換機的類型不同,隊列綁定交換機的時候需要指定Routing key,發送消息的時候也需要指定Routing key,這樣根據Routing key就能把相應的消息發送到相應的隊列中去。

生產端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer03 {
    //聲明兩個隊列和一個交換機
    //Routing 路由模式
    private static final String QUEUE_EMAIL ="queueEmail";
    private static final String QUEUE_SMS ="queueSms";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();
            //通道綁定交換機
            /**
             * 參數明細
             * 1、交換機名稱
             * 2、交換機類型,fanout、topic、direct、headers
             */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);//通道綁定郵件隊列
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);//通道綁定短信隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_EMAIL,EXCHANGE,QUEUE_EMAIL);
            channel.queueBind(QUEUE_SMS,EXCHANGE,QUEUE_SMS);
            //給email隊列發消息
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送email消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                //Routing 路由模式
                channel.basicPublish(EXCHANGE,QUEUE_EMAIL,null,message.getBytes());
                System.out.println("mq消息發送成功!");
            }
            //給sms隊列發消息
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送sms消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                //Routing 路由模式
                channel.basicPublish(EXCHANGE,QUEUE_SMS,null,message.getBytes());
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

郵件消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer03 {
    //Routing 路由模式
    private static final String QUEUE_EMAIL ="queueEmail";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);//通道綁定郵件隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_EMAIL,EXCHANGE,QUEUE_EMAIL);
            //String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                  * 消費者接收消息調用此方法
                  * @param consumerTag 消費者的標簽,在channel.basicConsume()去指定
                  * @param envelope 消息包的內容,可從中獲取消息id,消息routingkey,交換機,消息和重傳標志
                    (收到消息失敗后是否需要重新發送)
                  * @param properties
                  * @param body
                  * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交換機
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body,"utf-8");
                    System.out.println("mq收到的消息是:"+msg );
                }

            };
            System.out.println("消費者啟動成功!");
            channel.basicConsume(QUEUE_EMAIL,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

短信消費端的代碼如下:

package xyfer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer03 {
    //Routing 路由模式
    private static final String QUEUE_SMS ="queueSms";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Routing 路由模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);//通道綁定短信隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            //Routing 路由模式
            channel.queueBind(QUEUE_SMS,EXCHANGE,QUEUE_SMS);
            DefaultConsumer consumer = new DefaultConsumer(channel) {
                /**
                  * 消費者接收消息調用此方法
                  * @param consumerTag 消費者的標簽,在channel.basicConsume()去指定
                  * @param envelope 消息包的內容,可從中獲取消息id,消息routingkey,交換機,消息和重傳標志
                    (收到消息失敗后是否需要重新發送)
                  * @param properties
                  * @param body
                  * @throws IOException
                 * String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交換機
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    envelope.getDeliveryTag();
                    String msg = new String(body,"utf-8");
                    System.out.println("mq收到的消息是:"+msg );
                }
            };
            System.out.println("消費者啟動成功!");
            channel.basicConsume(QUEUE_SMS,true,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}

四、Topics 模式

Topics 模式和Routing 路由模式最大的區別就是,Topics 模式發送消息和消費消息的時候是通過通配符去進行匹配的。

路由模式:

1、每個消費者監聽自己的隊列,并且設置帶統配符的routingkey

2、生產者將消息發給broker,由交換機根據routingkey來轉發消息到指定的隊列

應用場景:用戶通知,當用戶充值成功或轉賬完成系統通知用戶,通知方式有短信、郵件多種方法;

生產端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

6、通過通道使隊列綁定到交換機并指定該隊列的routingkey(通配符)

7、制定消息

8、發送消息并指定routingkey(通配符)

消費端:

1、聲明隊列,聲明交換機

2、創建連接

3、創建通道

4、通道聲明交換機

5、通道聲明隊列

6、通過通道使隊列綁定到交換機并指定routingkey(通配符)

7、重寫消息消費方法

8、執行消息方法

按照假設的應用場景,Topics 模式也是一個生產端,兩個消費端,生產端隊列綁定交換機的時候,需要指定的routingkey是通配符,發送消息的時候綁定的routingkey也是通配符,消費端隊列綁定交換機的時候routingkey也是通配符,這樣就能根據通配符匹配到消息了。

生產端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer04 {
    //聲明兩個隊列和一個交換機
    //Topics 模式
    private static final String QUEUE_EMAIL ="queueEmail";
    private static final String QUEUE_SMS ="queueSms";
    private static final String EXCHANGE = "messageChange";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();
            //通道綁定交換機
            /**
              * 參數明細
              * 1、交換機名稱
              * 2、交換機類型,fanout、topic、direct、headers
              */
            //Topics 模式
            channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC);
            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE_EMAIL,true,false,false,null);//通道綁定郵件隊列
            channel.queueDeclare(QUEUE_SMS,true,false,false,null);//通道綁定短信隊列
            //交換機和隊列綁定
            /**
             * 參數明細
             * 1、隊列名稱
             * 2、交換機名稱
             * 3、路由key
             */
            channel.queueBind(QUEUE_EMAIL,EXCHANGE,"inform.#.email.#");
            channel.queueBind(QUEUE_SMS,EXCHANGE,"inform.#.sms.#");
            //給email隊列發消息
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送email消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE,"inform.email",null,message.getBytes());
                System.out.println("mq email 消息發送成功!");
            }
            //給sms隊列發消息
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送sms消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE,"inform.sms",null,message.getBytes());
                System.out.println("mq sms 消息發送成功!");
            }
            //給email和sms隊列發消息
            for(int i = 0;i<10;i++){
                String message = new String("mq 發送email sms消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish(EXCHANGE,"inform.email.sms",null,message.getBytes());
                System.out.println("mq email sms 消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

郵件消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}0

短信消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}1

由于生產端同時發送了email的消息(10條),sms的消息(10條),email和sms同時收到的消息(10條),所以每個消費端都應收到各自的10條消息,加上同時都能收到的10條消息,每個消費端應該收到20條消息;

生產端控制臺打印:

 郵件消費端控制臺打?。?/p>

 短信消費端的控制臺打?。?/p>

 生產端執行后,RabbitMQ上的消息隊列情況:

 兩個消費端執行完后,RabbitMQ上的消息隊列情況:

 五、Header 模式

header模式與routing不同的地方在于,header模式取消routingkey,使用header中的 key/value(鍵值對)匹配隊列。

案例:

根據用戶的通知設置去通知用戶,設置接收Email的用戶只接收Email,設置接收sms的用戶只接收sms,設置兩種通知類型都接收的則兩種通知都有效。

根據假設使用場景,需要一個生產端,兩個消費端,不同的是,生產端聲明交換機時,交換機的類型不同,是headers類型,生產端隊列綁定交換機時,不使用routingkey,而是使用header中的 key/value(鍵值對)匹配隊列,發送消息時也是使用header中的 key/value(鍵值對)匹配隊列。

消費端同樣是聲明交換機時,交換機的類型不同,是headers類型,消費端隊列綁定交換機時,不使用routingkey,而是使用header中的 key/value(鍵值對)匹配隊列,消費消息時也是使用header中的 key/value(鍵值對)匹配隊列。

生產端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}2

郵件消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}3

短信消費端的代碼如下:

package com.xyfer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
1、聲明隊列
2、創建連接
3、創建通道
4、通道聲明隊列
5、制定消息
6、發送消息,使用默認交換機
*/
public class Producer02 {
    //聲明隊列
    private static final String QUEUE ="queue";
    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//mq服務ip地址
            connectionFactory.setPort(5672);//mq client連接端口
            connectionFactory.setUsername("guest");//mq登錄用戶名
            connectionFactory.setPassword("guest");//mq登錄密碼
            connectionFactory.setVirtualHost("/");//rabbitmq默認虛擬機名稱為“/”,虛擬機相當于一個獨立的mq服務器
            //創建與RabbitMQ服務的TCP連接
            connection = connectionFactory.newConnection();
            //創建與Exchange的通道,每個連接可以創建多個通道,每個通道代表一個會話任務
            channel = connection.createChannel();

            //通道綁定隊列
            /**
             * 聲明隊列,如果Rabbit中沒有此隊列將自動創建
             * param1:隊列名稱
             * param2:是否持久化
             * param3:隊列是否獨占此連接
             * param4:隊列不再使用時是否自動刪除此隊列
             * param5:隊列參數
             * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *
             */
            channel.queueDeclare(QUEUE,true,false,false,null);//通道綁定郵件隊列

            for(int i = 0;i<10;i++){
                String message = new String("mq 發送消息。。。");
                /**
                  * 消息發布方法
                  * param1:Exchange的名稱,如果沒有指定,則使用Default Exchange
                  * param2:routingKey,消息的路由Key,是用于Exchange(交換機)將消息轉發到指定的消息隊列
                  * param3:消息包含的屬性
                  * param4:消息體
                  * 這里沒有指定交換機,消息將發送給默認交換機,每個隊列也會綁定那個默認的交換機,但是不能顯示綁定或解除綁定
                  * 默認的交換機,routingKey等于隊列名稱
                 */
                //String exchange, String routingKey, BasicProperties props, byte[] body
                channel.basicPublish("",QUEUE,null,message.getBytes("utf-8"));
                System.out.println("mq消息發送成功!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}4

生產端啟動后RabbitMQ上面的消息隊列情況:

六、RPC 模式

 RPC即客戶端遠程調用服務端的方法 ,使用MQ可以實現RPC的異步調用,基于Direct交換機實現,流程如下:

1、客戶端即是生產者也是消費者,向RPC請求隊列發送RPC調用消息,同時監聽RPC響應隊列。

2、服務端監聽RPC請求隊列的消息,收到消息后執行服務端的方法,得到方法返回的結果。

3、服務端將RPC方法 的結果發送到RPC響應隊列。

4、客戶端(RPC調用方)監聽RPC響應隊列,接收到RPC調用結果。

 

至此,RabbitMQ的六種工作模式已經介紹完畢,手動代碼實現,實際體驗六種工作模式的不同。

 

。轉載請注明來源地址:黑帽SEO http://www.790079.com 專注于SEO培訓,快速排名
黑帽WiKi_黑帽百科(www.790079.com),8年黑帽SEO優化技術,黑帽seo快速排名,黑帽seo技術培訓學習,黑帽SEO快速排名程序、泛目錄、寄生蟲技術,贈送免費黑帽SEO視頻教程

黑帽SEO技術,網站快速排名,蜘蛛池加速收錄,目錄程序定制)

掃一下添加微信:



協助本站SEO優化一下,謝謝!
關鍵詞不能為空

免責聲明

資料匯總于網絡,如有侵權 聯系站長刪除 http://www.790079.com

同類推薦
久久久国产一区_国产综合久久久久_欧美亚洲丝袜_成人综合国产精品
亚洲不卡中文字幕| 国产日韩一区二区在线| 国产精品免费入口| 久久久精品美女| 俺去了亚洲欧美日韩| 久久久久久这里只有精品| 久久婷婷五月综合色国产香蕉 | 国产99视频精品免视看7| 久久中文字幕国产| 久久97精品久久久久久久不卡| 欧美激情国产高清| 亚洲欧美影院| 日韩欧美亚洲v片| 欧美日韩国产综合视频在线| 狠狠色伊人亚洲综合网站色| 国产日韩欧美夫妻视频在线观看| 国产精品一线二线三线| 久热这里只精品99re8久| 久久久久欧美| 久久香蕉国产线看观看av| 欧美激情亚洲一区| 亚洲免费视频播放| 日本一二三区视频在线| 日韩欧美一区二区三区四区五区 | 国产精品无码专区av在线播放| 国产精品丝袜高跟| 九九热精品视频国产| 亚洲av首页在线| 日韩国产一级片| 国产日韩欧美在线| 久久香蕉综合色| 国产精品日韩一区二区三区| 久久中文精品视频| 亚洲欧美国产一区二区| 日本一道本久久| 国内精品模特av私拍在线观看| 国产精品综合不卡av| 国产高潮呻吟久久久| 日韩最新av在线| 精品中文字幕在线观看| 日本精品久久久久久久| 国产欧美一区二区三区视频| 国产夫妻自拍一区| 精品乱码一区二区三区| 日韩精品另类天天更新| 国产在线观看一区二区三区| 国产高清www| 久久这里有精品视频| 日本精品福利视频| 国产久一道中文一区| www.亚洲免费视频| 亚洲精品中文字幕乱码三区不卡| 欧美亚洲成人精品| 91精品视频在线播放| 国产精品热视频| 日本久久久a级免费| 高清不卡日本v二区在线| 国产成人欧美在线观看| 亚洲一区二区中文字幕| 国内精品久久久久久中文字幕| 国产经品一区二区| 欧美情侣性视频| 日韩女优人人人人射在线视频| 国产伦精品一区二区三区免| 久久精品国产欧美激情| 日产国产精品精品a∨| 成人免费午夜电影| 国产精品久久久久久中文字| 日本一本a高清免费不卡| 国产精品一区二区免费| 国产精品视频不卡| 日韩中文字幕三区| 国产精品午夜一区二区欲梦| 国产精品第七十二页| 欧美理论一区二区| 九色一区二区| 日韩在线三区| 高清一区二区三区视频| 国产精品精品软件视频| 欧美精品一区二区三区在线看午夜| 国产精品av在线播放 | 国内精品小视频在线观看| 国产成人免费av电影| 日本在线视频www色| 97国产精品久久| 一本久道中文无码字幕av| 国产精品夜夜夜一区二区三区尤| 色综合91久久精品中文字幕| 国产免费xxx| 一本色道久久99精品综合| 99久热在线精品视频| 亚洲影影院av| 91精品国产电影| 亚洲美女搞黄| 91蜜桃网站免费观看| 亚洲精品中文综合第一页| 97国产suv精品一区二区62| 亚洲精品欧美极品| 久久久影院一区二区三区| 视频一区三区| 国产成人一区二区三区小说| 青青草精品毛片| 久久久久久久久91| 欧美亚州在线观看| 国产精品美女诱惑| 国产又黄又大又粗视频| 欧美日韩不卡合集视频| 成人一区二区在线| 欧美一区二区视频在线| 91福利视频网| 日韩在线第三页| 久久99国产精品99久久| 欧美日韩亚洲一| 九九综合九九综合| 97成人在线观看视频| 日韩福利在线| 国产精品久久久999| 国产精品亚洲二区在线观看| 午夜精品美女自拍福到在线| 久久久久久久久一区| 黄色91av| 亚洲区一区二区三区| 久操网在线观看| 国产色视频一区| 午夜在线视频免费观看| 久艹视频在线免费观看| 加勒比在线一区二区三区观看| 欧美精品做受xxx性少妇| 91精品国产91久久久久久最新| 日韩精品一区中文字幕| 美女视频久久黄| 久久免费视频在线| 国内精品视频在线播放| 亚洲v欧美v另类v综合v日韩v| 日日狠狠久久偷偷四色综合免费| 国产又粗又长又爽视频| 日本一区二区在线| 欧美成人亚洲成人| 国产激情在线观看视频| 欧美日韩一区二区三区在线视频| 精品高清视频| 国产v亚洲v天堂无码| 国产一区二区三区高清视频| 日韩在线综合网| 欧美成人精品一区二区| 久久av秘一区二区三区| 成人av在线天堂| 欧美日韩高清在线一区| 亚洲视频电影| 欧美成在线观看| 久久国产精品 国产精品| 国产精品午夜一区二区欲梦| 激情六月丁香婷婷| 日韩在线第一区| 欧美理论片在线观看| 久久国产精品免费观看| 国产精品一区视频网站| 黄色国产精品一区二区三区| 日韩**中文字幕毛片| 久久久久久成人精品| 国产精品久久亚洲7777| 久久精品国产99精品国产亚洲性色| 国产日韩av高清| 欧美亚洲视频在线看网址| 色综合久久88色综合天天提莫| 中文字幕日韩精品久久| 国产精品久久久久久久久电影网| 久久精品国产sm调教网站演员| 99三级在线| 国产视频999| 激情欧美一区二区三区中文字幕| 日本精品久久电影| 色播亚洲婷婷| 欧美一区二区三区免费视| 久久久久久12| 久久国产精彩视频| 国产精品久久久久久久av电影 | 久久这里只有精品99| 久久精品视频中文字幕| 久久久噜噜噜久久中文字免| 国产高清在线一区二区| 国产精品av网站| 97人人模人人爽视频一区二区| 国产综合av一区二区三区| 激情六月天婷婷| 女女同性女同一区二区三区按摩| 欧美午夜性视频| 欧美日韩视频免费在线观看 | 日韩中文字幕在线视频| 日韩在线视频观看正片免费网站| 久久另类ts人妖一区二区| 久久人人九九| 久久久之久亚州精品露出| 国产不卡一区二区三区在线观看| 久久精品综合一区| 色噜噜狠狠狠综合曰曰曰| 国产精品我不卡| 久热精品视频在线观看| 欧美激情伊人电影| 亚洲精品人成|