您好,登錄后才能下訂單哦!
小編給大家分享一下RabbitMQ有哪些工作模式,相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
在SpringBoot環境下做的代碼測試,RabbitMQ的包是用SpringBoot的starter-amqp包引入的。
一個生產者對應一個消費者!!!
1、pom文件
SpringBoot導入rabbitmq 啟動包
2、工具類
/** * 〈簡述〉
* 〈連接RabbitMQ的工具類〉 * * @create 2020/7/1 * @since 1.0.0 */public class ConnectionUtil { public static Connection getConnection() throws Exception { return getConnection(new Properties()); } private static Connection getConnection(Properties properties) throws Exception { return getConnection(properties.getHost(), properties.getPort(), properties.getvHost(), properties.getUserName(), properties.getPassWord()); } public static Connection getConnection(String host, int port, String vHost, String userName, String passWord) throws Exception { //1、定義連接工廠 ConnectionFactory factory = new ConnectionFactory(); //2、設置服務器地址 factory.setHost(host); //3、設置端口 factory.setPort(port); //4、設置虛擬主機、用戶名、密碼 factory.setVirtualHost(vHost); factory.setUsername(userName); factory.setPassword(passWord); //5、通過連接工廠獲取連接 Connection connection = factory.newConnection(); return connection; } public static class Properties implements Serializable { String host = "192.168.1.103"; int port = 5672; String vHost = "/"; String userName = "guest"; String passWord = "guest"; public Properties() { } public Properties(String host, int port, String vHost, String userName, String passWord) { this.host = host; this.port = port; this.vHost = vHost; this.userName = userName; this.passWord = passWord; } public String getHost() { return host; } public Properties setHost(String host) { this.host = host; return self(); } public int getPort() { return port; } public Properties setPort(int port) { this.port = port; return self(); } public String getvHost() { return vHost; } public Properties setvHost(String vHost) { this.vHost = vHost; return self(); } public String getUserName() { return userName; } public Properties setUserName(String userName) { this.userName = userName; return self(); } public String getPassWord() { return passWord; } public Properties setPassWord(String passWord) { this.passWord = passWord; return self(); } private Properties self(){ return this; } } }
3、生產者 Producer
/** * 〈簡述〉
* 〈簡單隊列——消息生產者〉 * * @create 2020/7/1 * @since 1.0.0 */public class Producer { private final static String QUEUE_NAME = QueueName.test_simple_queue.toString(); public static void main(String[] args) throws Exception { sendMessage(); } public static void sendMessage() throws Exception { //1、獲取連接 Connection connection = ConnectionUtil.getConnection(); //2、聲明信道 Channel channel = connection.createChannel(); //3、聲明(創建)隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //4、定義消息內容 String message = "hello rabbitmq "; //5、發布消息 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x] Sent'" + message + "'"); //6、關閉通道 channel.close(); //7、關閉連接 connection.close(); } }
4、消費者Consumer
/** * 〈簡述〉
* 〈消息消費者〉 * * @create 2020/7/1 * @since 1.0.0 */public class Customer { private final static String QUEUE_NAME = QueueName.test_simple_queue.toString(); public static void main(String[] args) throws Exception { getMessage(); } public static void getMessage() throws Exception { //1、獲取連接 Connection connection = ConnectionUtil.getConnection(); //2、聲明通道 Channel channel = connection.createChannel(); //3、聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //4、定義隊列的消費者 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msgString = new String(body, "utf-8"); System.out.println("接收的消息:" + msgString); } }; //5、監聽隊列 /* true:表示自動確認,只要消息從隊列中獲取,無論消費者獲取到消息后是否成功消費,都會認為消息已經成功消費 false:表示手動確認,消費者獲取消息后,服務器會將該消息標記為不可用狀態,等待消費者的反饋, 如果消費者一直沒有反饋,那么該消息將一直處于不可用狀態,并且服務器會認為該消費者已經掛掉,不會再給其 發送消息,直到該消費者反饋。 */ channel.basicConsume(QUEUE_NAME, true, consumer); } }
注意這里消費者有自動確認消息和手動確認消息兩種模式。
一個生產者對應多個消費者,但是只能有一個消費者獲得消息!!!
1、生產者
/** * 〈簡述〉
* 〈輪詢分發——生產者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Send { private static final String QUEUE_NAME = QueueName.test_work_queue.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); for (int i = 0; i < 50; i++) { String msg = "hello " + i; System.out.println("[mq] send:" + msg); channel.basicPublish("", QUEUE_NAME, null, msg.getBytes()); Thread.sleep(i * 20); } channel.close(); connection.close(); } }
2、消費者
這里創建兩個消費者
消費者1:每接收一條消息后休眠1秒
/** * 〈簡述〉
* 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Receive1 { private static final String QUEUE_NAME = QueueName.test_work_queue.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel、 Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //定義一個消費這 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println("[1] Receive1 msg:" + msg); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[1] done"); } } }; boolean autoAck = false; channel.basicConsume(QUEUE_NAME, autoAck, consumer); } }
消費者2:每接收一條消息后休眠2秒
/** * 〈簡述〉
* 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Receive2 { private static final String QUEUE_NAME = QueueName.test_work_queue.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //定義一個消費這 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println("[2] Receive2 msg:" + msg); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[2] done"); } } }; boolean autoAck = false; channel.basicConsume(QUEUE_NAME, autoAck, consumer); } }
3、測試結果
生產者從0-49依次發送消息
消費者1:接收到偶數
消費者2:接收到奇數
4、結論
輪詢分發就是將消息隊列中的消息,依次發送給所有消費者。一個消息只能被一個消費者獲取。
消費者關閉自動應答,開啟手動回執
/** * 〈簡述〉
* 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Receive2 { private static final String QUEUE_NAME = QueueName.test_work_queue.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicQos(1);//保證一次只分發一個消息 //定義一個消費這 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println("[2] Receive2 msg:" + msg); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[2] done"); //手動回執 channel.basicAck(envelope.getDeliveryTag(),false); } } }; boolean autoAck = false;//自動應答 channel.basicConsume(QUEUE_NAME, autoAck, consumer); } }
手動回執:消費者完成業務接口方法后可以告知消息隊列處理完成,消息隊列從隊列中取一條消息發送給消費者。
能者多勞:效率高的消費者消費消息多。
一個消費者將消息首先發送到交換器,交換器綁定到多個隊列,然后被監聽該隊列的消費者所接收并消費。
ps:X表示交換器,在RabbitMQ中,交換器主要有四種類型:direct、fanout、topic、headers,這里的交換器是 fanout。下面我們會詳細介紹這幾種交換器。
1、生產者
/** * 〈簡述〉
* 〈訂閱模式——生產者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Send { private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明交換機 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");//分發 //發送消息 String msg = "hello exchange"; System.out.println("[mq] send:" + msg); channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes()); channel.close(); connection.close(); } }
2、消費者
注意:兩個消費者綁定不同的隊列,綁定相同的交換機;
消費者1:綁定隊列名=queue_fanout_email1
/** * 〈簡述〉
* 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Receive1 { private static final String QUEUE_NAME = MqName.queue_fanout_email.toString() + "1"; private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //綁定到交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); //定義一個消費這 Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println("[1] Receive1 msg:" + msg); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[1] done"); } } }; boolean autoAck = false; channel.basicConsume(QUEUE_NAME, autoAck, consumer); } }
消費者2:綁定隊列名=queue_fanout_email2
/** * 〈簡述〉
* 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */public class Receive2 { private static final String QUEUE_NAME = MqName.queue_fanout_email.toString() + "2"; private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //綁定到交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); //定義一個消費這 Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println("[2] Receive2 msg:" + msg); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[2] done"); } } }; boolean autoAck = false; channel.basicConsume(QUEUE_NAME, autoAck, consumer); } }
3、測試結果
如上圖,兩個消費者獲得了同一條消息。即就是,一個消息從交換機同時發送給了兩個隊列中,監聽這兩個隊列的消費者消費了這個消息;
如果沒有隊列綁定交換機,則消息將丟失。因為交換機沒有存儲能力,消息只能存儲在隊列中。
生產者將消息發送到direct交換器,在綁定隊列和交換器的時候有一個路由key,生產者發送的消息會指定一個路由key,那么消息只會發送到相應key相同的隊列,接著監聽該隊列的消費者消費消息。
也就是讓消費者有選擇性的接收消息。
1、生產者
/** * 〈簡述〉
* 〈路由模式-消息發送者〉 * * @create 2020/7/8 * @since 1.0.0 */public class Send { public static final String EXCHANGE_NAME = MqName.exchange_routing.toString(); public static final String ROUTING_KEY = MqName.routing_world.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明交換機 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); String msg = "route message ->" + ROUTING_KEY; System.out.println("對 " + ROUTING_KEY + " 發送消息:" + msg); channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, msg.getBytes()); //關閉連接 channel.close(); connection.close(); } }
2、消費者
注意:兩個消費者,綁定相同的交換機,不同的隊列,不一樣的路由
消費者1:路由=routing_hello
/** * 〈簡述〉
* 〈接收消息1〉 * * @create 2020/7/8 * @since 1.0.0 */public class Receive1 { public static final String QUEUE_NAME = MqName.queue_routing_001.toString(); public static final String EXCHANGE_NAME = MqName.exchange_routing.toString(); public static final String ROUTING_KEY_hello = MqName.routing_hello.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //設置預讀取數 channel.basicQos(1); //綁定交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_hello); Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println(envelope.getRoutingKey() + " [1] Receive1 msg:" + msg); } }; channel.basicConsume(QUEUE_NAME, true, consumer); } }
消費者2:路由1=routing_world 路由2=routing_hello
/** * 〈簡述〉
* 〈接收消息2〉 * * @create 2020/7/8 * @since 1.0.0 */public class Receive2 { public static final String QUEUE_NAME = MqName.queue_routing_002.toString(); public static final String EXCHANGE_NAME = MqName.exchange_routing.toString(); public static final String ROUTING_KEY_world = MqName.routing_world.toString(); public static final String ROUTING_KEY_hello = MqName.routing_hello.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //設置預讀取數 channel.basicQos(1); //綁定交換機和路由器,可以綁定多個路由 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_world); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_hello); //定義消息消費者 Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println(envelope.getRoutingKey() + " [2] Receive1 msg:" + msg); } }; //接收消息 channel.basicConsume(QUEUE_NAME, true, consumer); } }
3、測試結果
生產者發送:routing_world
消費者1:沒有接收到
消費者2:接收到了
生產者發送:routing_hello
消費者1:接收到了
消費者2:接收到了
路由模式,是以路由規則為導向,引導消息存入符合規則的隊列中。再由隊列的消費者進行消費的。
上面的路由模式是根據路由key進行完整的匹配(完全相等才發送消息),這里的通配符模式通俗的來講就是模糊匹配。
符號“#”表示匹配一個或多個詞,符號“*”表示匹配一個詞。
1、生產者
/** * 〈簡述〉
* 〈主題模式-消息發送者〉 * * @create 2020/7/8 * @since 1.0.0 */public class Send { public static final String EXCHANGE_NAME = MqName.exchange_topic.toString(); public static final String ROUTING_KEY = MqName.routing_goods.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明交換機 channel.exchangeDeclare(EXCHANGE_NAME, "topic");// String routingKey = ROUTING_KEY + ".add"; String routingKey = ROUTING_KEY + ".publish";// String routingKey = ROUTING_KEY + ".update"; String msg = "route message ->" + routingKey; channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes()); System.out.println("對 " + routingKey + " 發送消息:" + msg); //關閉連接 channel.close(); connection.close(); } }
2、消費者
注意兩個消費者,路由的不同
消費者1:路由1=routing_goods.add 路由2=routing_goods.update
/** * 〈簡述〉
* 〈接收消息1〉 * * @create 2020/7/8 * @since 1.0.0 */public class Receive1 { public static final String QUEUE_NAME = MqName.queue_topic_001.toString(); public static final String EXCHANGE_NAME = MqName.exchange_topic.toString(); public static final String ROUTING_KEY = MqName.routing_goods.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //設置預讀取數 channel.basicQos(1); //綁定交換機 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".add"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".update"); Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println(envelope.getRoutingKey() + " [1] Receive1 msg:" + msg); } }; channel.basicConsume(QUEUE_NAME, true, consumer); } }
消費者2:路由=routing_goods.*
/** * 〈簡述〉
* 〈接收消息2〉 * * @create 2020/7/8 * @since 1.0.0 */public class Receive2 { public static final String QUEUE_NAME = MqName.queue_routing_002.toString(); public static final String EXCHANGE_NAME = MqName.exchange_topic.toString(); public static final String ROUTING_KEY = MqName.routing_goods.toString(); public static void main(String args[]) throws Exception { //獲取連接 Connection connection = ConnectionUtil.getConnection(); //獲取channel Channel channel = connection.createChannel(); //聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //設置預讀取數 channel.basicQos(1); //綁定交換機和路由器,可以綁定多個路由 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".*"); //定義消息消費者 Consumer consumer = new DefaultConsumer(channel) { /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String msg = new String(body, "utf-8"); System.out.println(envelope.getRoutingKey() + " [2] Receive1 msg:" + msg); } }; //接收消息 channel.basicConsume(QUEUE_NAME, true, consumer); } }
3、測試結果
消費者1只能接收到.add 和 .update的消息
消費者2可以接收到.add .publish 和 .update的消息
與路由模式相似,但是,主題模式是一種模糊的匹配方式。
這五種工作模式,可以歸為三類:
1、direct 如果路由鍵完全匹配的話,消息才會被投放到相應的隊列。
2、fanout 當發送一條消息到fanout交換器上時,它會把消息投放到所有附加在此交換器上的隊列。
3、topic 設置模糊的綁定方式,“*”操作符將“.”視為分隔符,匹配單個字符;“#”操作符沒有分塊的概念,它將任意“.”均視為關鍵字的匹配部分,能夠匹配多個字符。
4、header headers 交換器允許匹配 AMQP 消息的 header 而非路由鍵,除此之外,header 交換器和 direct 交換器完全一致,但是性能卻差很多,因此基本上不會用到該交換器
以上是RabbitMQ有哪些工作模式的所有內容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內容對大家有所幫助,如果還想學習更多知識,歡迎關注億速云行業資訊頻道!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。