RabbitMQ的5种模式——再探RabbitMQ的模式,简单、工作,发布订阅(广播),路由、主题 页面分析

在这里插入图片描述

前言

RabbitMQ作为一款常用的消息中间件,在微服务项目中得到大量应用,其本身是微服务中的重点和难点,有不少概念我自己的也是一知半解,本系列博客尝试结合实际应用场景阐述RabbitMQ的应用,分析其为什么使用,并给出怎么用的案例。

本篇博客结合场景来阐述RabbitMQ的几种模式,描述了不同模式的应用场景,并给出相应的代码。(文末有惊喜~)

在这里插入图片描述
其他相关的rabbitmq博客文章列表如下:

RabbitMQ基础(1)——生产者消费者模型 & RabbitMQ简介 & Docker版本的安装配置 & RabbitMQ的helloworld + 分模块构建 & 解决大量注册案例

RabbitMQ基础(2)——发布订阅/fanout模式 & topic模式 & rabbitmq回调确认 & 延迟队列(死信)设计

RabbitMQ的Docker版本安装 + 延迟插件安装 & QQ邮箱和阿里云短信验证码的主题模式发送

目录

  • 前言
  • 引出
  • MQ场景:
    • 1异步处理
    • 2解耦
    • 3削峰填谷
    • 常见的MQ
  • 拆解控制台页面
    • 总览页面
    • 连接connection
    • 队列页面
  • 简单模式
    • 工具类:建立连接
    • 生产者:生产消息
    • 消费者:消费消息
    • 进行测试
      • Ack:项目中必须false
  • 工作模式
    • 一个生产者,两个消费者
    • QOS: 限流
      • Tips:队列参数怎么变?
    • 队列相关参数
      • x-max-length
      • x-overflow
      • x-max-length-bytes
  • 发布者订阅模式
    • 生产者:给fanout交换机发消息
    • 消费者1:队列q32
    • 消费者2:队列q321
  • 路由模式
    • 生产者:发给交换机,告诉路由键
    • 消费者1:根据路由键,接收3种消息
    • 消费者2:根据路由键,接收1种消息
  • 主题模式(Topic)
    • 生产者:主题交换机,带路由键
    • 消费者1:通配符的路由键
    • 消费者2:统配符
  • 总结

引出


1.MQ,消息队列的应用场景,几种MQ简单对比;
2.分析RabbitMQ的浏览器控制台页面;
3.结合场景来阐述RabbitMQ的几种模式,描述了不同模式的应用场景,并给出相应的代码;

在这里插入图片描述

MQ场景:

Message Queue 消息队列。

有比较大的负载,而且这些负载不用立刻马上给程序返回结构,可以有一些等待时间。可以用消息队列。

1异步处理

过去的项目,大部分都是同步解决问题。

在这里插入图片描述

UserinfoService{register(){//典型的同步//插入数据库//发短信//发邮件}
}
//如果你的操作基本没有任何延时操作,或者瓶颈,没有压力。那么你没有必要用MQ
UserinfoService{register(){//发起的是一个异步请求都不等待对方给我返回的结果//异步插入数据库//异步发短信//异步发邮件}
}

2解耦

在这里插入图片描述

库存有没有可能挂了。或者访问量巨大。因为库存慢,导致订单也慢。

在这里插入图片描述

解耦。

在这里插入图片描述

3削峰填谷

双12的时候,叫好早上的8点。秒杀的时候。

有些瞬间服务器压力是超大的,过了这个瞬间,几乎没有消耗量。

等服务器能够正常的时候,我慢慢执行就行了。

在这里插入图片描述

常见的MQ

MQ的前身,就是一个发布者订阅模式。

Kafka RabbitMQ(1W/s) RocketMQ ActiveMQ

Kafka : 10w/S 主要用于日志

Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群来提供实时的消息。

在这里插入图片描述

拆解控制台页面

总览页面

包括刷新时间间隔,rabbitmq节点,连接端口的信息等

在这里插入图片描述

配置可以导出和导入

在这里插入图片描述

连接connection

在这里插入图片描述

队列页面

包括队列的名字,状态,准备好的消息数量,未确认的消息数量,总计消息数量

在这里插入图片描述

简单模式

在这里插入图片描述

使用的依赖

	<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>

工具类:建立连接

在这里插入图片描述

package com.tianju.config;import com.rabbitmq.client.Connection;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 建立连接的工具类*/
public class ConnectionFactory {public static Connection createConnection() throws IOException, TimeoutException {com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();connectionFactory.setHost("192.168.111.130"); // http://192.168.111.130/connectionFactory.setPort(5672);connectionFactory.setUsername("admin");connectionFactory.setPassword("123");connectionFactory.setVirtualHost("/demo");// amqp://admin@192.168.111.130:5672/return connectionFactory.newConnection();}
}

在这里插入图片描述

生产者:生产消息

在这里插入图片描述

package com.tianju.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String QUEUE_ORDER = "queue_order";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列channel.queueDeclare(QUEUE_ORDER,false,false,false,null);// 发送消息,指定给哪个队列上发消息for (int i = 0; i < 100; i++) {String msg = "hello rabbitmq--"+i;channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());System.out.println("消息发布成功");}connection.close();}
}

消费者:消费消息

在这里插入图片描述

package com.tianju.simple;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static String QUEUE_ORDER = "queue_order";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列channel.queueDeclare(QUEUE_ORDER,false,false,false,null);// 队列必须声明,如果不存在,则自动创建// 声明一个消费者DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 回调函数,用来接收消息* @param consumerTag the <i>consumer tag</i> associated with the consumer* @param envelope packaging data for the message* @param properties content header data for the message* @param body the message body (opaque, client-specific byte array),字节数组* @throws IOException*/public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body); // 收到的信息System.out.println("消费者接收到:"+msg);try {Thread.sleep(3000);// 模拟一个耗时操作} catch (InterruptedException e) {throw new RuntimeException(e);}}};// 表明自己是消费者,接收消息/*** autoAck:自动确认设置成 true*/channel.basicConsume(QUEUE_ORDER, true, defaultConsumer);}
}

进行测试

生产者发送100条消息
在这里插入图片描述

消费者进行消费,假设突然之间服务宕机了,此时消费了4条消息,理论上还应该有96条消息

在这里插入图片描述

打开控制台页面查看,消息全部消失了,出现了数据丢失的情况

在这里插入图片描述

全部用默认值的情况下,如果发生异常,则消息全部丢失。

消费者一次性拿到了所有的消息。

Ack:项目中必须false

生产者 ==投递消息=》队列

消费者=接受==》队列

ack: false 必须要手工确认。

消费者接到这个消息的时候, 这个消息进入 unack状态。

1:手工确认。消息删除。完成

2: 没手工确认,断开连接或者超时。

3:这个消息重新进入ready状态。等待其他消费者进行消费。

在这里插入图片描述

先设置自动ack为false,表示需要手工确认,然后在消费消息的方法中,进行消息的确认。

在这里插入图片描述

package com.tianju.simple;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static String QUEUE_ORDER = "queue_order";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列channel.queueDeclare(QUEUE_ORDER,false,false,false,null);// 队列必须声明,如果不存在,则自动创建// 声明一个消费者DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 回调函数,用来接收消息* @param consumerTag the <i>consumer tag</i> associated with the consumer* @param envelope packaging data for the message* @param properties content header data for the message* @param body the message body (opaque, client-specific byte array),字节数组* @throws IOException*/public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException {// no work to doString msg = new String(body); // 收到的信息System.out.println("消费者接收到:"+msg);try {Thread.sleep(3000);// 模拟一个耗时操作} catch (InterruptedException e) {throw new RuntimeException(e);}// 消费者代码可能有失败,消息拿到之后,可能还没有处理,就宕机了// 消息确认的代码一定在最后一行// long deliveryTag 消息的下标;// boolean multiple 是否批量确认;channel.basicAck(envelope.getDeliveryTag(),true); // 确认消息,批量确认}};// 表明自己是消费者,接收消息/*** autoAck:自动确认设置成 true* 是否自动确认,* false:不进行自动确认;true:自动确认* 消费过程中可能产生异常* 如果产生异常,则必须进行消费补偿*/channel.basicConsume(QUEUE_ORDER, false, defaultConsumer); // 接收消息}
}

再次模拟宕机的情况,一开始消息全部被放到Unack中,当宕机时,又把消息吐了出来,至少消息没有出现丢失的情况。

在这里插入图片描述

工作模式

在这里插入图片描述

一个生产者,两个消费者

在这里插入图片描述

两个消费者消费的是同一个队列中的消息。

两个消费者上来后,默认是按照平均分配。结果: 有人瞬间干完。有人很久都没干完。

QOS: 限流

可以限制你一次拉取几个。

这样两个消费者,也能够节省服务器CPU了。

        // 创建频道Channel channel = connection.createChannel();channel.basicQos(1); /** 1次只能拿1个 **/

在这里插入图片描述

当消费的速度太慢,觉得不够,加入其他的消费者。

核心,只有一个消息队列。

消息是轮询的方式,发送给不同的消费者。

在这里插入图片描述

Tips:队列参数怎么变?

队列的参数发生变化后,要删除再添加。

在这里插入图片描述

        // 创建频道Channel channel = connection.createChannel();channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列boolean durable = true; // 当前队列中的消息持久化操作,重启之后,消息还在// 创建队列channel.queueDeclare(QUEUE_ORDER,durable,false,false,null);

在这里插入图片描述

队列相关参数

x-max-length

在这里插入图片描述

package com.tianju.work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String QUEUE_ORDER = "queue_order";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列boolean durable = true; // 当前队列中的消息持久化操作,重启之后,消息还在Map map = new HashMap();map.put("x-max-length", 10); // 设置最大的长度,只接受10个// 创建队列channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);// 发送消息,指定给哪个队列上发消息for (int i = 0; i < 1000; i++) {String msg = "hello rabbitmq--"+i;channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());System.out.println("消息发布成功");}connection.close();}
}

How many (ready) messages a queue can contain before it starts to drop them from its head.

在这里插入图片描述

x-overflow

Sets the queue overflow behaviour. This determines what happens to messages when the maximum length of a queue is reached. Valid values are drop-head, reject-publish or reject-publish-dlx. The quorum queue type only supports drop-head and reject-publish.

在这里插入图片描述
此时就是前10个了

在这里插入图片描述

x-max-length-bytes

在这里插入图片描述

package com.tianju.work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String QUEUE_ORDER = "queue_order";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列boolean durable = true; // 当前队列中的消息持久化操作,重启之后,消息还在Map map = new HashMap();map.put("x-max-length", 10); // 设置最大的长度,只接受10个map.put("x-overflow", "reject-publish"); // 拒绝发布,变成前10gemap.put("x-max-length-bytes", 4); // 消息的最大字节长度// 创建队列channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);// 发送消息,指定给哪个队列上发消息for (int i = 0; i < 20; i++) {String msg = "53"+i;channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());msg = "hello rabbitmq--"+i;channel.basicPublish("", QUEUE_ORDER, null, msg.getBytes());System.out.println("消息发布成功");}connection.close();}
}

在这里插入图片描述

参数

在这里插入图片描述

 		//创建队列boolean durable = true;//当前队列中的消息进行持久化操作  重启之后,消息还在Map map = new HashMap();map.put("x-max-length",10);//设置队列的最大长度map.put("x-overflow","reject-publish");//设置队列的最大长度  drop headmap.put("x-max-length-bytes",4);//设置消息的最大字节长度
//        map.put("x-expires",10000);//超时后,直接删除队列channel.queueDeclare(QUQUENAME,durable,false,false,map);

发布者订阅模式

在这里插入图片描述

工作模式中,只有一个队列。两个消费者轮询从同一个队列中取数据。

发布者订阅模式: 交换机后面会绑定多个消息队列。每个消息队列都有完整的信息。每个队列后的消费者都会有完整的消息。

交换机:就是一个无意识的广播。行为就是扇出.

在这里插入图片描述

场景:

下订单:

1:订单的数据入数据库
2:发送订单的短信
3: 物流

要每个队列都有完整的信息。

在这里插入图片描述

生产者:给fanout交换机发消息

package com.tianju.publish;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String QUEUE_ORDER = "queue_order";private static  String EXCHANGE = "pet_exchange";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建交换机channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出,类型只能用这个for (int i = 0; i < 100; i++) {channel.basicPublish(EXCHANGE, "", MessageProperties.TEXT_PLAIN, ("hello fanout"+i).getBytes());}}
}

消费者1:队列q32

package com.tianju.publish;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static  String EXCHANGE = "pet_exchange";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建交换机channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出,类型只能用这个channel.queueDeclare("q32", false, false, false, null);channel.queueBind("q32", EXCHANGE, ""); // 路由键fanout模式,必须为空,即使写了是无效的channel.basicConsume("q32", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("消费者1:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

消费者2:队列q321

package com.tianju.publish;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {private static  String EXCHANGE = "pet_exchange";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建交换机channel.exchangeDeclare(EXCHANGE, "fanout"); // 扇出,类型只能用这个channel.queueDeclare("q321", false, false, false, null);channel.queueBind("q321", EXCHANGE, ""); // 路由键fanout模式,必须为空,即使写了是无效的channel.basicConsume("q321", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("消费者2:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

路由模式

允许用不同的路由键来接受不同的信息。

在这里插入图片描述

队列会接到完整的全部的信息。

日志系统:

日志级别:

OFF INFO DEBUG ERROR FATAL ALL

如果是致命错误: 立刻给运维发短信。

全部信息: 放到专门的日志系统
Error: 进行发邮件

在这里插入图片描述

生产者:发给交换机,告诉路由键

package com.tianju.routing;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建交换机channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机String msg  = "this is fatal message";channel.basicPublish(EXCHANGE,"fatal", MessageProperties.TEXT_PLAIN,msg.getBytes());msg  = "this is error message";channel.basicPublish(EXCHANGE,"error", MessageProperties.TEXT_PLAIN,msg.getBytes());msg  = "this is debug message";channel.basicPublish(EXCHANGE,"debug", MessageProperties.TEXT_PLAIN,msg.getBytes());}
}

消费者1:根据路由键,接收3种消息

package com.tianju.routing;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.Calendar;
import java.util.concurrent.TimeoutException;public class Consumer {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列// 创建交换机channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机// 队列必须声明,如果不存在,则自动创建channel.queueDeclare("q58", false, false, false, null);channel.queueBind("q58", EXCHANGE, "fatal");channel.queueBind("q58", EXCHANGE, "error");channel.queueBind("q58", EXCHANGE, "debug");channel.basicConsume("q58", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("消费者1:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

消费者2:根据路由键,接收1种消息

package com.tianju.routing;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列// 创建交换机channel.exchangeDeclare(EXCHANGE, "direct"); // 直接交换机// 队列必须声明,如果不存在,则自动创建channel.queueDeclare("q581", false, false, false, null);channel.queueBind("q581", EXCHANGE, "fatal");channel.basicConsume("q581", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("消费者1:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

主题模式(Topic)

在这里插入图片描述

可以采取通配符模式:

* (star) can substitute for exactly one word.
# (hash) can substitute for zero or more words.

#`:匹配0个或多个词

*:匹配不多不少恰好1个词

举例:

item.#:能够匹配item.insert.abc 或者 item.insert

item.*:只能匹配item.insert

场景:

京东: 无锡,河南下订单。退货

==========总部 收到这个订单 .

无锡分销商也有这个消息 *.wuxi

在这里插入图片描述

生产者:主题交换机,带路由键

package com.tianju.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 生产者发送消息* 建立连接--> 创建频道 --> 创建队列 --> 发送消息*/
public class Provider {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建交换机channel.exchangeDeclare(EXCHANGE, "topic"); // 主题模式交换机String msg  = "this is wuxi order";channel.basicPublish(EXCHANGE,"order.wuxi", MessageProperties.TEXT_PLAIN,msg.getBytes());msg  = "this is henan order";channel.basicPublish(EXCHANGE,"order.henan", MessageProperties.TEXT_PLAIN,msg.getBytes());msg  = "this is wuxi back";channel.basicPublish(EXCHANGE,"back.wuxi", MessageProperties.TEXT_PLAIN,msg.getBytes());msg  = "this is henan back";channel.basicPublish(EXCHANGE,"back.henan", MessageProperties.TEXT_PLAIN,msg.getBytes());}
}

消费者1:通配符的路由键

package com.tianju.topic;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列// 创建交换机channel.exchangeDeclare(EXCHANGE, "topic"); // 主题交换机// 队列必须声明,如果不存在,则自动创建channel.queueDeclare("q58", false, false, false, null);channel.queueBind("q58", EXCHANGE, "*.wuxi"); // 意味着无锡的订单和退货都能收到channel.basicConsume("q58", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("无锡仓库:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

消费者2:统配符

package com.tianju.topic;import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {private static String EXCHANGE = "exchange_58";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接Connection connection = ConnectionFactory.createConnection();// 创建频道Channel channel = connection.createChannel();// 创建队列// 创建交换机channel.exchangeDeclare(EXCHANGE, "topic"); // 主题交换机// 队列必须声明,如果不存在,则自动创建channel.queueDeclare("q581", false, false, false, null);channel.queueBind("q581", EXCHANGE, "*.*"); // 意味着所有的订单和退货都能收到channel.basicConsume("q581", new DefaultConsumer(channel){public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{// no work to doString msg = new String(body);System.out.println("总部仓库:"+msg);channel.basicAck(envelope.getDeliveryTag(), false);}});}
}

总结

1.MQ,消息队列的应用场景,几种MQ简单对比;
2.分析RabbitMQ的浏览器控制台页面;
3.结合场景来阐述RabbitMQ的几种模式,描述了不同模式的应用场景,并给出相应的代码;

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/107376.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【SpringCloud-11】SCA-sentinel

sentinel是一个流量控制、熔断降级的组件&#xff0c;可以替换第一代中的hystrix。 hystrix用起来没有那么方便&#xff1a; 1、要在调用方引入hystrix&#xff0c;没有ui界面进行配置&#xff0c;需要在代码中进行配置&#xff0c;侵入了业务代码。 2、还要自己搭建监控平台…

MySQL进阶(再论JDBC)——JDBC编程思想的分析 JDBC的规范架构 JDBC相关的类分析

前言 SQL&#xff08;Structured Query Language&#xff09;是一种用于管理关系型数据库的标准化语言&#xff0c;它用于定义、操作和管理数据库中的数据。SQL是一种通用的语言&#xff0c;可以用于多种关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;如MySQ…

数据结构-----红黑树简介

目录 前言 1.什么是红黑树&#xff1f; 2.为什么需要红黑树&#xff1f;&#xff08;与AVL树对比&#xff09; 3.红黑树的特性 前言 在此之前我们学习过了二叉排序树和平衡二叉树&#xff08;AVL树&#xff09;&#xff0c;这两种树都是属于搜索树的一种&#xff0c;那么今天…

web 性能优化详解(Lighthouse工具、优化方式、强缓存和协商缓存、代码优化、算法优化)

1.性能优化包含的方面 优化性能概念宽泛&#xff0c;可以从信号、系统、计算机原理、操作系统、网络通信、DNS解析、负载均衡、页面渲染。只要结合一个实际例子讲述清楚即可。 2.什么是性能&#xff1f; Web 性能是客观的衡量标准&#xff0c;是用户对加载时间和运行时的直观…

2.3 初探Hadoop世界

文章目录 零、学习目标一、导入新课二、新课讲解&#xff08;一&#xff09;Hadoop的前世今生1、Google处理大数据三大技术2、Hadoop如何诞生3、Hadoop主要发展历程 &#xff08;二&#xff09;Hadoop的优势1、扩容能力强2、成本低3、高效率4、可靠性5、高容错性 &#xff08;三…

在C++和Python的项目中使用ROS

如果搜索如何使用ROS&#xff0c;搜索结果肯定是先建立工作空间&#xff0c;在创建功能包等等步骤&#xff0c;但其实不需要这么麻烦。 在Python中使用ROS&#xff0c;只需要在Pycharm的Project Structure中的Add Content Root加入ros的packages就可以了&#xff0c;如下图 在…

二叉树题目:从前序与中序遍历序列构造二叉树

文章目录 题目标题和出处难度题目描述要求示例数据范围 解法一思路和算法代码复杂度分析 解法二思路和算法代码复杂度分析 题目 标题和出处 标题&#xff1a;从前序与中序遍历序列构造二叉树 出处&#xff1a;105. 从前序与中序遍历序列构造二叉树 难度 5 级 题目描述 要…

<三>Qt斗地主游戏开发:主界面初始化显示

1. 主界面效果 效果关键点&#xff1a; 1&#xff09;拖动标题栏可实现主界面拖动 2&#xff09;logo图标名称及主界面背景 3&#xff09;最小化及关闭 2.思路分析 1&#xff09;背景图片及logo图标的设定比较简单&#xff0c;通过stylesheet即可实现。通过QWidget的拖动即可实…

算法leetcode|84. 柱状图中最大的矩形(rust重拳出击)

文章目录 84. 柱状图中最大的矩形&#xff1a;样例 1&#xff1a;样例 2&#xff1a;提示&#xff1a; 分析&#xff1a;题解&#xff1a;rust&#xff1a;go&#xff1a;c&#xff1a;python&#xff1a;java&#xff1a; 84. 柱状图中最大的矩形&#xff1a; 给定 n 个非负整…

【git的使用方法】——上传文件到gitlab仓库

先进入到你克隆下来的仓库的目录里面 比如&#xff1a;我的仓库名字为zhuox 然后将需要上传推送的文件拷贝到你的克隆仓库下 这里的话我需要拷贝的项目是t3 输入命令ls&#xff0c;就可以查看该文件目录下的所有文件信息 然后输入git add 文件名 我这边输入的是 &#x…

处理ElementUI组件默认样式多次重复问题

问题截图&#xff1a; 解决办法&#xff1a; 在postcss.config.js文件里添加配置项&#xff1a; module.exports {plugins: {autoprefixer: {},cssnano: {} //添加这行代码}, } 处理结果&#xff1a; github issues&#xff1a; https://github.com/ElemeFE/element/is…

【LeetCode刷题(数组and排序)】:存在重复元素

给你一个整数数组 nums 。如果任一值在数组中出现 至少两次 &#xff0c;返回 true &#xff1b;如果数组中每个元素互不相同&#xff0c;返回 false 示例 1&#xff1a; 输入&#xff1a;nums [1,2,3,1] 输出&#xff1a;true 示例 2&#xff1a; 输入&#xff1a;nums [1,2…

WorkPlus即时通讯办公软件,助力企业实现移动化办公

在移动互联网的时代背景下&#xff0c;企业对于高效的移动平台需求日益迫切。WorkPlus作为领先品牌&#xff0c;致力于为企业打造卓越的移动平台&#xff0c;助力企业实现协作与效率的突破。本文将探讨WorkPlus如何通过其专业的解决方案&#xff0c;为企业打造无限可能的移动办…

14.8 Socket 一收一发通信

通常情况下我们在编写套接字通信程序时都会实现一收一发的通信模式&#xff0c;当客户端发送数据到服务端后&#xff0c;我们希望服务端处理请求后同样返回给我们一个状态值&#xff0c;并以此判断我们的请求是否被执行成功了&#xff0c;另外增加收发同步有助于避免数据包粘包…

【Kotlin精简】第4章 函数

1 简介 函数是用来运行代码的载体&#xff0c;可以在一个函数里编写很多行代码&#xff0c;当运行这个函数时&#xff0c;函数中的所有代码会全部运行。 Kotlin中的函数同Java完全面向对象的规则不太一样&#xff0c;在Kotlin的世界里&#xff0c;函数也是准C位的&#xff0c;…

微服务架构 | 超时管理

INDEX LSA 级别与全年停机时间速查表LSA 级别实战TP 性能超时时间设计原则 LSA 级别与全年停机时间速查表 计算公式&#xff1a;60 * 60 * 24 * 365 * (1-LSA) 31,536,000‬ * (1-LSA) 系统级别LSA级别全年停机时间099.999%5分钟099.99%52分钟199.9%8.8小时299%3.65 天 LSA…

基于 ACK Fluid 的混合云优化数据访问(五):自动化跨区域中心数据分发

作者&#xff1a;车漾 前文回顾&#xff1a; 本系列将介绍如何基于 ACK Fluid 支持和优化混合云的数据访问场景&#xff0c;相关文章请参考&#xff1a; -基于 ACK Fluid 的混合云优化数据访问&#xff08;一&#xff09;&#xff1a;场景与架构 -基于 ACK Fluid 的混合云优…

科研上新 | 第2期:可驱动3D肖像生成;阅读文本密集图像的大模型;文本控制音色;基于大模型的推荐智能体

编者按&#xff1a;欢迎阅读“科研上新”栏目&#xff01;“科研上新”汇聚了微软亚洲研究院最新的创新成果与科研动态。在这里&#xff0c;你可以快速浏览研究院的亮点资讯&#xff0c;保持对前沿领域的敏锐嗅觉&#xff0c;同时也能找到先进实用的开源工具。 本期内容速览 …

基于yolov2深度学习网络的猫脸检测识别matlab仿真

目录 1.算法运行效果图预览 2.算法运行软件版本 3.部分核心程序 4.算法理论概述 5.算法完整程序工程 1.算法运行效果图预览 2.算法运行软件版本 matlab2022a 3.部分核心程序 load yolov2.mat% 加载训练好的目标检测器 img_size [224,224]; imgPath test/; % 图…

至强服务器BIOS/UEFI驱动开发笔记

至强服务器BIOS/UEFI驱动开发笔记 驱动开发基础Hello UEFI Driver 项目选择项目位置初始化驱动代码文件结构驱动程序入口和基本功能导入AMI工程AMI平台Hello UEFI Driver 编译问题测试结果打印设备列表继续开发`HelloWorldSupported`函数依赖配置使用脚本编译编译测试此DXE驱动…