商城网站怎么建,下载官方正版百度,wordpress 文章字体,湖北网络营销推广怎么做交换机概念
RabbitMQ消息传递模型的核心思想就是#xff1a;生产者生产的消息从不会直接发送到队列。实际上#xff0c;通常生产者不知道这些消息会传递到那些队列中
相反#xff0c;生产者只能将消息发送到交换机#xff0c;交换机的工作内容也很简单#xff0c;一方面…交换机概念
RabbitMQ消息传递模型的核心思想就是生产者生产的消息从不会直接发送到队列。实际上通常生产者不知道这些消息会传递到那些队列中
相反生产者只能将消息发送到交换机交换机的工作内容也很简单一方面是接受来自生产者的消息另一方面是将它们推入到队列中。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定的队列、还是把它们放到随即队列、还是直接丢弃这些都是由交换机来决定的 无名交换机
在本教程的前面部分我们对 exchange 一无所知但仍然能够将消息发送到队列。之前能实现的 原因是因为我们使用的是默认交换我们通过空字符串(“”)进行标识 第一个参数是交换机的名称。空字符串表示默认或无名称交换机消息能路由发送到队列中其实是由 routingKey(bindingkey)绑定 key 指定的如果它存在的话
临时队列
之前的文章我们使用的都是具有特定名称的队列例如hello和ack_queue队列的名称对我们来说至关重要-我们需要指定我们的消费者去消费哪个队列的消息
每当我们连接到RabbitMQ时我们都需要一个全新的空队列为此我们可以创建一个具有随机名称的队列或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接队列自动删除
创建临时队列的方式如下 创建成功的队列如下 绑定bindings
什么是bindingbinding其实是交换机和队列之间的桥梁他告诉我们交换机和那个队列进行绑定关系。比如说下面这张图告诉我们的就是 X 与 Q1 和 Q2 进行了绑定 交换机类型
交换机重要是分为以下几种类型直接direct、主题topic、标题headers、扇出fanout
Fanout
Fanout介绍
Fanout这种类型非常节俭但正如名称一样它是将接受得到的所有消息广播到它的所有队列中。系统中默认有些exchange类型 Fanout实战 logs和临时队列的绑定关系如下 ReceiveLogs01 将接收到的消息打印在控制台
/*** 消息接收* ReceiveLogs01 将接收到的消息打印在控制台*/public class ReceiveLogs01 {public static final String EXCHANGE_NAME logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();//创建交换机channel.exchangeDeclare(EXCHANGE_NAME,fanout);/*** 生成一个临时队列 队列名称是随机的* 当消费者断开和该队列的链接时 队列自动删除*/String queueName channel.queueDeclare().getQueue();//把该临时队列绑定我们的交换机 其中routingKeybinding key为空字符串 这意味着队列会接收交换机发送的所有消息channel.queueBind(queueName,EXCHANGE_NAME,);System.out.println(等待接收消息把接收到的消息打印到屏幕上......);DeliverCallback deliverCallback (consumerTag, delivery)-{String message new String(delivery.getBody(), UTF-8);System.out.println(控制台打印接收到的消息message);};channel.basicConsume(queueName,true,deliverCallback,consumerTag-{});}
}
ReceiveLogs02 将接收到的消息存储在磁盘
/*** ReceiveLogs02 将接收到的消息存储在磁盘*/
public class ReceiveLogs02 {private static final String EXCHANGE_NAME logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, fanout);String queueName channel.queueDeclare().getQueue();channel.queueBind(queueName,EXCHANGE_NAME,);System.out.println(等待接收消息,把接收到的消息写到文件.....);DeliverCallback deliverCallback (consumerTag,delivery)-{String message new String(delivery.getBody(), UTF-8);File file new File(D:rabbitmq_info.txt);FileUtils.writeStringToFile(file,message,UTF-8,true);System.out.println(数据写入文件成功message);};channel.basicConsume(queueName,true,deliverCallback, consumerTag - { });}
}EmitLog 发送消息给两个消费者接收
/*** EmitLog 发送消息给两个消费者接收*/
public class EmitLog {private static final String EXCHANGE_NAME logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();//创建交换机 1、名称 2、类型channel.exchangeDeclare(EXCHANGE_NAME,fanout);Scanner scanner new Scanner(System.in);while(scanner.hasNext()){String message scanner.nextLine();channel.basicPublish(EXCHANGE_NAME,,null,message.getBytes(StandardCharsets.UTF_8));System.out.println(生产者发出消息message);}}
}
运行结果; Direct
Direct介绍
上一节中我们的日志系统将所有消息广播给所有消费者对此我们想做一些改变例如我们希望将日志消息写入磁盘的程序仅接受严重错误errors而不存储那些警告warning或者信息info日志消息避免浪费磁盘空间。Fanout 这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的 广播在这里我们将使用 direct 这种类型来进行替换这种类型的工作方式是消息只去到它绑定的 routingKey 队列中去
单独绑定 在上面这张图中我们可以看到 X 绑定了两个队列绑定类型是 direct。队列 Q1 绑定键为 orange 队列 Q2 绑定键有两个:一个绑定键为 black另一个绑定键为 green 在这种绑定情况下生产者发布消息到 exchange 上绑定键为 orange 的消息会被发布到队列 Q1。绑定键为 blackgreen 和的消息会被发布到队列 Q2其他消息类型的消息将被丢弃
多重绑定 当交换机绑定类型是direct但是它绑定的多个队列的key相同这种情况下虽然绑定类型都是direct但是它的表现就和fanout类似
Direct实战 logs和临时队列的绑定关系如下 生产者代码如下
public class EmitLogDirect {public static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME,direct);//创建多个bindingKeyMapString,String bindingKeyMap new HashMap();bindingKeyMap.put(info,普通info信息);bindingKeyMap.put(warning,警告warning信息);bindingKeyMap.put(error,错误error信息);//debug 没有消费这接收这个消息 所有就丢失了bindingKeyMap.put(debug,调试 debug 信息);for (Map.EntryString, String bindingKeyEntry: bindingKeyMap.entrySet()) {String bindingKey bindingKeyEntry.getKey();String message bindingKeyEntry.getValue();/*** 发送一个消息* 1.发送到那个交换机* 2.路由的KEY值是哪个 本次是队列的名称* 3.其他参数信息* 4.发送消息的消息体*/channel.basicPublish(EXCHANGE_NAME, bindingKey, null,message.getBytes(UTF-8));System.out.println(生产者发出消息: message);}}
} 消费者1将日志保存到文件中
public class ReceiveLogsDirect01 {public static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME,direct);String queueName channel.queueDeclare().getQueue();channel.queueBind(queueName,EXCHANGE_NAME,error);System.out.println(等待接受消息......);DeliverCallback deliverCallback (consumeTag,delivery)-{String message new String(delivery.getBody(),UTF-8);message接收绑定键:delivery.getEnvelope().getRoutingKey(),消息:message;File file new File(D:rabbitmq_info.txt);FileUtils.writeStringToFile(file,message,UTF-8,true);System.out.println(数据写入文件成功message);};/*** 消费者信息* 1.消费哪个队列* 2.消费成功之后是否要自动应答 true自动应答 false手动应答* 3.消费者微车才能更改消费的回调* 4.消费者取消消费回调*/channel.basicConsume(queueName,true,deliverCallback, consumerTag - {});}
}
消费者2将日志打印到控制台
public class ReceiveLogsDirect02 {private static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws IOException {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME,direct);String queueName channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, info);channel.queueBind(queueName, EXCHANGE_NAME, warning);System.out.println(等待接受消息......);/*** 消费者信息* 1.消费哪个队列* 2.消费成功之后是否要自动应答 true自动应答 false手动应答* 3.消费者微车才能更改消费的回调* 4.消费者取消消费回调*/DeliverCallback deliverCallback (consumeTag,delivery)-{String message new String(delivery.getBody(), UTF-8);System.out.println( 接收绑定键 :delivery.getEnvelope().getRoutingKey(), 消息:message);};channel.basicConsume(queueName,true,deliverCallback, consumerTag - {});}
}
运行结果 Topics
Topics介绍
在上一个小节中我们改进了日志记录系统。我们没有使用只能进行随意广播的 fanout 交换机而是 使用了 direct 交换机从而有能实现有选择性地接收日志 尽管使用 direct 交换机改进了我们的系统但是它仍然存在局限性-比方说我们想接收的日志类型有 info.base 和 info.advantage某个队列只想 info.base 的消息那这个时候 direct 就办不到了。这个时候 就只能使用 topic 类型
Topics使用
发送到类型是 topic 交换机的消息的 routing_key 不能随意写必须满足一定的要求它必须是一个单 词列表以点号分隔开。这些单词可以是任意单词比如说stock.usd.nyse, nyse.vmw, quick.orange.rabbit.这种类型的。当然这个单词列表最多不能超过 255 个字节 在这个规则列表中其中有两个替换符
*(星号)可以代替一个单词 #(井号)可以替代零个或多个单词
示例如下
Q1--绑定的是 中间带 orange 带 3 个单词的字符串(*.orange.*)Q2--绑定的是 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)第一个单词是 lazy 的多个单词(lazy.#) 上图是一个队列绑定关系图我们来看看他们之间数据接收情况是怎么样的
quick.orange.rabbit 被队列 Q1Q2 接收到lazy.orange.elephant 被队列 Q1Q2 接收到quick.orange.fox 被队列 Q1 接收到lazy.brown.fox 被队列 Q2 接收到lazy.pink.rabbit 虽然满足两个绑定但只被队列 Q2 接收一次quick.brown.fox 不匹配任何绑定不会被任何队列接收到会被丢弃quick.orange.male.rabbit 是四个单词不匹配任何绑定会被丢弃lazy.orange.male.rabbit 是四个单词但匹配 Q2
当队列绑定关系是下列这种情况时需要引起注意
当一个队列绑定键是#,那么这个队列将接收所有数据就有点像 fanout 了如果队列绑定键当中没有#和*出现那么该队列绑定类型就是 direct 了
Topics实战 logs和临时队列的绑定关系如下 生产者代码如下
public class EmitLogTopic {private static final String EXCHANGE_NAME topic_logs;public static void main(String[] argv) throws Exception {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, topic);/*** Q1--绑定的是* 中间带 orange 带 3 个单词的字符串(*.orange.*)* Q2--绑定的是* 最后一个单词是 rabbit 的 3 个单词(*.*.rabbit)* 第一个单词是 lazy 的多个单词(lazy.#)**/MapString, String bindingKeyMap new HashMap();bindingKeyMap.put(quick.orange.rabbit, 被队列 Q1Q2 接收到);bindingKeyMap.put(lazy.orange.elephant, 被队列 Q1Q2 接收到);bindingKeyMap.put(quick.orange.fox, 被队列 Q1 接收到);bindingKeyMap.put(lazy.brown.fox, 被队列 Q2 接收到);bindingKeyMap.put(lazy.pink.rabbit, 虽然满足两个绑定但只被队列 Q2 接收一次);bindingKeyMap.put(quick.brown.fox, 不匹配任何绑定不会被任何队列接收到会被丢弃);bindingKeyMap.put(quick.orange.male.rabbit, 是四个单词不匹配任何绑定会被丢弃);bindingKeyMap.put(lazy.orange.male.rabbit, 是四个单词但匹配 Q2);for (Map.EntryString, String bindingKeyEntry : bindingKeyMap.entrySet()) {String bindingKey bindingKeyEntry.getKey();String message bindingKeyEntry.getValue();channel.basicPublish(EXCHANGE_NAME, bindingKey, null,message.getBytes(UTF-8));System.out.println(生产者发出消息 message);}}
} 消费者1的代码如下
public class ReceiveLogsTopic01 {private static final String EXCHANGE_NAME topic_logs;public static void main(String[] argv) throws Exception {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, topic);//声明 Q1 队列与绑定关系String queueNameQ1;channel.queueDeclare(queueName, false, false, false, null);channel.queueBind(queueName, EXCHANGE_NAME, *.orange.*);System.out.println(等待接收消息.....);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( 接收队列 :queueName 绑定键:delivery.getEnvelope().getRoutingKey(),消息:message);};channel.basicConsume(queueName, true, deliverCallback, consumerTag - {});}
}消费者2代码如下
public class ReceiveLogsTopic02 {private static final String EXCHANGE_NAME topic_logs;public static void main(String[] argv) throws Exception {Channel channel RabbitMqUtils.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, topic);//声明 Q2 队列与绑定关系String queueName Q2;channel.queueDeclare(queueName, false, false, false, null);channel.queueBind(queueName, EXCHANGE_NAME, *.*.rabbit);channel.queueBind(queueName, EXCHANGE_NAME, lazy.#);System.out.println(等待接收消息.....);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( 接收队列 : queueName 绑定键: delivery.getEnvelope().getRoutingKey() ,消息: message);};channel.basicConsume(queueName, true, deliverCallback, consumerTag - {});}
} 运行结果如下
