日韩性视频-久久久蜜桃-www中文字幕-在线中文字幕av-亚洲欧美一区二区三区四区-撸久久-香蕉视频一区-久久无码精品丰满人妻-国产高潮av-激情福利社-日韩av网址大全-国产精品久久999-日本五十路在线-性欧美在线-久久99精品波多结衣一区-男女午夜免费视频-黑人极品ⅴideos精品欧美棵-人人妻人人澡人人爽精品欧美一区-日韩一区在线看-欧美a级在线免费观看

歡迎訪問(wèn) 生活随笔!

生活随笔

當(dāng)前位置: 首頁(yè) > 编程资源 > 编程问答 >内容正文

编程问答

RabbitMQ入门(三)-Publish/Subscribe(发布/订阅)

發(fā)布時(shí)間:2023/12/31 编程问答 28 豆豆
生活随笔 收集整理的這篇文章主要介紹了 RabbitMQ入门(三)-Publish/Subscribe(发布/订阅) 小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

首先可以去官方看看第三章節(jié)講的一些內(nèi)容:https://www.rabbitmq.com/tutorials/tutorial-three-java.html

?

RabbitMQ中消息傳遞模型的核心思想是生產(chǎn)者永遠(yuǎn)不會(huì)將任何消息直接發(fā)送到隊(duì)列。

實(shí)際上,生產(chǎn)者通常甚至不知道消息是否會(huì)被傳遞到任何隊(duì)列。

相反,生產(chǎn)者只能向Exchanges(交換機(jī))發(fā)送消息。交換是一件非常簡(jiǎn)單的事情。

一方面,它接收來(lái)自生產(chǎn)者的消息,另一方面將它們推送到隊(duì)列。

Exchanges(交換機(jī))必須確切知道如何處理它收到的消息。

Exchanges(交換機(jī))有幾種交換類(lèi)型可供選擇:direct,?topic,?headers?and?fanout

本文采用的fanout(廣播模式)

1.生產(chǎn)者?

package com.baidu.RabbitMQ.mq02;import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;public class OneProducer {private static final String Exchange_Name="rabbit:mq02:exchange:e01";public static void main(String[] args) {try {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//TODO:fanout-exchange無(wú)意識(shí)分發(fā)消息模型channel.exchangeDeclare(Exchange_Name, BuiltinExchangeType.FANOUT);String message = "fanoutExchange-publish的消息";channel.basicPublish(Exchange_Name, "", null, message.getBytes("UTF-8"));System.out.println("生產(chǎn)者發(fā)送消息成功---> ");channel.close();connection.close();} catch (Exception e) {e.printStackTrace();}} }

?2.消費(fèi)者one

package com.baidu.RabbitMQ.mq02;import java.io.IOException;import com.rabbitmq.client.AMQP; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope;public class MultiConsumerOne {private static final String Exchange_Name="rabbit:mq02:exchange:e01";private static final String Queue_Name_01="rabbit:mq02:queue:q01";public static void main(String[] args) {try {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//TODO:fanout-exchange無(wú)意識(shí)分發(fā)消息模型-綁定channel.exchangeDeclare(Exchange_Name, BuiltinExchangeType.FANOUT);channel.queueDeclare(Queue_Name_01, true, false, false, null);channel.queueBind(Queue_Name_01, Exchange_Name, "");Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("消費(fèi)者1接收到消息成功---> "+message);}};channel.basicConsume(Queue_Name_01, true, consumer);} catch (Exception e) {e.printStackTrace();}} }

?3.消費(fèi)者two

package com.baidu.RabbitMQ.mq02;import java.io.IOException;import com.rabbitmq.client.AMQP; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope;public class MultiConsumerTwo {private static final String Exchange_Name="rabbit:mq02:exchange:e01";private static final String Queue_Name_02="rabbit:mq02:queue:q02";public static void main(String[] args) {try {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//TODO:fanout-exchange無(wú)意識(shí)分發(fā)消息模型channel.exchangeDeclare(Exchange_Name, BuiltinExchangeType.FANOUT);channel.queueDeclare(Queue_Name_02, true, false, false, null);channel.queueBind(Queue_Name_02, Exchange_Name, "");Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("消費(fèi)者2接收到消息成功---> "+message);}};channel.basicConsume(Queue_Name_02, true, consumer);} catch (Exception e) {e.printStackTrace();}} }

?

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ入门(三)-Publish/Subscribe(发布/订阅)的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò),歡迎將生活随笔推薦給好友。